# Survey: which online APs are NOT running rxg_sdan, and what do they have in
# common? Read-only. Nothing is written, loaded, restarted or reconfigured.
#
# An AP whose rxg_sdanctl binary or rxg_sdan.ko never downloaded enters
# dataplane HARD-FAIL. It stays online and manageable, hostapd still associates
# and authorizes clients, but nothing passes traffic -- and the module is not
# re-probed at runtime, so the AP sits there looking healthy.
#
# OUTPUT SHAPE -- this is built for a 400+ AP fleet, where a per-AP listing is
# the least useful thing you can print. A run over 434 APs produced 226 failure
# rows, which is a wall of text that answers nothing. So the console report is
# bounded at roughly 40 lines no matter how large the fleet is, and is all
# cross-tabs: the question is never "which APs" but "what do the broken ones
# share". Every per-AP detail goes to a TSV instead (see OUT), which is
# greppable, sortable, and attachable to a ticket.
#
# Progress goes to stderr and the report to stdout, so `... > report.txt`
# captures a clean report while you still watch the dispatch live.
#
# The three cross-tabs are chosen because they are the three candidate causes,
# and each one is free -- firmware and client version are already on the AP
# record, so no extra round-trip:
#
#   by target/kernel     the API resolves <platform>/<kernel>/<file> as an
#                        EXACT string match on kernel version. A pair with no
#                        directory on the rXg is an unconditional 404 that no
#                        amount of retrying will fix, and a firmware bump that
#                        moves the kernel patch level breaks every AP on it at
#                        once. If broken and working APs share a pair the rXg
#                        does have, the cause is the download, not the build.
#   by firmware          isolates "this release regressed the fetch".
#   by client version    isolates "this client is too old to fetch at all",
#                        which looks identical from the AP side.
#
# CLIENT OUTCOMES -- added because a healthy sdan module does not mean
# residents have service. Each AP also counts, from its own log ring:
#
#   associations vs inactivity timeouts   hostapd drops a station that stops
#                        transmitting. NOTE this is mostly NOT service denial:
#                        a client that roams away rarely sends a disassociation
#                        frame, so the AP it left holds the association until
#                        the 300s timer fires. The ratio therefore tracks
#                        ROAMING more than failure -- read it beside the
#                        roaming block, not as an outage count. It is still
#                        comparable between APs without normalising for ring
#                        span, since both counts come from the same ring.
#   orphaned VLAN ifaces sdan drops any VLAN interface with no tunnel, so a
#                        wlanX-vNNNN with no matching tunnel is a client being
#                        discarded right now.
#   VLAN ifaces with <100 packets received  associated and passing nothing.
#   roaming, both kinds  INTRA-AP band steering is the same MAC associating
#                        on both radios of one AP. INTER-AP cannot be seen
#                        from any single AP -- the client stops appearing in
#                        one log and starts in another -- so each AP reports
#                        the MACs it saw (capped at 60) and the rXg correlates
#                        them across the fleet. Band-steer rate is measured
#                        against each AP's own client count, never against
#                        fleet-wide distinct clients, which would double-count
#                        the numerator and exceed 100%.
#   NOT measured: whether a VLAN's path actually works. A counter-based
#                        test (rx climbing while tx stays flat) was built and
#                        removed. On per-unit VLANs there is normally one
#                        client and nothing else on the VLAN to talk back, so
#                        tx==0 is the resting state of a quiet VLAN, while rx
#                        keeps ticking from unsolicited mDNS/SSDP/ARP that
#                        nothing must answer. Over eight fleet runs every
#                        flagged AP/VLAN pair occurred exactly once and none
#                        recurred -- noise. Proving a path needs an ACTIVE
#                        test: rxg_sdanctl create-diag <vlan> --dhcp, which
#                        requires kmod-veth in the image.
#
# This exists because a snapshot of connected clients cannot answer it: a
# client that gets nothing is evicted within minutes and has left the rXg's
# tables before anyone looks. Only the APs remember.
#
# Verdicts:
#   NO BUILD ON RXG      AP asks for a platform/kernel this rXg cannot serve.
#   FETCH FAILED         rXg has the build, AP has no files. Retryable.
#   CORRUPT / PARTIAL    files present but wrong size against the rXg's copy,
#                        or a missing .sha256 sidecar meaning the download
#                        never completed. insmod fails on these.
#   PRESENT NOT LOADED   files look right, module still is not in /proc/modules.
#
# Sizes are compared rather than hashed: hashing 600 KB on 400 APs is not worth
# the airtime, and a truncated or empty download -- the realistic corruption
# mode for an interrupted HTTP fetch -- changes the size.
#
# USAGE -- on the rXg, as root:
#
#   curl -o /tmp/sdan-fleet-check.rb \
#     https://pre-alpha.henryhaller.com/sdan-fleet-check.rb
#   cd /space/rxg/console && bin/rails runner /tmp/sdan-fleet-check.rb
#
# Settings go EITHER as trailing arguments (simplest -- the rXg root shell is
# csh, which has no `VAR=value command` form):
#
#   bin/rails runner /tmp/sdan-fleet-check.rb LIMIT=5
#
# or via env, which in csh needs an explicit `env`:
#
#   env LIMIT=5 bin/rails runner /tmp/sdan-fleet-check.rb
#
# Do NOT put them between `runner` and the filename -- Rails will evaluate the
# first one as Ruby and never run this script, exiting 0 with no output.
#
#   LIMIT=5       cap the AP count -- do this first to confirm replies come back
#   ZONE=37       restrict to one access point zone
#   TIMEOUT=180   seconds to wait for replies (default 180)
#   EXAMPLES=3    example APs named per verdict group (default 3, 0 to suppress)
#   OUT=path      per-AP TSV (default /tmp/sdan-fleet-check-<timestamp>.tsv);
#                 per-station detail goes beside it as <OUT>-stations.tsv
#
#   RADIUS=1          also query RadiusServerLog (off by default). The AP log
#                     ring is short and rotates; RADIUS is server-side and kept
#                     a month, so it sees sessions the APs have forgotten.
#                     Reports Access-Accept/Reject counts, what VLAN the Accept
#                     assigned, and -- from accounting Stop -- how many sessions
#                     ended having moved essentially no data.
#
#                     The headline statistic is "devices NEVER authenticated":
#                     MACs rejected in the window that were never accepted in
#                     it. A device rejected once and then admitted is a typo; a
#                     device that never got in is a resident locked out. Raw
#                     reject counts cannot tell those apart -- twenty devices
#                     retrying looks identical to four hundred failing.
#   RADIUS_HOURS=2    window (default 2). Widen only after seeing the row count.
#   RADIUS_MAX=200000 refuse to aggregate beyond this many rows in the window.
#   RADIUS_SAMPLE=2000 Access-Accepts read for VLAN extraction.
#
#   PROBE_SECS=60     also capture probe requests for this many seconds (off
#                     by default). Each AP records how loudly it hears every
#                     device that probes on its channel, which is how one
#                     client's signal at its own unit's AP and at the
#                     neighbours' becomes comparable. The capture runs detached
#                     in the background and a second short command collects it
#                     -- rg_wss_client runs each rXg shell command synchronously
#                     on its event loop, so a capture inside the probe itself
#                     would freeze every AP's client for the whole window.
#
# CELL OVERLAP -- whether clients sit on their own unit's AP. A client's unit
# comes from its own MAC's current VlanTagAssignment (PMS room, account rooms
# or PMS guest rooms), else its registered Device's account; an AP's from its
# PMS room. Where both sides link to PMS rooms the room IDs are compared. Unit
# numbers parsed from names (RM227, 441b) are only a fallback, because naming
# is not consistent between properties: the report shows how often AP names
# agree with their own PMS rooms, how many matches came from each method, and
# whether each unit's busiest AP is its own (a self-check on the mapping).
# Unmapped clients are never guessed. Independently of any mapping, "unit VLAN
# spread" shows how each unit VLAN's devices spread across APs, and with
# PROBE_SECS the probe section measures overlap per probing device.
#
#   UNIT_LAYOUT=floor   split "another unit's AP" into above/below, same floor
#                       and diagonal, assuming units are numbered floor + two
#                       digits (227 = floor 2). Off by default.
#   UNIT_FROM_NAME=0    never take a unit from an AP's name.
#   UNIT_FROM_LOGIN=1   also take a unit from a numeric account login.
#   VLAN_UNIT_MAX=25    VLANs with more devices are shared, not a unit's.
# DATAPLANE READINESS -- from each AP's log ring: every association, whether
# its unit VLAN already had a tunnel on that AP (warm) or not (cold: sdan logs
# "unconfigured" until rg_wss_client builds it), how long a cold one waited
# for forwarding, whether that wait overlapped an rXg shell command (the
# client is blocked while one runs), and whether the station then left and
# joined another AP. The question it answers: do devices give up on an AP
# whose dataplane was not ready, and end up on someone else's?
#
#   BLOCK_UNITS=140-149,240-249   a power-test block, as unit-number ranges.
#                       Splits the unit VLAN spread into units whose busiest
#                       AP is in the block and the rest of the building, which
#                       is the control. Run before and after the change.
#
# Raise TIMEOUT on a large fleet. The wait loop only exits early once EVERY AP
# has replied, so one AP marked online that is actually unreachable holds it
# open for the full window; too short a timeout turns those into false
# NO_REPLY rows rather than finishing sooner.

# Settings may be passed as trailing arguments as well as environment
# variables. The rXg root shell is csh, which does not support the
# `VAR=value command` prefix form at all -- and putting the assignments after
# `rails runner` makes Rails evaluate the first one as Ruby (`REMEDIATE=1` is a
# valid assignment) and never run this file, which fails silently with exit 0.
# Accepting them here means the obvious-looking invocation works:
#
#   bin/rails runner /tmp/sdan-fleet-check.rb REMEDIATE=1 REMEDIATE_MAX=5
#
# A real environment variable still wins over an argument.
ARGV.each do |arg|
  key, value = arg.split('=', 2)
  next if value.nil? || !key.match?(/\A[A-Z][A-Z0-9_]*\z/)
  ENV[key] ||= value
end

CTL      = '/usr/sbin/rxg_sdanctl'
CACHE    = '/root/rxg_sdan'
KO       = "#{CACHE}/rxg_sdan.ko"
KO_HASH  = "#{CACHE}/rxg_sdan.ko.sha256"
CTL_HASH = "#{CACHE}/rxg_sdanctl.sha256"
SDAN_DIR = Rails.root.join('app', 'lib', 'openwifi', 'ap', 'rxg_sdan')

TIMEOUT  = (ENV['TIMEOUT']  || 180).to_i
EXAMPLES = (ENV['EXAMPLES'] || 3).to_i
OUT      = ENV['OUT'].presence ||
           "/tmp/sdan-fleet-check-#{Time.current.strftime('%Y%m%d-%H%M%S')}.tsv"
OUT_STA  = OUT.sub(/\.tsv\z/, '') + '-stations.tsv'

PROBE_SECS = (ENV['PROBE_SECS'] || 0).to_i
PRB_DIR    = '/tmp/sdan_prb'
PRB_CAP    = 200   # strongest probe sightings returned per AP
PRB_FLOOR  = -92   # weaker sightings are dropped on the AP
STQ_CAP    = 80    # stations reported per AP
DPE_CAP    = 250   # dataplane log events reported per AP (the most recent)
COLD_SLOW_S    = 5     # cold associations waiting longer than this get a cause
OLD_DPE_CAP    = 150   # events the probe sent before A events and the old-method line existed
CLIENT_START_S = 600   # associations this soon after rg_wss_client started are shown apart

# Probe-request capture, detached so the probe returns at once. ubus delivers
# hostapd's "probe" notification (address, signal, freq) to every subscriber;
# rg_wss_client already subscribes, so hostapd's behaviour does not change.
# Aggregated on the AP per (mac, freq): max and mean signal, count, printed
# strongest first so the cap keeps the closest devices. BusyBox sort does not
# order negative numbers, hence the descending sweep in END instead.
#
# The same stream carries "auth" and "assoc" events. hostapd checks the
# association RSSI threshold before it publishes "assoc", so an "auth" that
# no "assoc" from the same station follows is a join that went no further --
# a threshold refusal when its signal is below ASSOC_T. Per (mac, freq):
#   A:mac,freq,auths,followed,refused_below,unanswered_at_or_above,open,max_unanswered_signal
# printed first, so the per-AP cap never cuts them.
ASSOC_T = (ENV['ASSOC_T'] || -75).to_i
PRB_AGG = "function key(){ if(!match($0, /\"address\":\"[0-9a-f:]+\"/)) return \"\"; m=substr($0, RSTART+11, 17); " \
    "q=0; if(match($0, /\"freq\":[0-9]+/)) q=substr($0, RSTART+7, RLENGTH-7)+0; return m \",\" q } " \
  "function sig(){ return match($0, /\"signal\":-?[0-9]+/) ? substr($0, RSTART+9, RLENGTH-9)+0 : -999 } " \
  "function unanswered(k){ v=PD[k]; if(v < #{ASSOC_T}) NL[k]++; else NH[k]++; " \
    "if(!(k in NX) || v > NX[k]) NX[k]=v; delete PD[k] } " \
  "/^\\{ \"auth\"/{ k=key(); if(k == \"\") next; AU[k]++; if(k in PD) unanswered(k); PD[k]=sig(); next } " \
  "/^\\{ \"assoc\"/{ k=key(); if(k != \"\" && (k in PD)){ OK[k]++; delete PD[k] } next } " \
  "/^\\{ \"probe\"/{ " \
  "if(!match($0, /\"address\":\"[0-9a-f:]+\"/)) next; m=substr($0, RSTART+11, 17); " \
  "if(!match($0, /\"signal\":-?[0-9]+/)) next; s=substr($0, RSTART+9, RLENGTH-9)+0; " \
  "if(!match($0, /\"freq\":[0-9]+/)) next; q=substr($0, RSTART+7, RLENGTH-7)+0; " \
  "if(s < #{PRB_FLOOR}) next; k=m \",\" q; C[k]++; T[k]+=s; if(!(k in X) || s > X[k]) X[k]=s } " \
  "END{ for(k in PD) UN[k]++; " \
    "for(k in AU) printf \"A:%s,%d,%d,%d,%d,%d,%d\\n\", k, AU[k], OK[k]+0, NL[k]+0, NH[k]+0, UN[k]+0, " \
      "((k in NX) ? NX[k] : 0); " \
    "for(t=0; t>=#{PRB_FLOOR}; t--) for(k in C) if(X[k]==t) " \
    "printf \"%s,%d,%d,%d\\n\", k, X[k], int(T[k]/C[k]), C[k] }"
PRB_LAUNCH =
  if PROBE_SECS.positive?
    # Braced so the statement ends in "}" -- joining with "; " would otherwise
    # produce "&;", which sh rejects and which would sink the whole probe.
    "{ ( rm -rf #{PRB_DIR}; mkdir -p #{PRB_DIR}; mkfifo #{PRB_DIR}/fifo; " \
      "o=$(ubus list 2>/dev/null | grep '^hostapd\\.'); " \
      "awk '#{PRB_AGG}' < #{PRB_DIR}/fifo > #{PRB_DIR}/out.tmp & a=$!; " \
      "ubus subscribe $o > #{PRB_DIR}/fifo 2>/dev/null & u=$!; " \
      "sleep #{PROBE_SECS}; kill $u 2>/dev/null; wait $a; " \
      "mv #{PRB_DIR}/out.tmp #{PRB_DIR}/out; rm -f #{PRB_DIR}/fifo " \
    ") </dev/null >/dev/null 2>&1 & }"
  else
    ':'
  end
PRB_COLLECT =
  "if [ -f #{PRB_DIR}/out ]; then " \
    "echo \"SDANPRB n=$(wc -l < #{PRB_DIR}/out) " \
    "prb=$(head -#{PRB_CAP} #{PRB_DIR}/out | tr '\\n' ';')\"; rm -rf #{PRB_DIR}; " \
  "elif [ -d #{PRB_DIR} ]; then echo 'SDANPRB n=-1 prb=pending'; " \
  "else echo 'SDANPRB n=-2 prb=none'; fi"

# One line, semicolon-separated, POSIX sh -- the AP shell is BusyBox ash.
# echo always succeeds, so the CommandExecution exit code is 0 even when the
# probe inside it failed; the verdict has to come from this line's fields. The
# CE exit code is only good for detecting that a reply arrived at all.
PROBE = [
  PRB_LAUNCH,
  "rc=$(#{CTL} -j status >/dev/null 2>&1; echo $?)",
  "c=$([ -x #{CTL} ] && echo 1 || echo 0)",
  "k=$([ -f #{KO} ] && echo 1 || echo 0)",
  "kh=$([ -f #{KO_HASH} ] && echo 1 || echo 0)",
  "ch=$([ -f #{CTL_HASH} ] && echo 1 || echo 0)",
  "cs=$([ -f #{CTL} ] && wc -c < #{CTL} || echo 0)",
  "ks=$([ -f #{KO} ] && wc -c < #{KO} || echo 0)",
  "t=$(sed -n 's/^DISTRIB_TARGET=//p' /etc/openwrt_release | tr -d \"'\\\"\")",
  "kv=$(uname -r)",
  # NB: grep -c exits 1 when the count is zero but still prints "0". Any
  # `grep -c ... || echo 0` here would print 0 twice and embed a newline in
  # the value, splitting this probe's output across two lines.
  "lm=$(grep -c '^rxg_sdan ' /proc/modules 2>/dev/null)", 'lm=${lm:-0}',
  # Memory-pressure evidence. The sdan fetch and insmod happen ONCE at client
  # startup with no runtime re-probe, so an OOM kill during that window leaves
  # the AP dataplane-dead indefinitely while still looking healthy.
  "ma=$(awk '/^MemAvailable:/{print $2}' /proc/meminfo)",
  "mt=$(awk '/^MemTotal:/{print $2}' /proc/meminfo)",
  "up=$(awk '{print int($1)}' /proc/uptime)",
  # ufp is the known leaker (recursive ucv_get refcount leak in __reader_get_value).
  # ufp detection is deliberately not `pidof`. The package is `ufp`, the daemon
  # procd starts is /usr/sbin/ufpd, and that file is a script -- so pidof would
  # match the interpreter's name, not ufpd, and report nothing even when it is
  # running. Scan cmdlines instead, anchored at the start so this probe's own
  # shell (whose cmdline contains the string "ufpd") cannot match itself.
  # ufpbin separates "not installed" from "installed but not running", so an
  # RSS of 0 is never ambiguous.
  "ub=$([ -x /usr/sbin/ufpd ] && echo 1 || echo 0)",
  'fpid=""', 'cp=""',
  # Skip this probe's own process tree first. Its cmdline necessarily contains
  # both match strings (they are written right here), so without this the scan
  # finds itself and reports a client age of ~0. "SDANCHK" appears only in this
  # probe, which makes it a reliable self-marker.
  'for d in /proc/[0-9]*; do ' \
    'cl=$(tr "\\000" " " < $d/cmdline 2>/dev/null); ' \
    'case "$cl" in ' \
      '*SDANCHK*) continue ;; ' \
      '/usr/sbin/ufpd*|ufpd*) fpid=${d#/proc/} ;; ' \
      '*rg_wss_client*) cp=${d#/proc/} ;; ' \
    'esac; done',
  "fp=0", 'if [ -n "$fpid" ] && [ -r /proc/$fpid/status ]; then ' \
    'fp=$(awk \'/^VmRSS:/{print $2}\' /proc/$fpid/status); fi',
  'fp=${fp:-0}',
  # Age of the rg_wss_client process. AP uptime cannot answer whether this AP
  # re-fetched, because a miniloader restart does not reboot the AP -- only the
  # client restarts, and the sdan fetch runs once per client start. If the
  # failing APs' clients are all about as old as the mass restart while the
  # healthy ones are far older, the healthy ones simply never re-fetched.
  'ca=0',
  'if [ -n "$cp" ] && [ -r /proc/$cp/stat ]; then ' \
    'st=$(awk \'{print $22}\' /proc/$cp/stat); ' \
    'u=$(awk \'{print int($1)}\' /proc/uptime); ' \
    'ca=$((u - st / 100)); fi',
  'ca=${ca:-0}',
  # RSS of our own client. An OOM panic on this fleet named python3 as the
  # allocating task at ~21 MB while nothing else on the box reached 2 MB, so
  # the client's footprint is now evidence rather than trivia. Paired with
  # clientage above this is a leak test: if RSS climbs with process age it is
  # ours, and if it is flat across a one-hour-old client and a week-old one
  # the memory is going somewhere that does not appear in a task list.
  'cr=0',
  'if [ -n "$cp" ] && [ -r /proc/$cp/status ]; then ' \
    'cr=$(awk \'/^VmRSS:/{print $2}\' /proc/$cp/status); fi',
  'cr=${cr:-0}',
  # Crash records. The existing OOM check greps logread, which the panic's own
  # reboot wipes -- it returns 0 by construction and has done on every run.
  # pstore and OpenWrt's crashlog survive the reboot, so they are the only
  # honest detector. The location differs by platform, so probe the candidates
  # and report which one answered rather than assuming.
  # csrc distinguishes the two kinds of record, which mean different things.
  # /tmp/crashlog is tmpfs -- the boot script copies it out of the reserved
  # crash buffer at startup, so its presence says this AP's CURRENT BOOT
  # followed a crash, and AP uptime is therefore the time since that crash.
  # pstore survives across boots, so it is a history with no bound on age.
  # Reporting an mtime was the earlier attempt and it produced nothing:
  # busybox stat here has no -c, so every age came back 0. Uptime is already
  # in this probe, needs no new binary, and answers the same question for the
  # records that matter.
  'clg=0', 'cpan=0', 'coom=0', 'cpath=-', 'csrc=-',
  'for cnd in /sys/fs/pstore/* /sys/kernel/debug/crashlog ' \
      '/overlay/crashlog /tmp/crashlog; do ' \
    '[ -f "$cnd" ] || continue; ' \
    'clg=$((clg + 1)); ' \
    'if [ "$cpath" = "-" ]; then cpath=$cnd; fi; ' \
    'case "$cnd" in ' \
      '/tmp/crashlog|/sys/kernel/debug/crashlog|/overlay/crashlog) ' \
        'csrc=boot; cpath=$cnd ;; ' \
      '*) if [ "$csrc" = "-" ]; then csrc=pstore; fi ;; ' \
    'esac; ' \
    'if grep -qiE "kernel panic" "$cnd" 2>/dev/null; then cpan=$((cpan + 1)); fi; ' \
    'if grep -qiE "out of memory|oom-kill" "$cnd" 2>/dev/null; then ' \
      'coom=$((coom + 1)); fi; ' \
  'done',
  # Leftover .tmp = a download killed mid-stream. Every clean failure path
  # unlinks it; SIGKILL cannot, so this is an OOM/power-loss fingerprint.
  "tm=$(ls #{CACHE}/*.tmp 2>/dev/null | wc -l | tr -d ' ')",
  "ov=$(df -k /overlay 2>/dev/null | awk 'NR==2{print $4}')",
  # Both rings are small and wrap, so a 0 here does NOT rule out an earlier
  # OOM -- low headroom and a leftover .tmp are the stronger signals.
  "od=$(dmesg 2>/dev/null | grep -c -i 'out of memory\\|oom-kill')", 'od=${od:-0}',
  "ol=$(logread 2>/dev/null | grep -c -i 'out of memory\\|oom-kill')", 'ol=${ol:-0}',
  'oom=$((od + ol))',
  # Client outcomes. hostapd drops a station that stops transmitting, which
  # happens for TWO reasons and they are easy to conflate: the client got no
  # service and gave up, or -- far more commonly -- it roamed away without
  # sending a disassociation frame, leaving the old AP to time it out at 300s.
  # So this ratio tracks roam-away churn more than failure, and because both
  # counts come from the
  # same ring, the ratio needs no normalising for a buffer span that differs
  # per AP. One streamed pass; the buffer is too large to hold in a variable
  # on an AP with 16 MB free.
  # Evictions are counted ONLY for stations whose association is also in this
  # buffer. Counting them independently let the ratio exceed 100%: the ring is
  # short -- the 90s subscription churn writes ~20 lines a cycle -- so a client
  # that joined an hour ago has had its AP-STA-CONNECTED rotated away while its
  # eviction is still present. Pairing on MAC bounds the ratio and makes it a
  # real per-session measure. The MAC pattern is spelled out rather than using
  # {n} intervals, which busybox awk does not reliably support.
  "M=^[0-9a-f][0-9a-f]:[0-9a-f][0-9a-f]:[0-9a-f][0-9a-f]:" \
    "[0-9a-f][0-9a-f]:[0-9a-f][0-9a-f]:[0-9a-f][0-9a-f]$",
  # Roaming, both kinds. Intra-AP: the token before AP-STA-CONNECTED is the
  # BSS (wlan0:, wlan1-2:), and trimming at the first - or : leaves the radio,
  # so a MAC recorded against two radios has been band-steered. Inter-AP
  # cannot be seen from one AP at all -- the client simply stops appearing
  # here and starts appearing somewhere else -- so each AP also reports the
  # MACs it saw and the rXg correlates them across the fleet.
  # Ring span. Every count above is "per log ring", and rings rotate at very
  # different rates -- an AP churning subscriptions every 90s holds far less
  # history than a quiet one. logread stamps each line "Mon Sep 14 14:38:02
  # 2026", so field 4 is the clock; first-to-last gives the window the counts
  # actually cover, which turns them from totals into rates. A ring spanning
  # more than 24h would wrap and under-report, but at this churn rate they
  # cover minutes to hours.
  # The cross-VLAN teardown chain, counted in this same pass.
  # release_tunnel_for is keyed on manager name ('wireless'/'wired'), not per
  # client, so ONE call deletes a VLAN's tunnel however many clients are on it
  # -- and the add-port rollback calls it without consulting the per-client
  # refcount. A BSS already registered on another VLAN makes the kernel return
  # EEXIST (rxg_genl_add_port refuses duplicate devs), which is exactly what
  # trips that rollback. Net effect: one client associating tears the tunnel
  # off a DIFFERENT VLAN that other clients are actively using.
  # Three lines, keyed on VLAN id so the chain is proven rather than inferred:
  #   1  add-port failed for wireless BSS <bss> vlan N      (rg_wss_client)
  #   2  rxg_sdan: tunnel deleted vlan=N                    (kernel)
  #   3  rxg_sdan: <if> (vlan N) unconfigured, ...          (kernel)
  # Steps 1 and 3 sharing a VLAN is the finding; step 2 corroborates it.
  #
  # Was a teardown premature, or correct? Not every deletion is damage. These
  # are per-unit VLANs and residents roam, so an AP legitimately builds a
  # tunnel when someone from another unit drifts into range and tears it down
  # once they leave -- one AP was seen carrying nine different unit VLANs in an
  # evening. The discriminator is what happens NEXT: a VLAN deleted and then
  # re-added within a minute had a client that was still around, which is the
  # bug; deleted and never re-added is the roamer genuinely leaving, which is
  # correct. Both the kernel 'tunnel deleted' and 'tunnel added' lines carry
  # the VLAN, so the gap is measurable in this same pass. The userland log
  # repeats 'tunnel added', but the first match clears the pending entry so the
  # duplicate cannot double-count.
  #
  # Driver faults, counted here because they are upstream of everything above:
  # an ath11k peer that cannot be found leaves a station associated and
  # authorized with no usable radio path, and it is what drives the
  # association churn that empties VLANs in the first place.
  #
  # Association thrash and what it does to the tunnel refcount.
  # _vlan_clients only counts clients that reached sta-authorized (post
  # 4-way-handshake). A station that has associated and been handed its RADIUS
  # VLAN but has not finished the handshake is invisible to it, so during a
  # band-steer ping-pong the count passes through zero while stations are
  # genuinely present -- and if nobody completes inside the 10s linger, the
  # tunnel is deleted out from under them. The scheduled/cancelled/fired split
  # measures how often the linger actually saves it; `fired` is the count that
  # cost somebody their dataplane.
  # The ath11k line is the upstream cause rather than a symptom: a peer that
  # cannot be deleted keeps the station bouncing between radios, and every
  # bounce is another chance for the refcount to hit a false zero.
  "lo=$(logread 2>/dev/null | awk -v m=\"$M\" '" \
    "{ if($4 ~ /^[0-9][0-9]:[0-9][0-9]:[0-9][0-9]$/){ split($4,tt,\":\"); " \
        "sc=tt[1]*3600+tt[2]*60+tt[3]; if(!fs){ fa=sc; fs=1 } la=sc } } " \
    "/AP-STA-CONNECTED/{ifc=\"\"; " \
      "for(i=1;i<=NF;i++) if($i==\"AP-STA-CONNECTED\") ifc=$(i-1); " \
      "sub(/[-:].*/, \"\", ifc); " \
      "for(i=1;i<=NF;i++) if($i ~ m){seen[$i]=1; c++; n[$i]++; at[c]=sc; " \
        "if(index(r[$i], \" \" ifc \" \")==0) r[$i]=r[$i] \" \" ifc \" \"; break}} " \
    "/disassociated due to inactivity/{for(i=1;i<=NF;i++) " \
      "if($i ~ m && ($i in seen)){e++; ev[$i]=1; break}} " \
    "/Timeout in receiving peer delete response/{pd++} " \
    "/failed to find the peer/{pf++} " \
    "/failed to setup rx tid/{pt++} " \
    "/did not acknowledge authentication response/{na++} " \
    "/Prune association for/{pr++} " \
    "/last wireless client gone/ && !/kept resident/{rg++} " \
    "/last wireless client moved away/ && !/kept resident/{rm++} " \
    "/cancelled pending sdan tunnel release/{rc++} " \
    "/no wireless client returned within/{rf++} " \
    "/add-port failed for wireless BSS/{af++; " \
      "for(i=1;i<=NF;i++) if($i==\"vlan\"){ w=$(i+1); gsub(/[^0-9]/,\"\",w); " \
        "if(w != \"\") AF[w]=1; break }} " \
    "/tunnel deleted vlan=/{td++; " \
      "for(i=1;i<=NF;i++) if($i ~ /^vlan=/){ w=substr($i,6); " \
        "gsub(/[^0-9]/,\"\",w); if(w != \"\"){ TD[w]=1; DT[w]=sc } break }} " \
    "/tunnel added vlan=/{ " \
      "for(i=1;i<=NF;i++) if($i ~ /^vlan=/){ w=substr($i,6); " \
        "gsub(/[^0-9]/,\"\",w); " \
        "if(w != \"\" && (w in DT)){ gp=sc-DT[w]; if(gp<0) gp+=86400; " \
          "if(gp<=60) pm++; else lg++; delete DT[w] } break }} " \
    "/unconfigured, traffic will be dropped/{u++; " \
      "for(i=1;i<=NF;i++) if($i==\"(vlan\"){ w=$(i+1); gsub(/[^0-9]/,\"\",w); " \
        "if(w != \"\") UD[w]=1; break }} " \
    "END{ra=0; br=0; tot=0; ml=\"\"; k=0; " \
      "for(x in n){ tot++; if(n[x]>1) ra++; t=r[x]; if(gsub(/wlan/, \"\", t)>1) br++; " \
        "if(k<60){ ml=ml \",\" x; k++ } } " \
      "el=\"\"; j=0; for(x in ev){ if(j<60){ el=el \",\" x; j++ } } " \
      "sp=la-fa; if(sp<0) sp+=86400; lam=la; if(lam<fa) lam+=86400; " \
      "ce=0; for(i=1;i<=c;i++){ t=at[i]; if(t<fa) t+=86400; " \
        "if(lam-t>=300) ce++ } " \
      "for(x in DT) lg++; " \
      "ch=0; cl=\"\"; for(x in AF){ if(x in UD){ ch++; " \
        "if(length(cl) < 60) cl = cl \",\" x } } " \
      "print c+0, e+0, u+0, ra+0, br+0, tot+0, sp+0, ce+0, " \
        "(ml==\"\" ? \"-\" : substr(ml,2)), (el==\"\" ? \"-\" : substr(el,2)), " \
        "af+0, td+0, ch+0, (cl==\"\" ? \"-\" : substr(cl,2)), " \
        "pd+0, na+0, pr+0, rg+0, rm+0, rc+0, rf+0, " \
        "pf+0, pt+0, pm+0, lg+0 }')",
  'set -- $lo', 'conn=${1:-0}', 'inact=${2:-0}', 'undrop=${3:-0}',
  'reassoc=${4:-0}', 'bandroam=${5:-0}', 'nmac=${6:-0}', 'lspan=${7:-0}',
  'celig=${8:-0}', 'macs=${9:--}', 'emacs=${10:--}',
  'apfail=${11:-0}', 'tundel=${12:-0}', 'chain=${13:-0}', 'chainvl=${14:--}',
  'pdto=${15:-0}', 'noack=${16:-0}', 'prune=${17:-0}',
  'relgone=${18:-0}', 'relmove=${19:-0}', 'relcan=${20:-0}', 'relfire=${21:-0}',
  'peerfind=${22:-0}', 'rxtid=${23:-0}', 'tdprem=${24:-0}', 'tdok=${25:-0}',
  # Connected clients right now, read from the driver rather than the log.
  # It also answers a question the log cannot: whether an associated client is
  # moving any bytes. A station settled past 5 minutes, at a signal its AP can
  # serve, that has passed under 10 KB has no RF explanation -- that is the
  # dataplane, and the metric is deliberately agnostic about which part of it
  # (wrong VLAN, absent tunnel, trunk gap, upstream) failed. Per-station fields
  # arrive on separate lines, so they accumulate and are classified at the next
  # Station header and once more at END.
  # The log says a client left; this says what condition it was in. A station
  # that associates at -78 dBm and is gone in 40s is a coverage problem; one
  # at -50 dBm gone in 40s is not, and no amount of log counting separates
  # them. Buckets, not a median: busybox awk has no asort, and the shape of
  # the distribution is what the question needs anyway.
  # The match must be anchored on the FIRST field. iw prints three lines per
  # station whose text ends in "signal:" -- "signal:", "last ack signal:" and
  # "avg ack signal:" -- so a bare /signal:/ fires three times, and on the two
  # ack lines $2 is the word "ack", which coerces to 0 and lands every bogus
  # read in the strongest bucket. That inflated "signal >= -60 dBm" to 82% on
  # the first run. $1=="signal:" also excludes "signal avg:" for free.
  "sd=$(for i in $(iw dev 2>/dev/null | awk '/Interface/{print $2}'); do " \
    "iw dev \"$i\" station dump 2>/dev/null; done | awk '" \
    "function fl(){ if(h && sg > -900 && ct >= 300 && sg >= -70){ el++; " \
      "if(rb + tb < 10240) si++ } h=0; sg=-999; ct=0; rb=0; tb=0 } " \
    "/^Station/{fl(); h=1; n++} " \
    "$1==\"signal:\"{sg=$2+0; if(sg>=-60)a++; else if(sg>=-70)b++; " \
      "else if(sg>=-80)c++; else d++} " \
    "$1==\"rx\" && $2==\"bytes:\"{rb=$3+0} " \
    "$1==\"tx\" && $2==\"bytes:\"{tb=$3+0} " \
    "$1==\"connected\" && $2==\"time:\"{ct=$3+0; " \
      "if(ct<60)x++; else if(ct<300)y++; else z++} " \
    "$1==\"inactive\" && $2==\"time:\"{iv=$3+0; " \
      "if(iv<1000)p++; else if(iv<60000)q++; else r++} " \
    "END{fl(); print n+0,a+0,b+0,c+0,d+0,x+0,y+0,z+0,el+0,si+0,p+0,q+0,r+0}')",
  'set -- $sd', 'nsta=${1:-0}', 'sg60=${2:-0}', 'sg70=${3:-0}',
  'sg80=${4:-0}', 'sg81=${5:-0}', 'ct60=${6:-0}', 'ct300=${7:-0}',
  'ctlong=${8:-0}', 'stelig=${9:-0}', 'stsil=${10:-0}',
  'iact=${11:-0}', 'iamid=${12:-0}', 'iastale=${13:-0}',
  # Radio settings as the driver actually has them, which is not necessarily
  # what the rXg intended -- regulatory and hardware clamping both apply
  # after the config is pushed. Channel and width share one line and txpower
  # follows it, so the parse is ordered; resetting on Interface stops a VAP
  # with no channel line from attributing its power to the previous radio.
  # Band is taken from the channel number because these radios are 2G/5G only.
  #
  # Only type AP counts. The sdan per-VLAN VIFs (<bss>-v<vlan>) are type
  # AP/VLAN and carry no transmit power of their own -- they inherit the
  # parent radio, so nl80211 reports txpower 0.00 for them. iw dev lists in
  # descending ifindex, which puts those VIFs BEFORE the base VAPs, so a
  # first-match parse on an AP that happens to have clients on dynamic VLANs
  # reports 0 dBm for a radio that is actually at full power. That is what
  # produced the phantom "0 dBm" radios in earlier surveys.
  #
  # Matching is field-anchored rather than by regex: /txpower/ and /channel /
  # also match substrings of unrelated lines, which is the same trap that once
  # made /signal:/ pick up "last ack signal:".
  "rd=$(iw dev 2>/dev/null | awk '" \
    "$1==\"Interface\"{cch=-1; ct=\"\"} " \
    "$1==\"type\"{ct=$2} " \
    "$1==\"channel\"{cch=$2+0; cw=0; for(i=1;i<=NF;i++) if($i==\"width:\") cw=$(i+1)+0} " \
    "$1==\"txpower\"{if(cch>0 && ct==\"AP\"){ tp=$2+0; " \
      "if(cch<=14){ if(!g2){c2=cch; w2=cw; p2=tp; g2=1} } " \
      "else { if(!g5){c5=cch; w5=cw; p5=tp; g5=1} } }} " \
    "END{print c2+0, w2+0, p2+0, c5+0, w5+0, p5+0}')",
  'set -- $rd', 'ch2=${1:-0}', 'bw2=${2:-0}', 'txp2=${3:-0}',
  'ch5=${4:-0}', 'bw5=${5:-0}', 'txp5=${6:-0}',
  # Channel utilisation. Once uniform TX power and an uncorrelated weak-signal
  # share have both been ruled out as the cause of the per-building spread,
  # contention is what is left, and busy/active time is the direct measure of
  # it -- a client that hears its AP at -50 dBm but cannot get airtime behaves
  # exactly like one that is out of range.
  # Two samples 2s apart, NOT the raw counters: those accumulate from
  # interface-up and most of this fleet has been up over a week, so the
  # lifetime average is dominated by history and says nothing about now.
  # Survey data is per-phy, so every VAP on a radio returns the same block and
  # taking the first [in use] one per band is enough. "channel active time"
  # precedes "channel busy time" in iw's output, so capturing on the first and
  # committing on the second keeps them paired.
  "SVA='/^Survey data/{u=0} " \
    "/frequency:/{f=$2+0; u=($0 ~ /in use/)} " \
    "u && /channel active time:/{at=$4+0} " \
    "u && /channel busy time:/{bt=$4+0; " \
      "if(f<3000){ if(!g2){a2=at; b2=bt; g2=1} } " \
      "else { if(!g5){a5=at; b5=bt; g5=1} } } " \
    "END{print a2+0,b2+0,a5+0,b5+0}'",
  "svd() { for i in $(iw dev 2>/dev/null | awk '/Interface/{print $2}'); do " \
    "iw dev \"$i\" survey dump 2>/dev/null; done | awk \"$SVA\"; }",
  'sv1=$(svd)', 'sleep 2', 'sv2=$(svd)',
  'set -- $sv1', 'p1=${1:-0}', 'p2=${2:-0}', 'p3=${3:-0}', 'p4=${4:-0}',
  'set -- $sv2', 'q1=${1:-0}', 'q2=${2:-0}', 'q3=${3:-0}', 'q4=${4:-0}',
  'sa2=$((q1 - p1))', 'sb2=$((q2 - p2))',
  'sa5=$((q3 - p3))', 'sb5=$((q4 - p4))',
  # sdan drops any VLAN interface that has no tunnel, so a wlanX-vNNNN with no
  # matching tunnel is a client being silently discarded right now. Capture the
  # tunnel list once rather than shelling out per interface.
  "tl=$(#{CTL} tunnels 2>/dev/null | sed -n 's/.*vlan=\\([0-9]*\\).*/\\1/p')",
  # Per-VLAN counters, absolute. NOTE a one-way test was tried here and
  # removed: on per-unit VLANs there is usually a single client and nothing
  # else on the VLAN to talk to it, so tx==0 is the normal state of a quiet
  # VLAN rather than evidence of a missing path -- while rx keeps ticking from
  # unsolicited mDNS/SSDP/ARP that nothing is obliged to answer. Across eight
  # fleet runs every flagged AP/VLAN pair appeared exactly once and none
  # recurred, i.e. pure noise. Proving a VLAN's path needs an ACTIVE test
  # (rxg_sdanctl create-diag <vlan> --dhcp), which needs kmod-veth present.
  'vifs=0', 'orph=0', 'vlow=0', 'vtx0=0',
  'for d in /sys/class/net/wlan*-v*; do ' \
    '[ -d "$d" ] || continue; ' \
    'vifs=$((vifs+1)); v=${d##*-v}; ' \
    'if ! echo "$tl" | grep -qx "$v"; then orph=$((orph+1)); fi; ' \
    'rx=$(cat "$d/statistics/rx_packets" 2>/dev/null); rx=${rx:-0}; ' \
    'tx=$(cat "$d/statistics/tx_packets" 2>/dev/null); tx=${tx:-0}; ' \
    'if [ "$rx" -lt 100 ]; then vlow=$((vlow+1)); fi; ' \
    'if [ "$tx" -eq 0 ]; then vtx0=$((vtx0+1)); fi; ' \
  'done',
  'tuns=$(echo "$tl" | grep -c "[0-9]")', 'tuns=${tuns:-0}',
  # The MAC table is client ground truth -- it is what the kernel actually uses
  # to deliver the return path, and it carries the VLAN the AP believes each
  # client is on. Two questions come out of it that netdev counters cannot
  # answer:
  #   macdrop  a MAC learned on a VLAN with no tunnel. Not "an idle unit" the
  #            way a bare orphan netdev is -- a client, present, being dropped.
  #   smv      mac:vlan pairs, so the rXg can compare the AP's belief against
  #            the VLAN its own Access-Accept assigned. Independent sources, so
  #            a disagreement is observed rather than inferred from the code.
  # The vlan token is optional in this output, so it is found by prefix rather
  # than by column. tl is newline-separated; flatten it first, because awk -v
  # with embedded newlines is not worth relying on.
  "tls=$(echo \"$tl\" | tr '\\n' ' ')",
  "mo=$(#{CTL} macs 2>/dev/null | awk -v tl=\"$tls\" '" \
    "BEGIN{ nt=split(tl, ta, \" \"); " \
      "for(i=1;i<=nt;i++) if(ta[i] != \"\") T[ta[i]]=1 } " \
    "$1 ~ /^[0-9a-f][0-9a-f]:/{ v=\"\"; " \
      "for(i=1;i<=NF;i++) if($i ~ /^vlan=/) v=substr($i, 6); " \
      "if(v == \"\") next; " \
      "c++; if(!(v in T)) dr++; " \
      "if(k < 40){ pl = pl \",\" $1 \":\" v; k++ } } " \
    "END{print c+0, dr+0, (pl == \"\" ? \"-\" : substr(pl, 2))}')",
  'set -- $mo', 'macn=${1:-0}', 'macdrop=${2:-0}', 'smv=${3:--}',
  # Switch VLAN drift. On swconfig APs an untunneled ("plain VLAN on uplink")
  # tunnel only carries traffic if its VID is in the switch chip's VLAN table.
  # rg_wss_client programs that once, when the tunnel is first added, by
  # committing a switch_vlan section to /etc/config/network. Every uCentral
  # apply copies /etc/config-shadow over /etc/config/network and reloads, which
  # wipes those sections and the hardware entries with them -- while the
  # client's in-memory tunnel cache still says "programmed", and since 1.0.38
  # tunnels stay resident, so nothing re-adds them. The client then associates,
  # authorizes and passes nothing.
  #   vtun     sdan tunnels of type vlan (the only kind that needs the switch)
  #   swok     of those, VIDs present with member ports on some switch
  #   swmiss   of those, VIDs on NO switch -- the dead ones
  #   ucimiss  of those, VIDs with no switch_vlan section in UCI
  #   swmac    sdan-learned client MACs sitting on a swmiss VLAN right now
  #   netcfgage  seconds since /etc/config/network was last written; less than
  #            clientage means it was rewritten under a running client, which
  #            is the precondition for drift
  # swen is enable_vlan per switch: with 802.1Q off the table does not filter
  # and a missing VID is harmless.
  "swl=$(swconfig list 2>/dev/null | awk '{print $2}')",
  'swv=""', 'swe=""', 'swn=0',
  'for s in $swl; do swn=$((swn+1)); ' \
    'swv="$swv $(swconfig dev $s show 2>/dev/null | awk \'' \
      '$1=="VLAN"{v=$2; sub(/:$/,"",v)} $1=="vid:"{v=$2} ' \
      '$1=="ports:" && NF>1 && v!=""{print v}\' | tr "\\n" " ")"; ' \
    'e=$(swconfig dev $s get enable_vlan 2>/dev/null); swe="$swe,$s:${e:-x}"; ' \
  'done',
  'swe=${swe#,}', 'swe=${swe:--}',
  # swpath: is the uplink behind the switch at all? Mirrors the client's own
  # test -- it only programs a switch for which capabilities.json or board.json
  # names a WAN port. Without one (ipq60xx, ipq53xx, mediatek here) the switch
  # is not in the uplink path, a missing VID there means nothing, and the client
  # never programs it.
  'swp=0',
  "if [ -n \"$(jsonfilter -i /etc/ucentral/capabilities.json " \
    "-e '@.switch_ports.*.wan[0]' 2>/dev/null)$(jsonfilter -i /etc/board.json " \
    "-e '@.switch.*.ports[@.role=\"wan\"].num' 2>/dev/null)\" ]; then swp=1; fi",
  "vtl=$(#{CTL} tunnels 2>/dev/null | awk '{t=0; v=\"\"; " \
    "for(i=1;i<=NF;i++){ if($i ~ /^vlan=/) v=substr($i,6); if($i==\"type=vlan\") t=1 } " \
    "if(t && v!=\"\") print v}' | tr '\\n' ' ')",
  "ucv=$(uci -q show network 2>/dev/null | awk -F= '$1 ~ /^network[.][^.]*[.]vlan$/{ " \
    "v=$2; gsub(/[^0-9]/,\"\",v); if(v!=\"\") print v}' | tr '\\n' ' ')",
  "swd=$(echo | awk -v vt=\"$vtl\" -v sw=\"$swv\" -v uc=\"$ucv\" '" \
    "END{ n=split(sw,a,\" \"); for(i=1;i<=n;i++) S[a[i]]=1; " \
      "n=split(uc,b,\" \"); for(i=1;i<=n;i++) U[b[i]]=1; " \
      "n=split(vt,c,\" \"); for(i=1;i<=n;i++){ v=c[i]; " \
        "if(v==\"\" || (v in D)) continue; D[v]=1; t++; " \
        "if(v in S) ok++; else { m++; ml=ml \",\" v } " \
        "if(!(v in U)) um++ } " \
      "print t+0, ok+0, m+0, um+0, (ml==\"\" ? \"-\" : substr(ml,2)) }')",
  'set -- $swd', 'vtun=${1:-0}', 'swok=${2:-0}', 'swm=${3:-0}', 'ucm=${4:-0}',
  'swmlf=${5:--}', 'swml=$(echo "$swmlf" | cut -d, -f1-30)',
  # Matched against the full missing list, not the capped one printed above.
  "smo=$(#{CTL} macs 2>/dev/null | awk -v ml=\"$swmlf\" '" \
    "BEGIN{ n=split(ml,a,\",\"); for(i=1;i<=n;i++) M[a[i]]=1 } " \
    "$1 ~ /^[0-9a-f][0-9a-f]:/{ v=\"\"; " \
      "for(i=1;i<=NF;i++) if($i ~ /^vlan=/) v=substr($i,6); " \
      "if(v != \"\" && (v in M)){ c++; if(k<20){ l=l \",\" $1 \":\" v; k++ } } } " \
    "END{print c+0, (l==\"\" ? \"-\" : substr(l,2))}')",
  'set -- $smo', 'swmac=${1:-0}', 'swmacl=${2:--}',
  # Per-station view of every client sdan knows, so the rXg can join it to its
  # own DHCP log: mac,vlan,dead(1 = its VLAN is in swmlf),connected_s,
  # tx_bytes(AP->STA),rx_bytes; ';'-separated, clients on missing VLANs first so
  # the cap never drops them. A missing VLAN that really drops traffic shows up
  # here as an associated client the AP has sent almost nothing to.
  "stv=$({ #{CTL} macs 2>/dev/null | sed 's/^/M /'; echo '@@'; " \
    "for i in $(iw dev 2>/dev/null | awk '$1==\"Interface\"{print $2}'); do " \
      "iw dev \"$i\" station dump 2>/dev/null; done; } | awk -v ml=\"$swmlf\" '" \
    "function fl(){ if(h && !(m in S) && (m in V)){ S[m]=1; v=V[m]; d=((v in D) ? 1 : 0); " \
      "rec=m \",\" v \",\" d \",\" ct \",\" tb \",\" rb; " \
      "if(d){ nd++; DL[nd]=rec } else { nl++; LL[nl]=rec } } h=0 } " \
    "BEGIN{ n=split(ml, a, \",\"); for(i=1;i<=n;i++) D[a[i]]=1 } " \
    "$1==\"M\" && $2 ~ /^[0-9a-fA-F][0-9a-fA-F]:/{ " \
      "for(i=3;i<=NF;i++) if($i ~ /^vlan=/) V[tolower($2)]=substr($i,6); next } " \
    "$0==\"@@\"{ ph=1; next } " \
    "ph && $1==\"Station\"{ fl(); m=tolower($2); h=1; ct=0; tb=0; rb=0 } " \
    "ph && $1==\"rx\" && $2==\"bytes:\"{ rb=$3+0 } " \
    "ph && $1==\"tx\" && $2==\"bytes:\"{ tb=$3+0 } " \
    "ph && $1==\"connected\" && $2==\"time:\"{ ct=$3+0 } " \
    "END{ fl(); out=\"\"; k=0; " \
      "for(i=1;i<=nd && k<60;i++){ out=out \";\" DL[i]; k++ } " \
      "for(i=1;i<=nl && k<60;i++){ out=out \";\" LL[i]; k++ } " \
      "print (out==\"\" ? \"-\" : substr(out,2)) }')",
  'nfa=-1', 'nm=$(date -r /etc/config/network +%s 2>/dev/null)',
  'if [ -n "$nm" ]; then nfa=$(( $(date +%s) - nm )); fi',
  # The client also falls back to board.json network.wan.ports ("switch0:4"
  # style) for switch0, and the board files are not the last word anyway --
  # so also count the client's own verdicts from the log ring. swprog > 0 is
  # proof it programmed a switch; swnowan > 0 is proof it declined to.
  "wp=$(jsonfilter -i /etc/board.json -e '@.network.wan.ports[0]' 2>/dev/null)",
  'case "$wp" in *:[0-9]*) swp=1 ;; esac',
  "swprog=$(logread 2>/dev/null | grep -c 'swconfig VLAN [0-9]* \\(already \\)\\{0,1\\}programmed on')",
  'swprog=${swprog:-0}',
  "swnowan=$(logread 2>/dev/null | grep -c 'no WAN/CPU port discoverable')", 'swnowan=${swnowan:-0}',
  # TX power as the radio actually has it, its ceiling, and the clients it
  # serves -- per band. The older radio block above keys band on channel
  # number, which files a 6 GHz radio under 2.4; this one keys on frequency.
  #   tpXa  applied dBm (first type-AP VAP on that band; VLAN VIFs report 0)
  #   tpXm  ceiling: highest entry in iwinfo txpowerlist for that VAP
  #   sXn / sXw / sXe   stations on that band / below -70 / below -80 dBm
  #   sXs   sum of their signals, for a mean without a second pass
  # Stations are attributed through the parent VAP: sdan's per-VLAN VIFs
  # (wlan1-v2645) carry no channel of their own.
  "tpw=$({ iw dev 2>/dev/null; echo '@@'; " \
    "for i in $(iw dev 2>/dev/null | awk '$1==\"Interface\"{print $2}'); do " \
      "echo \"IF $i\"; iw dev \"$i\" station dump 2>/dev/null; done; } | awk '" \
    "function bd(f){ return f < 3000 ? 2 : (f < 5925 ? 5 : 6) } " \
    "!ph && $1==\"Interface\"{ ifc=$2; ty=\"\"; fq=0 } " \
    "!ph && $1==\"type\"{ ty=$2 } " \
    "!ph && $1==\"channel\"{ fq=$3; gsub(/[^0-9]/, \"\", fq); fq+=0; " \
      "if(ty==\"AP\" && fq>0) B[ifc]=bd(fq) } " \
    "!ph && $1==\"txpower\"{ if(ty==\"AP\" && fq>0){ b=bd(fq); " \
      "if(!(b in TA)){ TA[b]=$2+0; RI[b]=ifc } } } " \
    "$0==\"@@\"{ ph=1; next } " \
    "ph && $1==\"IF\"{ cur=$2; base=cur; sub(/-v[0-9]+$/, \"\", base); cb=B[base] } " \
    "ph && $1==\"signal:\" && cb{ sg=$2+0; N[cb]++; S[cb]+=sg; " \
      "if(sg < -70) W[cb]++; if(sg < -80) E[cb]++ } " \
    "END{ for(b=2; b<=6; b++){ if(b==3 || b==4) continue; mx=-1; " \
        "if(b in RI){ cmd=\"iwinfo \" RI[b] \" txpowerlist 2>/dev/null\"; " \
          "while((cmd | getline ln) > 0){ sub(/^[* ]+/, \"\", ln); split(ln, z, \" \"); " \
            "if(z[1] ~ /^-?[0-9]+$/ && z[1]+0 > mx) mx=z[1]+0 } close(cmd) } " \
        "printf \"%s %s %d %d %d %d \", ((b in TA) ? TA[b] : -1), mx, " \
          "N[b]+0, W[b]+0, E[b]+0, S[b]+0 } print \"\" }')",
  'set -- $tpw',
  'tp2a=${1:--1}', 'tp2m=${2:--1}', 's2n=${3:-0}', 's2w=${4:-0}', 's2e=${5:-0}', 's2s=${6:-0}',
  'tp5a=${7:--1}', 'tp5m=${8:--1}', 's5n=${9:-0}', 's5w=${10:-0}', 's5e=${11:-0}', 's5s=${12:-0}',
  'tp6a=${13:--1}', 'tp6m=${14:--1}', 's6n=${15:-0}', 's6w=${16:-0}', 's6e=${17:-0}', 's6s=${18:-0}',
  # Per-station link quality, for the cell-overlap section:
  # mac,band,vlan,signal,tx_mbps,rx_mbps,tx_retries,tx_failed,tx_packets,
  # connected_s,inactive_ms; ';'-separated, '-' where the driver reports none.
  # Band comes from the parent VAP, as in tpw above; a station listed under
  # both a VLAN VIF and its parent is reported once.
  "stq=$({ #{CTL} macs 2>/dev/null | sed 's/^/M /'; echo '@@'; iw dev 2>/dev/null; echo '@@'; " \
    "for i in $(iw dev 2>/dev/null | awk '$1==\"Interface\"{print $2}'); do " \
      "echo \"IF $i\"; iw dev \"$i\" station dump 2>/dev/null; done; } | awk '" \
    "function bd(f){ return f < 3000 ? 2 : (f < 5925 ? 5 : 6) } " \
    "function fl(){ if(h && !(m in S)){ S[m]=1; k++; if(k <= #{STQ_CAP}) " \
      "out=out \";\" m \",\" cb \",\" ((m in V) ? V[m] : \"-\") \",\" sg \",\" txr \",\" rxr " \
      "\",\" rt \",\" tf \",\" tp \",\" ct \",\" iv } h=0 } " \
    "ph==0 && $1==\"M\" && $2 ~ /^[0-9a-fA-F][0-9a-fA-F]:/{ " \
      "for(i=3;i<=NF;i++) if($i ~ /^vlan=/) V[tolower($2)]=substr($i,6); next } " \
    "$0==\"@@\"{ ph++; next } " \
    "ph==1 && $1==\"Interface\"{ ifc=$2; ty=\"\" } " \
    "ph==1 && $1==\"type\"{ ty=$2 } " \
    "ph==1 && $1==\"channel\"{ fq=$3; gsub(/[^0-9]/, \"\", fq); fq+=0; " \
      "if(ty==\"AP\" && fq>0) B[ifc]=bd(fq) } " \
    "ph==2 && $1==\"IF\"{ fl(); base=$2; sub(/-v[0-9]+$/, \"\", base); " \
      "cb=((base in B) ? B[base] : 0) } " \
    "ph==2 && $1==\"Station\"{ fl(); m=tolower($2); h=1; sg=\"-\"; txr=\"-\"; rxr=\"-\"; " \
      "rt=\"-\"; tf=\"-\"; tp=\"-\"; ct=\"-\"; iv=\"-\" } " \
    "ph==2 && $1==\"signal:\"{ sg=$2+0 } " \
    "ph==2 && $1==\"tx\" && $2==\"bitrate:\"{ txr=$3+0 } " \
    "ph==2 && $1==\"rx\" && $2==\"bitrate:\"{ rxr=$3+0 } " \
    "ph==2 && $1==\"tx\" && $2==\"retries:\"{ rt=$3+0 } " \
    "ph==2 && $1==\"tx\" && $2==\"failed:\"{ tf=$3+0 } " \
    "ph==2 && $1==\"tx\" && $2==\"packets:\"{ tp=$3+0 } " \
    "ph==2 && $1==\"connected\" && $2==\"time:\"{ ct=$3+0 } " \
    "ph==2 && $1==\"inactive\" && $2==\"time:\"{ iv=$3+0 } " \
    "END{ fl(); print (out==\"\" ? \"-\" : substr(out,2)) }')",
  # DFS: channels the 5 GHz radio currently may not use because radar was
  # detected on them (the 30-minute non-occupancy period), and the most recent
  # detection's age. iw reports "DFS state: unavailable (for N sec)".
  # Also the 5 GHz channel, keyed on frequency: ch5 above is the first
  # non-2.4 channel, which on a tri-band AP can be the 6 GHz radio's.
  "dfs=$(pc=$(iw dev 2>/dev/null | awk '$1==\"wiphy\"{w=$2} " \
      "$1==\"channel\"{ f=$3; gsub(/[^0-9]/, \"\", f); f+=0; " \
        "if(f >= 5150 && f < 5925 && !d){ print \"phy\" w, $2+0; d=1 } }'); " \
    "set -- $pc; " \
    "if [ -n \"$1\" ]; then iw phy \"$1\" channels 2>/dev/null | awk -v c=\"$2\" '" \
    "/DFS state: unavailable/{ u++; s=$0; sub(/.*for /, \"\", s); s+=0; " \
      "if(mn == \"\" || s < mn) mn=s } " \
    "END{ print u+0, (mn == \"\" ? -1 : mn), c+0 }'; else echo 0 -1 0; fi)",
  'set -- $dfs', 'dfsu=${1:-0}', 'dfsmin=${2:--1}', 'dfsch=${3:-0}',
  "radar=$(logread 2>/dev/null | grep -ci 'radar')", 'radar=${radar:-0}',
  # Dataplane readiness, as raw events for the rXg to join (the most recent
  # DPE_CAP, ';'-separated). t is seconds into the month from the syslog
  # stamp; kernel events also carry their uptime, so sub-second gaps between
  # them are exact. The shell-command rule runs first: rg_wss_client logs each
  # command's full text, and this probe's text contains every pattern below.
  #   C,t,mac,ifc,vlan   hostapd AP-STA-CONNECTED (vlanid=, or '-')
  #   D,t,mac,ifc        hostapd AP-STA-DISCONNECTED
  #   U,t,vif,uptime     sdan: the VIF's VLAN has no tunnel -- not forwarding
  #   H,t,vif,uptime     sdan: ingress hook on the VIF -- forwarding
  #   S,t                rg_wss_client began an rXg shell command (it is
  #                      blocked until it finishes)
  #   A,t,mac,code       rg_wss_client handled the station's authorization:
  #                      a tunnel applied, q deferred (tunnel config not yet
  #                      known), z sdan down, n no VLAN, f apply failed
  #   E,first,last       span of the ring, first record
  "dpe=$(logread 2>/dev/null | awk '" \
    "function ts(){ split($4, tt, \":\"); return $3*86400 + tt[1]*3600 + tt[2]*60 + tt[3] } " \
    "function up(){ for(i=1;i<=NF;i++) if($i ~ /^\\[/){ u=$i; if(u == \"[\") u=$(i+1); " \
      "gsub(/[^0-9.]/, \"\", u); return u } return \"\" } " \
    "function add(s){ n++; R[n]=s } " \
    "{ t=ts(); if(!f) f=t; la=t } " \
    "/Received shell command to run/{ add(\"S,\" t); next } " \
    "/rg_wss_client.*(sta-authorized|sdan tunnel not applied for )/{ " \
      "c=\"a\"; if(/deferring/) c=\"q\"; else if(/dataplane is down/) c=\"z\"; " \
      "else if(/no VLAN from RADIUS/) c=\"n\"; else if(/not applied/) c=\"f\"; " \
      "for(i=1;i<=NF;i++) if(length($i) == 17 && split($i, mm, \":\") == 6){ " \
        "add(\"A,\" t \",\" tolower($i) \",\" c); break } " \
      "next } " \
    "/AP-STA-CONNECTED/{ for(k=1;k<=NF;k++) if($k == \"AP-STA-CONNECTED\") break; " \
      "ifc=$(k-1); sub(/:$/, \"\", ifc); v=\"-\"; " \
      "for(i=k+2;i<=NF;i++) if($i ~ /^vlanid=/) v=substr($i, 8); " \
      "add(\"C,\" t \",\" tolower($(k+1)) \",\" ifc \",\" v); next } " \
    "/AP-STA-DISCONNECTED/{ for(k=1;k<=NF;k++) if($k == \"AP-STA-DISCONNECTED\") break; " \
      "ifc=$(k-1); sub(/:$/, \"\", ifc); add(\"D,\" t \",\" tolower($(k+1)) \",\" ifc); next } " \
    "/rxg_sdan: .*unconfigured, traffic will be dropped/{ " \
      "for(k=1;k<=NF;k++) if($k == \"rxg_sdan:\") break; add(\"U,\" t \",\" $(k+1) \",\" up()); next } " \
    "/rxg_sdan: ingress hook registered on /{ " \
      "for(k=1;k<=NF;k++) if($k == \"on\") break; add(\"H,\" t \",\" $(k+1) \",\" up()); next } " \
    "END{ out=\"E,\" f \",\" la; s=(n > #{DPE_CAP} ? n - #{DPE_CAP} + 1 : 1); " \
      "for(i=s;i<=n;i++) out=out \";\" R[i]; print out }')",
  'echo "SDANCHK rc=$rc ctl=$c ko=$k kohash=$kh ctlhash=$ch ' \
    'ctlsz=$cs kosz=$ks tgt=$t kern=$kv loaded=$lm ' \
    'memavail=$ma memtotal=$mt uptime=$up ufprss=$fp clientage=$ca ufpbin=$ub ' \
    'tmpleft=$tm overlayfree=$ov oomlogs=$oom ' \
    'conn=$conn inact=$inact undrop=$undrop ' \
    'vifs=$vifs tuns=$tuns orph=$orph vlow=$vlow ' \
    'vtx0=$vtx0 ' \
    'reassoc=$reassoc bandroam=$bandroam nmac=$nmac lspan=$lspan ' \
    'celig=$celig ' \
    'nsta=$nsta sg60=$sg60 sg70=$sg70 sg80=$sg80 sg81=$sg81 ' \
    'ct60=$ct60 ct300=$ct300 ctlong=$ctlong ' \
    'ch2=$ch2 bw2=$bw2 txp2=$txp2 ch5=$ch5 bw5=$bw5 txp5=$txp5 ' \
    'sa2=$sa2 sb2=$sb2 sa5=$sa5 sb5=$sb5 ' \
    'apfail=$apfail tundel=$tundel chain=$chain chainvl=$chainvl ' \
    'pdto=$pdto noack=$noack prune=$prune ' \
    'peerfind=$peerfind rxtid=$rxtid tdprem=$tdprem tdok=$tdok ' \
    'iact=$iact iamid=$iamid iastale=$iastale ' \
    'relgone=$relgone relmove=$relmove relcan=$relcan relfire=$relfire ' \
    'stelig=$stelig stsil=$stsil ' \
    'macn=$macn macdrop=$macdrop smv=$smv ' \
    'swn=$swn swpath=$swp swen=$swe vtun=$vtun swok=$swok swmiss=$swm ucimiss=$ucm ' \
    'swml=$swml swmac=$swmac swmacl=$swmacl netcfgage=$nfa ' \
    'swprog=$swprog swnowan=$swnowan stv=$stv ' \
    'tp2a=$tp2a tp2m=$tp2m s2n=$s2n s2w=$s2w s2e=$s2e s2s=$s2s ' \
    'tp5a=$tp5a tp5m=$tp5m s5n=$s5n s5w=$s5w s5e=$s5e s5s=$s5s ' \
    'tp6a=$tp6a tp6m=$tp6m s6n=$s6n s6w=$s6w s6e=$s6e s6s=$s6s ' \
    'clirss=$cr crashn=$clg crashpanic=$cpan crashoom=$coom ' \
    'crashsrc=$csrc crashpath=$cpath ' \
    'stq=$stq dfsu=$dfsu dfsmin=$dfsmin dfsch=$dfsch radar=$radar dpe=$dpe ' \
    'macs=$macs emacs=$emacs"'
].join('; ')

def pct(n, total)
  total.zero? ? '  - ' : format('%3d%%', (100.0 * n / total).round)
end

# Print a count/percent cross-tab, worst first. `key_label` heads column one;
# each row is [key, ok_count, bad_count, breakdown_string].
def cross_tab(title, key_label, rows_by_key, width: 30)
  puts title
  puts format('  %-*s %5s %5s %5s  %s', width, key_label, 'OK', 'BAD', 'BAD%', 'VERDICTS')
  rows_by_key.each do |key, group|
    g_ok  = group.count { |r| r[:verdict] == 'OK' }
    g_bad = group.size - g_ok
    seen  = group.reject { |r| r[:verdict] == 'OK' }
                 .group_by { |r| r[:verdict] }
                 .sort_by { |_, rs| -rs.size }
                 .map { |v, rs| "#{v} x#{rs.size}" }
    puts format('  %-*s %5d %5d %5s  %s', width, key.to_s[0, width],
                g_ok, g_bad, pct(g_bad, group.size), seen.join(', '))
  end
  puts
end

# What this rXg can actually serve: platform/kernel -> {file => size}.
builds = {}
if SDAN_DIR.directory?
  # Glob the kernel directories rather than the .ko, so a directory that holds
  # only one of the two files is still visible -- a half-populated build serves
  # ctl with a 404 on ko, which is indistinguishable on the AP from a failed
  # ko download.
  Dir.glob(SDAN_DIR.join('*', '*', '*')).select { |d| File.directory?(d) }.each do |dir|
    rel = Pathname.new(dir).relative_path_from(SDAN_DIR).to_s
    ko  = File.join(dir, 'rxg_sdan.ko')
    ctl = File.join(dir, 'rxg_sdanctl')
    builds[rel] = {
      'rxg_sdan.ko' => (File.size(ko)  if File.exist?(ko)),
      'rxg_sdanctl' => (File.size(ctl) if File.exist?(ctl)),
    }
  end
end

scope = AccessPoint.joins(:infrastructure_device)
                   .where(infrastructure_devices: { device: 'openwifi_wss' })
                   .where(online: true)
scope = scope.where(access_point_zone_id: ENV['ZONE']) if ENV['ZONE'].present?
scope = scope.limit(ENV['LIMIT'].to_i) if ENV['LIMIT'].present?

aps = scope.order(:id).to_a
if aps.empty?
  puts 'No online openwifi_wss APs matched.'
  exit 0
end

# Dispatch the probe to `aps` and return one row per AP. Extracted so the
# post-remediation verification pass can reuse it unchanged.
def probe_aps(aps, builds, timeout, label: 'Probing')
  warn "#{label} #{aps.size} AP(s), waiting up to #{timeout}s for replies..."

  ce_for = {}
  aps.each_slice(25).each_with_index do |slice, i|
    slice.each do |ap|
      ce_for[ap.id] = CommandExecution.create!(source: ap, command: PROBE,
                                               created_by: 'sdan_fleet_check')
    rescue StandardError => e
      warn "  dispatch failed for AP #{ap.id} (#{ap.name}): #{e.class}: #{e.message}"
    end
    warn "  dispatched #{ce_for.size}/#{aps.size}"
    sleep 2 if (i + 1) * 25 < aps.size
  end

  ids = ce_for.values.map(&:id)
  deadline = Time.current + timeout
  last_seen = -1
  until CommandExecution.where(id: ids).where.not(exit_code: nil).count >= ids.size ||
        Time.current >= deadline
    replied = CommandExecution.where(id: ids).where.not(exit_code: nil).count
    if replied != last_seen
      warn "  replied #{replied}/#{ids.size} (#{(deadline - Time.current).to_i}s left)"
      last_seen = replied
    end
    sleep 3
  end
  results = CommandExecution.where(id: ids).index_by(&:id)

  aps.map do |ap|
    base = { ap: ap, fw: ap.version.presence || '(unknown)',
             cv: ap.wss_client_version.presence || '(unknown)', f: {}, raw: '' }
    ce = ce_for[ap.id]
    next base.merge(verdict: 'DISPATCH_FAILED') unless ce

    ce_r = results[ce.id] || ce
    text = ce_r.result_text.to_s
    marker = text.index('SDANCHK')
    # No reply inside the window, or output we could not read. Either way we
    # cannot say the AP is healthy, so it is reported rather than assumed OK.
    next base.merge(verdict: 'NO_REPLY', raw: text.strip[0, 60]) unless marker

    # Whitespace-normalise the whole payload before scanning. Reading only the
    # first line meant one stray newline inside a value discarded every field
    # after it, which silently emptied the memory columns on exactly the APs
    # that were failing -- the keys are distinctive, so scanning wider is safe.
    line = text[marker..].gsub(/\s+/, ' ').strip
    f = line.scan(/(\w+)=(\S+)/).to_h
    # The client looks up the kernel as uname -r with any suffix stripped.
    pair = "#{f['tgt']}/#{f['kern'].to_s.split('-').first}"
    have = builds[pair]

    verdict =
      if f['rc'] == '0'                                        then 'OK'
      elsif f['ctl'] != '1' && f['ko'] != '1'
        have ? 'FETCH FAILED (both)' : 'NO BUILD ON RXG'
      elsif f['ko'] != '1'
        (have && have['rxg_sdan.ko']) ? 'FETCH FAILED (ko)' : 'NO BUILD ON RXG (ko)'
      elsif f['kohash'] != '1'                                 then 'CORRUPT (ko incomplete)'
      elsif have && have['rxg_sdan.ko'] && f['kosz'].to_i != have['rxg_sdan.ko']
        'CORRUPT (ko wrong size)'
      elsif have && have['rxg_sdanctl'] && f['ctlsz'].to_i != have['rxg_sdanctl']
        'CORRUPT (ctl wrong size)'
      else 'PRESENT NOT LOADED'
      end

    base.merge(verdict: verdict, f: f, raw: line.strip, pair: pair, have: !have.nil?,
               at: ce_r.updated_at)
  end
end

probe_started = Time.current
rows = probe_aps(aps, builds, TIMEOUT)

# Collect the detached probe-request captures. Each AP started its window when
# the probe reached it; dispatch is paced at 25 APs per 2 s, so the last window
# opens about that long after the first. An AP still capturing answers
# "pending" and gets one more try.
def collect_probe_captures(rows, started)
  targets = rows.select { |r| r[:f].present? }
  return if targets.empty?
  ready_at = started + (targets.size / 25.0).ceil * 2 + PROBE_SECS + 10
  wait = (ready_at - Time.current).ceil
  if wait.positive?
    warn "Waiting #{wait}s for the #{PROBE_SECS}s probe-request captures to finish..."
    sleep wait
  end
  2.times do |attempt|
    pending = targets.reject { |r| r[:prb] }
    break if pending.empty?
    ce_for = {}
    pending.each do |r|
      ce_for[r] = CommandExecution.create!(source: r[:ap], command: PRB_COLLECT,
                                           created_by: 'sdan_fleet_check')
    rescue StandardError => e
      warn "  capture collect failed for AP #{r[:ap].id}: #{e.class}: #{e.message}"
    end
    ids = ce_for.values.map(&:id)
    deadline = Time.current + 60
    sleep 3 until CommandExecution.where(id: ids).where.not(exit_code: nil).count >= ids.size ||
                  Time.current >= deadline
    results = CommandExecution.where(id: ids).index_by(&:id)
    ce_for.each do |r, ce|
      text = (results[ce.id] || ce).result_text.to_s
      next unless (i = text.index('SDANPRB'))
      f = text[i..].gsub(/\s+/, ' ').scan(/(\w+)=(\S+)/).to_h
      next if f['n'] == '-1' && attempt.zero?   # still capturing
      recs = f['prb'].to_s.split(';')
      r[:asc] = recs.select { |rec| rec.start_with?('A:') }.filter_map do |rec|
        mac, freq, auths, ok, low, high, open, maxun = rec.delete_prefix('A:').split(',')
        next unless maxun
        { mac: mac, freq: freq.to_i, auths: auths.to_i, ok: ok.to_i, low: low.to_i,
          high: high.to_i, open: open.to_i, maxun: maxun.to_i }
      end
      r[:prb] = recs.reject { |rec| rec.start_with?('A:') }.filter_map do |rec|
        mac, freq, max, mean, count = rec.split(',')
        next unless mac && count
        { mac: mac, freq: freq.to_i, max: max.to_i, mean: mean.to_i, count: count.to_i }
      end
    end
    sleep 15 if attempt.zero? && targets.any? { |r| r[:prb].nil? }
  end
  got = targets.count { |r| r[:prb]&.any? }
  warn "  probe captures collected from #{got}/#{targets.size} AP(s)"
end

collect_probe_captures(rows, probe_started) if PROBE_SECS.positive?

total = rows.size
ok    = rows.count { |r| r[:verdict] == 'OK' }
bad   = rows.reject { |r| r[:verdict] == 'OK' }

puts
puts "=== rxg_sdan fleet survey: #{ok}/#{total} online AP(s) running it " \
     "(#{pct(bad.size, total).strip} not) ==="
puts

retryable = bad.count { |r| r[:verdict].start_with?('FETCH FAILED', 'CORRUPT') }
blocked   = bad.count { |r| r[:verdict].start_with?('NO BUILD') }
unknown   = bad.size - retryable - blocked
puts 'Actionability:'
puts format('  %-34s %4d   a re-fetch could fix these', 'retryable (fetch/corrupt)', retryable)
puts format('  %-34s %4d   rXg cannot serve them; retrying will not help', 'blocked (no build on rXg)', blocked)
puts format('  %-34s %4d   no reply / loaded-but-failing; needs a look', 'other', unknown)
puts

puts "rXg can serve #{builds.size} platform/kernel pair(s):"
if builds.empty?
  puts '  (none -- every AP will 404; is this a dev tree with sync_assets unrun?)'
else
  builds.sort.each do |key, f|
    puts format('  %-34s ko=%-8s ctl=%s', key, f['rxg_sdan.ko'] || '-',
                f['rxg_sdanctl'] || 'MISSING')
  end
end
puts

reported = rows.reject { |r| r[:pair].nil? }
missing_pair = total - reported.size
cross_tab('By target / kernel  (does the rXg have this build?):', 'TARGET/KERNEL',
          reported.group_by { |r| "#{r[:pair]} #{r[:have] ? '[have]' : '[MISSING]'}" }
                  .sort_by { |_, g| -g.count { |r| r[:verdict] != 'OK' } },
          width: 38)
puts "  (#{missing_pair} AP(s) never reported target/kernel)\n\n" if missing_pair > 0

cross_tab('By AP firmware:', 'FIRMWARE',
          rows.group_by { |r| r[:fw] }
              .sort_by { |_, g| -g.count { |r| r[:verdict] != 'OK' } })

cross_tab('By rg_wss_client version:', 'CLIENT',
          rows.group_by { |r| r[:cv] }
              .sort_by { |_, g| -g.count { |r| r[:verdict] != 'OK' } },
          width: 14)

# Zone is not populated at every property, but the AP naming convention is:
# "AP-Room-8207-MP-Bldg-8" / "AP-Game-Room-Clubhouse-2". Taking the trailing
# <word>-<number> recovers the building, which is the unit a switch or riser
# fault actually follows.
def building_of(name)
  # "AP329-Bldg 20-EAP111-RM2008" carries the building mid-name, with a space.
  return "Bldg-#{$1}" if name.to_s =~ /\bBldg[\s-]*(\d+)\b/i
  parts = name.to_s.split('-')
  return '(unparsed)' if parts.size < 2
  last, prev = parts[-1], parts[-2]
  (last =~ /\A\d+\z/ && prev =~ /\A[A-Za-z]+\z/) ? "#{prev}-#{last}" : '(unparsed)'
end

# MAC formats differ between tables (RADIUS normalises, Device may not), so
# match on the hex digits alone. Lookups are always by an explicit variant
# list rather than a lower(replace(...)) expression, which would not use the
# index on devices.mac.
def fmt_dur(sec)
  s = sec.to_i
  return "#{s}s" if s < 90
  return "#{(s / 60.0).round}m" if s < 5400
  # Crash records and uptimes are measured in days; the 48h threshold keeps
  # every existing caller's output unchanged, since log-ring spans are bounded
  # near 24h by the wrap they would otherwise suffer.
  return format('%.1fh', s / 3600.0) if s < 172_800

  format('%.1fd', s / 86_400.0)
end

# busy/active as a percentage, or nil when the sample cannot be trusted: a
# radio with no active time in the window (idle or survey unsupported), or a
# negative delta, which means the counters were reset under us mid-sample.
def median(vals)
  return nil if vals.empty?

  sorted = vals.sort
  sorted[sorted.size / 2]
end

def busy_pct(f, active_key, busy_key)
  a = f[active_key].to_i
  b = f[busy_key].to_i
  return nil if a <= 0 || b.negative? || b > a

  100.0 * b / a
end

def norm_mac(m)
  m.to_s.downcase.gsub(/[^0-9a-f]/, '')
end

def mac_variants(macs)
  macs.flat_map do |m|
    n = norm_mac(m)
    next [] if n.length != 12
    colon = n.scan(/../).join(':')
    [m, n, colon, colon.upcase, n.scan(/../).join('-')]
  end.uniq
end

def account_label(acct)
  return '(no account)' unless acct
  full = [acct.first_name, acct.last_name].map(&:presence).compact.join(' ')
  acct.login.presence || full.presence || acct.email.presence || "account ##{acct.id}"
end

# MAC -> Account via the registered Device only. For a device that is FAILING
# authentication the other candidates actively mislead: the RADIUS row's own
# account_id belongs to a request that was rejected, and a ConnectedClient row
# is some other session that succeeded. Device.mac is the only record that
# says who owns this hardware regardless of whether it can get on.
def accounts_for_macs(macs)
  v = mac_variants(macs)
  return {} if v.empty?
  by_norm = {}
  Device.where(mac: v).where.not(account_id: nil).pluck(:mac, :account_id)
        .each { |m, a| by_norm[norm_mac(m)] ||= a }
  accts = Account.where(id: by_norm.values.uniq).index_by(&:id)
  macs.each_with_object({}) { |m, h| h[m] = accts[by_norm[norm_mac(m)]] }
end

# ---- Client outcomes -------------------------------------------------------
# This is the question a resident complaint actually asks: did people get on,
# and did they get anything? hostapd evicts a station that stops transmitting,
# so a client that associates and then passes nothing leaves as "disassociated
# due to inactivity". Evictions per association measures exactly that, and it
# is comparable across APs without normalising for log-ring span, because both
# counts come from the same ring.
#
# A snapshot of connected clients cannot answer this: the victims are evicted
# within minutes and are gone from the rXg's tables by the time anyone looks.
out_rows = rows.reject { |r| r[:f]['conn'].nil? }
if out_rows.any?
  conn_t   = out_rows.sum { |r| r[:f]['conn'].to_i }
  inact_t  = out_rows.sum { |r| r[:f]['inact'].to_i }
  undrop_t = out_rows.sum { |r| r[:f]['undrop'].to_i }
  orph_t   = out_rows.sum { |r| r[:f]['orph'].to_i }
  orph_aps = out_rows.count { |r| r[:f]['orph'].to_i > 0 }
  vlow_t   = out_rows.sum { |r| r[:f]['vlow'].to_i }

  spans = out_rows.map { |r| r[:f]['lspan'].to_i }.reject(&:zero?).sort
  med_span = spans.any? ? spans[spans.size / 2] : 0
  puts "Client outcomes  (each AP's own log ring, #{out_rows.size} AP(s) reporting):"
  if spans.any?
    short = spans.count { |x| x < 600 }
    puts format('  %-38s %6s   median ring; min %s, max %s%s',
                'OBSERVATION WINDOW', fmt_dur(med_span), fmt_dur(spans.first), fmt_dur(spans.last),
                short.positive? ? " -- #{short} AP(s) under 10m" : '')
  end
  puts format('  %-38s %6d', 'associations (AP-STA-CONNECTED)', conn_t)
  puts format('  %-38s %6d   %s of associations -- mostly roam-away, see Roaming',
              'ended by inactivity timeout', inact_t, pct(inact_t, conn_t).strip)
  # A timeout needs its association PLUS 300s of idle inside the same ring.
  # On an 8-minute ring only associations in the first ~3 minutes can produce
  # a visible eviction, so the raw ratio above shrinks as the ring shrinks and
  # is not comparable between runs. Dividing by the associations that actually
  # had 300s of ring left after them makes the number mean the same thing at
  # any window length. This has misread two earlier runs as improvements.
  elig_t = out_rows.sum { |r| r[:f]['celig'].to_i }
  if elig_t.positive?
    puts format('  %-38s %6d   the only ones that CAN show a timeout',
                'assocs with 300s of ring after them', elig_t)
    puts format('  %-38s %6s   compare THIS across runs, not the %s above',
                'window-normalised timeout ratio', pct(inact_t, elig_t).strip,
                pct(inact_t, conn_t).strip)
  end
  puts format('  %-38s %6d   VLAN seen before its tunnel existed',
              'unconfigured-drop events', undrop_t)
  puts
  puts 'Silent-drop state at probe time:'
  puts format('  %-38s %6d   VLAN iface up with no sdan tunnel',
              'orphaned VLAN interfaces', orph_t)
  puts format('  %-38s %6d', 'APs with at least one orphan', orph_aps)
  puts format('  %-38s %6d   fewer than 100 packets received',
              'VLAN ifaces passing ~nothing', vlow_t)
  puts format('  %-38s %6d   zero return traffic, any rx',
              'VLAN ifaces with tx_packets == 0', out_rows.sum { |r| r[:f]['vtx0'].to_i })
  # An orphan NETDEV is ambiguous -- an empty unit looks the same as a broken
  # one. An orphan MAC is not: the kernel learned that client, so the client
  # is there, and there is no tunnel for its VLAN, so its traffic is being
  # discarded at this instant.
  # ---- Association thrash --------------------------------------------------
  # Upstream of everything in the sdan sections. An ath11k peer that cannot be
  # deleted leaves the station bouncing between radios; each bounce is an
  # associate/disassociate pair, and enough of them drive the tunnel refcount
  # to a false zero. Reported before the teardown numbers because it explains
  # them -- the sdan churn is the symptom, this is the cause.
  thrash_rows = out_rows.reject { |r| r[:f]['pdto'].nil? }
  if thrash_rows.any?
    pd_t  = thrash_rows.sum { |r| r[:f]['pdto'].to_i }
    pd_ap = thrash_rows.count { |r| r[:f]['pdto'].to_i.positive? }
    puts 'Association thrash  (ath11k peer management):'
    puts format('  %-38s %6d   on %d AP(s)  %s',
                'peer-delete timeouts', pd_t, pd_ap, pct(pd_ap, thrash_rows.size).strip)
    puts format('  %-38s %6d   station never completed auth',
                'auth responses not acknowledged',
                thrash_rows.sum { |r| r[:f]['noack'].to_i })
    puts format('  %-38s %6d   hostapd evicting the other radio',
                'association prunes', thrash_rows.sum { |r| r[:f]['prune'].to_i })
    # A peer the driver cannot find leaves the station associated, authorized
    # and unusable -- everything above sdan succeeded and the radio still
    # cannot carry it. No client-side change fixes this one.
    pf_t  = thrash_rows.sum { |r| r[:f]['peerfind'].to_i }
    pf_ap = thrash_rows.count { |r| r[:f]['peerfind'].to_i.positive? }
    puts format('  %-38s %6d   on %d AP(s)  %s   <<< driver, not us',
                'peer not found for rx tid setup', pf_t, pf_ap,
                pct(pf_ap, thrash_rows.size).strip)
    puts format('  %-38s %6d   block-ack setup failed (-ENOENT)',
                '..rx tid setup failures', thrash_rows.sum { |r| r[:f]['rxtid'].to_i })

    # 1.0.38+ keeps tunnels resident and logs "kept resident" instead of
    # scheduling a release; the probe does not count those.
    sch  = thrash_rows.sum { |r| r[:f]['relgone'].to_i + r[:f]['relmove'].to_i }
    fire = thrash_rows.sum { |r| r[:f]['relfire'].to_i }
    puts format('  %-38s %6d   %d last-client-gone, %d VLAN-move',
                'tunnel releases scheduled', sch,
                thrash_rows.sum { |r| r[:f]['relgone'].to_i },
                thrash_rows.sum { |r| r[:f]['relmove'].to_i })
    if (sch + fire).positive?
      puts format('  %-38s %6d   %s   linger did its job',
                  '..cancelled, client returned in time',
                  thrash_rows.sum { |r| r[:f]['relcan'].to_i },
                  pct(thrash_rows.sum { |r| r[:f]['relcan'].to_i }, sch).strip)
      puts format('  %-38s %6d   %s   <<< tunnel actually deleted',
                  '..FIRED after the linger expired', fire, pct(fire, sch).strip)
    end
    # Not every deletion is damage. Residents roam, so an AP legitimately
    # builds a unit's tunnel when someone drifts into range and drops it when
    # they leave. A VLAN re-added within a minute of being deleted had a client
    # that was still there; one never re-added is a roamer genuinely gone.
    prem = thrash_rows.sum { |r| r[:f]['tdprem'].to_i }
    tdok = thrash_rows.sum { |r| r[:f]['tdok'].to_i }
    if (prem + tdok).positive?
      puts format('  %-38s %6d   %s   <<< a client was still there',
                  'deletions re-added within 60s', prem, pct(prem, prem + tdok).strip)
      puts format('  %-38s %6d   %s   roamer left, correct cleanup',
                  'deletions never re-added', tdok, pct(tdok, prem + tdok).strip)
    end

    worst_thrash = thrash_rows.select { |r| r[:f]['relfire'].to_i.positive? }
                              .sort_by { |r| -r[:f]['relfire'].to_i }.first(12)
    if worst_thrash.any?
      puts
      puts 'APs losing tunnels to the thrash  (fired releases, worst first):'
      puts format('  %-30s %6s %7s %7s %7s %6s  %s',
                  'AP', 'FIRED', 'SCHED', 'PEERTO', 'PRUNES', 'EVICT', 'ZONE')
      worst_thrash.each do |r|
        puts format('  %-30s %6s %7d %7s %7s %6s  %s', r[:ap].name.to_s[0, 30],
                    r[:f]['relfire'], r[:f]['relgone'].to_i + r[:f]['relmove'].to_i,
                    r[:f]['pdto'], r[:f]['prune'], r[:f]['inact'], r[:ap].zone)
      end
    end
    puts

    # ---- Driver faults by model --------------------------------------------
    # The peer fault is model-specific, not fleet-wide, and that is the single
    # most useful thing to know about it -- it turns "the APs are broken" into
    # "this model is broken", which is a firmware conversation about one SKU.
    # Worth stating in the report rather than leaving to be re-derived: the
    # -MP- name pattern that first looked like a physical cause (media panel:
    # metal enclosure, heat, long PoE run) carries no information of its own,
    # because that is simply where this model is installed.
    puts 'Driver faults by AP model:'
    puts format('  %-24s %5s %9s %7s %7s %7s %8s',
                'MODEL', 'APs', 'AFFECTED', 'EVENTS', 'PEERTO', 'STA/AP', 'BUSY2.4')
    by_model = thrash_rows.group_by { |r| r[:ap].model.to_s }
    by_model.sort_by { |_, g| -g.size }.each do |model, g|
      aff = g.count { |r| r[:f]['peerfind'].to_i.positive? }
      bus = g.filter_map { |r| busy_pct(r[:f], 'sa2', 'sb2') }
      puts format('  %-24s %5d %4d %4s %7d %7d %7.1f %8s',
                  model[0, 24], g.size, aff, pct(aff, g.size).strip,
                  g.sum { |r| r[:f]['peerfind'].to_i },
                  g.sum { |r| r[:f]['pdto'].to_i },
                  g.sum { |r| r[:f]['nsta'].to_i } / g.size.to_f,
                  bus.empty? ? '-' : format('%.0f%%', bus.sum / bus.size))
    end

    # Load-triggered, or something particular to those APs? If the affected
    # ones are simply the busiest, the fault is latent across the whole model
    # and only firmware fixes it. If they look like their peers, a bounded list
    # of APs differs somehow and that list is worth working through by hand.
    # Shares rather than per-AP counts: affected APs carry more clients, so a
    # per-AP count of weak or idle stations measures the client load, not the
    # fault. That mistake made a 1.2-vs-0.7 look meaningful when normalising
    # it gave 21% vs 19%.
    focus, fg = by_model.max_by { |_, g| g.count { |r| r[:f]['peerfind'].to_i.positive? } }
    if fg && fg.any? { |r| r[:f]['peerfind'].to_i.positive? }
      puts
      puts "Within #{focus}  (is the fault load-triggered?):"
      puts format('  %-12s %5s %8s %9s %8s %6s %9s',
                  'GROUP', 'APs', 'STA/AP', 'ASSOC/AP', 'BUSY2.4', 'WEAK', 'IDLE>60s')
      [['affected',   fg.select { |r| r[:f]['peerfind'].to_i.positive? }],
       ['unaffected', fg.reject { |r| r[:f]['peerfind'].to_i.positive? }]].each do |label, grp|
        next if grp.empty?

        sta = grp.sum { |r| r[:f]['nsta'].to_i }
        bus = grp.filter_map { |r| busy_pct(r[:f], 'sa2', 'sb2') }
        wk  = grp.sum { |r| r[:f]['sg80'].to_i + r[:f]['sg81'].to_i }
        idl = grp.sum { |r| r[:f]['iastale'].to_i }
        puts format('  %-12s %5d %8.1f %9.1f %8s %6s %9s',
                    label, grp.size, sta / grp.size.to_f,
                    grp.sum { |r| r[:f]['conn'].to_i } / grp.size.to_f,
                    bus.empty? ? '-' : format('%.0f%%', bus.sum / bus.size),
                    sta.zero? ? '-' : pct(wk, sta).strip,
                    sta.zero? ? '-' : pct(idl, sta).strip)
      end
    end
    puts
  end

  # ---- Cross-VLAN teardown -------------------------------------------------
  # Distinct from everything above: not a client that failed to get set up, but
  # a WORKING client on another VLAN losing its tunnel because somebody else
  # associated. The add-port rollback calls release_tunnel_for without checking
  # the per-client refcount, and that function is keyed on manager name, so one
  # call removes the tunnel out from under every client on the VLAN.
  chain_rows = out_rows.reject { |r| r[:f]['apfail'].nil? }
  if chain_rows.any?
    apf_t   = chain_rows.sum { |r| r[:f]['apfail'].to_i }
    apf_aps = chain_rows.count { |r| r[:f]['apfail'].to_i.positive? }
    del_t   = chain_rows.sum { |r| r[:f]['tundel'].to_i }
    hit     = chain_rows.select { |r| r[:f]['chain'].to_i.positive? }
    puts 'Cross-VLAN teardown  (add-port rollback deleting a live VLAN tunnel):'
    puts format('  %-38s %6d   on %d AP(s)', 'add-port failures logged', apf_t, apf_aps)
    puts format('  %-38s %6d   kernel confirmed a tunnel went away',
                'sdan tunnel deletions logged', del_t)
    puts format('  %-38s %6d   %s   <<< same VLAN: failure THEN drops',
                'APs where the full chain fired', hit.size,
                pct(hit.size, chain_rows.size).strip)
    if hit.any?
      puts
      puts 'APs with a confirmed cross-VLAN teardown:'
      puts format('  %-30s %6s %8s %6s  %s',
                  'AP', 'FAILS', 'DELETES', 'VLANS', 'AFFECTED VLANs')
      hit.sort_by { |r| -r[:f]['chain'].to_i }.first(12).each do |r|
        puts format('  %-30s %6s %8s %6s  %s', r[:ap].name.to_s[0, 30],
                    r[:f]['apfail'], r[:f]['tundel'], r[:f]['chain'],
                    r[:f]['chainvl'].to_s[0, 34])
      end
    elsif apf_t.zero?
      # Said explicitly because a zero here is easy to over-read: the ring is
      # short and this fires on association, not continuously.
      puts format('  %-38s %6s   in a %s ring -- absence is weak evidence',
                  'no add-port failure seen at all', '', fmt_dur(med_span))
    end
    puts
  end

  mac_rows = out_rows.reject { |r| r[:f]['macn'].nil? }
  if mac_rows.any?
    macn_t  = mac_rows.sum { |r| r[:f]['macn'].to_i }
    macdr_t = mac_rows.sum { |r| r[:f]['macdrop'].to_i }
    puts format('  %-38s %6d   sdan MAC table, all APs', 'clients learned by sdan', macn_t)
    puts format('  %-38s %6d   %s   <<< dropped RIGHT NOW',
                '..on a VLAN with no tunnel', macdr_t, pct(macdr_t, macn_t).strip)
    worst_drop = mac_rows.select { |r| r[:f]['macdrop'].to_i.positive? }
                         .sort_by { |r| -r[:f]['macdrop'].to_i }.first(10)
    if worst_drop.any?
      puts
      puts 'APs dropping learned clients  (MAC on a VLAN with no tunnel):'
      puts format('  %-30s %9s %7s %7s %s', 'AP', 'DROP/MACS', 'TUNNELS', 'ORPHANS', 'ZONE')
      worst_drop.each do |r|
        puts format('  %-30s %4d/%-4d %7s %7s %s', r[:ap].name.to_s[0, 30],
                    r[:f]['macdrop'].to_i, r[:f]['macn'].to_i,
                    r[:f]['tuns'], r[:f]['orph'], r[:ap].zone)
      end
    end
  end
  puts

  # ---- Connected clients now -----------------------------------------------
  # Everything above is reconstructed from the log. This is a direct read of
  # the stations the driver is holding right now, and it is the only thing
  # here that can separate "roamed because coverage ran out" from "roamed for
  # some other reason": the first shows up as clients sitting below -70 dBm.
  SIG_KEYS = %w[sg60 sg70 sg80 sg81].freeze
  sta_rows = out_rows.reject { |r| r[:f]['nsta'].nil? }
  if sta_rows.any?
    nsta_t = sta_rows.sum { |r| r[:f]['nsta'].to_i }
    sg     = SIG_KEYS.map { |k| sta_rows.sum { |r| r[:f][k].to_i } }
    ct     = %w[ct60 ct300 ctlong].map { |k| sta_rows.sum { |r| r[:f][k].to_i } }
    sig_n  = sg.sum
    ct_n   = ct.sum
    puts "Connected clients at probe time  (#{sta_rows.size} AP(s) reporting):"
    puts format('  %-38s %6d', 'stations associated', nsta_t)
    # The buckets must partition the stations exactly. Anything else means the
    # station dump parse is matching the wrong lines, which is not obvious from
    # the percentages alone -- they still sum to 100% of a wrong denominator.
    if sig_n != nsta_t
      puts format('  %-38s %6s   signal reads %d vs %d stations -- PARSE BUG',
                  '!! bucket total does not match', '', sig_n, nsta_t)
    end
    [['signal >= -60 dBm', sg[0], 'strong'],
     ['signal -60 to -70 dBm', sg[1], ''],
     ['signal -70 to -80 dBm', sg[2], 'marginal'],
     ['signal < -80 dBm', sg[3], 'cell edge']].each do |label, v, note|
      puts format('  %-38s %6d   %s%s', label, v, pct(v, sig_n).strip,
                  note.empty? ? '' : "   #{note}")
    end
    [['session < 60s', ct[0], 'churn'],
     ['session 60s to 5m', ct[1], ''],
     ['session > 5m', ct[2], 'settled']].each do |label, v, note|
      puts format('  %-38s %6d   %s%s', label, v, pct(v, ct_n).strip,
                  note.empty? ? '' : "   #{note}")
    end
    # The dataplane question, asked of the only population that can answer it.
    # Signal and airtime cannot explain a station that is settled, well within
    # range, and has moved nothing -- so if this is non-trivial the failure is
    # below the radio, and if it is near zero the radio is where to keep
    # looking. Deliberately silent about WHICH part of the dataplane failed.
    # hostapd decides idleness from the DRIVER's inactive_msec, so a station
    # whose peer the driver has lost reports as permanently idle and gets
    # deauthed at ap_max_inactivity even while it is transmitting. A large
    # stale bucket next to a large peer-not-found count is that mechanism.
    ia = %w[iact iamid iastale].map { |k| sta_rows.sum { |r| r[:f][k].to_i } }
    if ia.sum.positive?
      [['idle < 1s', ia[0], ''],
       ['idle 1s to 60s', ia[1], ''],
       ['idle > 60s', ia[2], 'deauth candidates']].each do |label, v, note|
        puts format('  %-38s %6d   %s%s', label, v, pct(v, ia.sum).strip,
                    note.empty? ? '' : "   #{note}")
      end
    end
    elig_st = sta_rows.sum { |r| r[:f]['stelig'].to_i }
    sil_st  = sta_rows.sum { |r| r[:f]['stsil'].to_i }
    if elig_st.positive?
      puts format('  %-38s %6d   settled >5m AND >= -70 dBm',
                  'stations that should be working', elig_st)
      puts format('  %-38s %6d   %s   <<< no RF explanation for these',
                  '..passed under 10 KB', sil_st, pct(sil_st, elig_st).strip)
    end
    # Cross-reference, not a standalone list: an AP with a high eviction ratio
    # AND a high weak share is a coverage hole, while the same eviction ratio
    # with every client at -50 dBm is something else entirely.
    edge = sta_rows.map { |r|
             n = SIG_KEYS.sum { |k| r[:f][k].to_i }
             [r, n, r[:f]['sg80'].to_i + r[:f]['sg81'].to_i]
           }.select { |_, n, w| n >= 3 && w.positive? }
            .sort_by { |_, n, w| -(100.0 * w / n) }.first(10)
    if edge.any?
      puts
      puts 'Worst APs by clients below -70 dBm  (>= 3 stations):'
      puts format('  %-30s %-9s %-6s %-8s %s',
                  'AP', 'WEAK/STA', 'RATIO', 'EVICT', 'ZONE')
      edge.each do |r, n, w|
        puts format('  %-30s %4d/%-4d %-6s %-8s %s', r[:ap].name.to_s[0, 30], w, n,
                    pct(w, n).strip, r[:f]['inact'], r[:ap].zone)
      end
    end
    puts
  end

  # ---- Radio configuration as applied --------------------------------------
  # The rXg knows what it pushed; this is what the radios ended up with after
  # regulatory and hardware clamping. It matters for reading every other table
  # in this report: if the fleet is uniform, a per-building difference in
  # outcomes is the RF environment, and if it is not, the difference may just
  # be the config -- which would make the odd building an accidental control.
  rf_rows = out_rows.reject { |r| r[:f]['txp2'].nil? }
  if rf_rows.any?
    puts "Radio configuration as applied  (#{rf_rows.size} AP(s) reporting):"
    [['2.4 GHz', 'ch2', 'bw2', 'txp2'], ['5 GHz', 'ch5', 'bw5', 'txp5']].each do |band, ck, wk, pk|
      shown = 0
      rf_rows.group_by { |r| [r[:f][ck], r[:f][wk], r[:f][pk]] }
             .reject { |(ch, _, _), _| ch.to_i.zero? }
             .sort_by { |_, g| -g.size }.each do |(ch, bw, tp), g|
        puts format('  %-9s ch%-5s %3s MHz  %6s dBm  %4d AP(s)%s',
                    shown.zero? ? band : '', ch, bw, tp, g.size,
                    g.size < 5 ? "   e.g. #{g.first[:ap].name}" : '')
        shown += 1
      end
      puts format('  %-9s (no radio reported)', band) if shown.zero?
    end
    pairs = rf_rows.map { |r| "#{r[:f]['txp2']}/#{r[:f]['txp5']}" }
    puts format('  %-38s %6d', 'distinct 2.4/5 power pairs', pairs.uniq.size)
    if pairs.uniq.size > 1
      puts '  TX power 2.4/5 by building:'
      rf_rows.group_by { |r| building_of(r[:ap].name) }.sort.each do |b, g|
        mix = g.group_by { |r| "#{r[:f]['txp2']}/#{r[:f]['txp5']}" }
               .sort_by { |_, x| -x.size }
               .map { |k, x| "#{k} x#{x.size}" }.first(3).join('   ')
        puts format('    %-20s %s', b, mix)
      end
    end
    puts
  end

  # ---- Channel utilisation -------------------------------------------------
  # The third candidate after power (uniform) and signal (uncorrelated). Read
  # per band because they contend separately: 2.4 GHz has 3 channels for the
  # whole property, 5 GHz has 4 at 40 MHz, and only the 5 GHz figure responds
  # to narrowing the channel width.
  UTIL_BANDS = [['2.4 GHz', 'sa2', 'sb2'], ['5 GHz', 'sa5', 'sb5']].freeze
  util_rows = out_rows.select { |r| r[:f]['sa2'] || r[:f]['sa5'] }
  if util_rows.any?
    puts "Channel utilisation  (busy/active, 2s sample, #{util_rows.size} AP(s)):"
    puts format('  %-9s %6s %7s %7s %11s %11s',
                'BAND', 'APs', 'MEDIAN', 'P90', '>50% BUSY', '>70% BUSY')
    UTIL_BANDS.each do |band, ak, bk|
      vals = util_rows.filter_map { |r| busy_pct(r[:f], ak, bk) }.sort
      if vals.empty?
        puts format('  %-9s %6s   no usable sample', band, '-')
        next
      end
      p90 = vals[[(vals.size * 0.9).floor, vals.size - 1].min]
      puts format('  %-9s %6d %6.0f%% %6.0f%% %11d %11d', band, vals.size,
                  vals[vals.size / 2], p90,
                  vals.count { |v| v > 50 }, vals.count { |v| v > 70 })
    end
    # Paired with the eviction count so a busy radio can be told apart from a
    # busy radio that is actually costing anyone anything.
    hot = util_rows.filter_map { |r|
            v = busy_pct(r[:f], 'sa5', 'sb5')
            u = busy_pct(r[:f], 'sa2', 'sb2')
            best = [v, u].compact.max
            best && best > 30 ? [r, u, v] : nil
          }.sort_by { |_, u, v| -[u, v].compact.max }.first(10)
    if hot.any?
      puts
      puts 'Busiest radios  (either band over 30%):'
      puts format('  %-30s %8s %8s %7s %s', 'AP', 'BUSY2.4', 'BUSY5', 'EVICT', 'ZONE')
      hot.each do |r, u, v|
        puts format('  %-30s %7s %8s %7s %s', r[:ap].name.to_s[0, 30],
                    u ? format('%.0f%%', u) : '-', v ? format('%.0f%%', v) : '-',
                    r[:f]['inact'], r[:ap].zone)
      end
    end
    puts
  end

  # ---- Roaming -------------------------------------------------------------
  # Intra-AP is band steering: the same MAC associating on both radios of one
  # AP. Inter-AP cannot be seen from any single AP -- the client just stops
  # appearing in one log and starts in another -- so it is reconstructed here
  # by correlating the MAC lists from all APs. Each AP reports at most 60
  # MACs, so on a very busy AP this undercounts rather than overcounts.
  def mac_list(field)
    field.to_s.split(',').reject { |v| v.empty? || v == '-' }
  end

  reassoc_t = out_rows.sum { |r| r[:f]['reassoc'].to_i }
  band_t    = out_rows.sum { |r| r[:f]['bandroam'].to_i }
  aps_by_mac = Hash.new { |h, k| h[k] = [] }
  out_rows.each { |r| mac_list(r[:f]['macs']).each { |m| aps_by_mac[m] << r[:ap].name } }
  seen_macs = aps_by_mac.size
  multi     = aps_by_mac.select { |_, aps| aps.uniq.size > 1 }
  # Band steering is per-AP, so it must be measured against per-AP client
  # sightings. Dividing a sum of per-AP counts by fleet-wide DISTINCT clients
  # double-counts the numerator and can exceed 100%; nmac is each AP's own
  # uncapped client count, so the two sides agree.
  sightings = out_rows.sum { |r| r[:f]['nmac'].to_i }

  puts 'Roaming:'
  puts format('  %-38s %6d   %s of associations', 're-associations (same AP)', reassoc_t, pct(reassoc_t, conn_t).strip)
  puts format('  %-38s %6d   %s of per-AP client sightings -- INTRA-AP',
              'clients on both radios of one AP', band_t, pct(band_t, sightings).strip)
  puts format('  %-38s %6d   %s of clients seen -- INTER-AP roam, in %s',
              'clients seen on 2+ APs', multi.size, pct(multi.size, seen_macs).strip,
              med_span.positive? ? fmt_dur(med_span) : 'an unknown window')
  puts format('  %-38s %6d   (%d per-AP sightings)', 'distinct clients observed', seen_macs, sightings)

  if multi.any?
    hist = Hash.new(0)
    multi.each_value do |aps|
      n = aps.uniq.size
      hist[n >= 6 ? 'f. 6+ APs' : format('%s. %d APs', ('a'.ord + n - 2).chr, n)] += 1
    end
    puts '  APs visited per roaming client:'
    hist.sort_by { |k, _| k }.each { |k, v| puts format('    %-34s %6d', k, v) }

    top = multi.sort_by { |_, aps| -aps.uniq.size }.first(5)
    puts '  most mobile clients:'
    top.each do |mac, aps|
      n = aps.uniq.size
      rate = med_span.positive? ? format('%.1f APs/h', n / (med_span / 3600.0)) : 'rate unknown'
      puts format('    %-19s %2d APs  %-12s e.g. %s', mac, n, rate, aps.uniq.first.to_s[0, 28])
    end
  end

  # Cause or symptom? If the eviction rate climbs with the number of APs a
  # client visited, roaming and failure are one population -- clients that
  # cannot settle anywhere long enough to get service. If the rate is flat
  # across buckets, roaming is RF churn and the evictions have another cause,
  # and treating the roaming would not help the complaints.
  evicted = {}
  out_rows.each { |r| mac_list(r[:f]['emacs']).each { |m| evicted[m] = true } }
  if aps_by_mac.any?
    buck = Hash.new { |h, k| h[k] = [0, 0] }
    aps_by_mac.each do |mac, aps|
      n = aps.uniq.size
      key = n >= 6 ? 'f. 6+ APs' : format('%s. %d AP%s', ('a'.ord + n - 1).chr, n, n == 1 ? ' ' : 's')
      buck[key][0] += 1
      buck[key][1] += 1 if evicted[mac]
    end
    puts '  eviction rate by APs visited (roaming the cause, or the symptom?):'
    puts format('    %-14s %8s %9s %s', 'APs VISITED', 'CLIENTS', 'EVICTED', 'RATE')
    buck.sort_by { |k, _| k }.each do |k, (tot, ev)|
      puts format('    %-14s %8d %9d %s', k, tot, ev, pct(ev, tot).strip)
    end
  end

  steer = out_rows.select { |r| r[:f]['bandroam'].to_i > 0 && r[:f]['conn'].to_i >= 5 }
                  .sort_by { |r| -(100.0 * r[:f]['bandroam'].to_i / r[:f]['conn'].to_i) }.first(8)
  if steer.any?
    puts '  worst APs by band-steer rate (>= 5 associations):'
    puts format('    %-30s %-10s %s', 'AP', 'STEER/CONN', 'RATIO')
    steer.each do |r|
      b = r[:f]['bandroam'].to_i
      c = r[:f]['conn'].to_i
      puts format('    %-30s %4d/%-5d %s', r[:ap].name.to_s[0, 30], b, c, pct(b, c).strip)
    end
  end
  puts

  # Denominator is window-normalised (associations with 300s of ring left),
  # so these buckets mean the same thing on an 8-minute ring as on an hour.
  cross_tab('By inactivity-timeout ratio  (window-normalised; largely roam-away):', 'RATIO',
            out_rows.group_by { |r|
              c = r[:f]['celig'].to_i
              next 'f. ring too short to judge' if c.zero?
              share = 100.0 * r[:f]['inact'].to_i / c
              case share
              when 0       then 'a. none timed out'
              when 0...10  then 'b. under 10%'
              when 10...25 then 'c. 10-25%'
              when 25...50 then 'd. 25-50%'
              else              'e. 50%+ of associations'
              end
            }.sort_by { |k, _| k }, width: 30)

  # The one per-AP list in this report, and deliberately bounded. A ratio is
  # meaningless on an AP with two associations, hence the floor.
  worst = out_rows.select { |r| r[:f]['celig'].to_i >= 5 && r[:f]['inact'].to_i.positive? }
                  .sort_by { |r| -(100.0 * r[:f]['inact'].to_i / r[:f]['celig'].to_i) }
                  .first(10)
  if worst.any?
    puts 'Worst APs by inactivity-timeout ratio  (>= 5 window-normalised assocs):'
    puts format('  %-30s %-11s %-6s %-7s %-8s %s',
                'AP', 'EVICT/ELIG', 'RATIO', 'WEAK', 'ORPHANS', 'ZONE')
    worst.each do |r|
      c = r[:f]['celig'].to_i
      i = r[:f]['inact'].to_i
      n = SIG_KEYS.sum { |k| r[:f][k].to_i }
      w = r[:f]['sg80'].to_i + r[:f]['sg81'].to_i
      puts format('  %-30s %4d/%-6d %-6s %-7s %-8s %s', r[:ap].name.to_s[0, 30], i, c,
                  pct(i, c).strip, n.zero? ? '-' : "#{w}/#{n}",
                  r[:f]['orph'], r[:ap].zone)
    end
    puts
  end

  # RATIO is on the window-normalised denominator (ELIG); WEAK is the share of
  # currently-associated stations below -70 dBm. A building that is worst on
  # both is a coverage problem; worst on RATIO alone is not.
  puts 'By building  (parsed from AP name):'
  puts format('  %-18s %5s %6s %6s %6s %6s %5s %6s %6s %5s',
              'BUILDING', 'APs', 'CONN', 'ELIG', 'EVICT', 'RATIO', 'WEAK',
              'BUSY2', 'BUSY5', 'ORPH')
  out_rows.group_by { |r| building_of(r[:ap].name) }
          .sort_by { |b, g|
            el = g.sum { |r| r[:f]['celig'].to_i }
            [-(el.zero? ? 0 : 100.0 * g.sum { |r| r[:f]['inact'].to_i } / el), b]
          }.each do |b, g|
    c  = g.sum { |r| r[:f]['conn'].to_i }
    el = g.sum { |r| r[:f]['celig'].to_i }
    e  = g.sum { |r| r[:f]['inact'].to_i }
    o  = g.sum { |r| r[:f]['orph'].to_i }
    n  = g.sum { |r| SIG_KEYS.sum { |k| r[:f][k].to_i } }
    w  = g.sum { |r| r[:f]['sg80'].to_i + r[:f]['sg81'].to_i }
    # Median rather than mean: one radio pinned at 100% would drag a building
    # average far more than it drags that building's residents' experience.
    bu = UTIL_BANDS.map { |_, ak, bk|
      v = g.filter_map { |r| busy_pct(r[:f], ak, bk) }.sort
      v.empty? ? '-' : format('%.0f%%', v[v.size / 2])
    }
    puts format('  %-18s %5d %6d %6d %6d %6s %5s %6s %6s %5d', b, g.size, c, el, e,
                pct(e, el).strip, n.zero? ? '-' : pct(w, n).strip, bu[0], bu[1], o)
  end
  puts
end

# ---- RADIUS enrichment (opt-in) -------------------------------------------
# The AP log ring is short and rotates; RadiusServerLog is server-side and
# retained a month, so it can see the weekend the APs have long forgotten.
# Access-Accept tells us what the rXg promised each client -- including the
# VLAN it assigned -- and the accounting Stop tells us what the session
# actually moved. A Stop carrying almost no octets is "got on, got nothing"
# measured authoritatively, with no ring-buffer truncation.
#
# Off by default and guarded by a row count first: this table is large at a
# 1000-client property and an unbounded scan on a live rXg is not a diagnostic.
if ENV['RADIUS'].present?
  hours  = (ENV['RADIUS_HOURS'] || 2).to_f
  cap    = (ENV['RADIUS_MAX'] || 200_000).to_i
  sample = (ENV['RADIUS_SAMPLE'] || 2000).to_i
  since  = hours.hours.ago
  # created_at, not event_timestamp: the retention purger keys on created_at,
  # so that is the column most likely to carry an index.
  scope  = RadiusServerLog.where('created_at > ?', since)
  n      = scope.count

  puts "RADIUS  (last #{hours}h, #{n} row(s)):"
  if n > cap
    puts format('  %-38s %6d   over RADIUS_MAX=%d; narrow with RADIUS_HOURS=', 'rows in window', n, cap)
  elsif n.zero?
    puts '  no rows in the window -- is RADIUS logging enabled?'
  else
    scope.where.not(response_type: nil).group(:response_type).count
         .sort_by { |_, v| -v }.each { |t, v| puts format('  %-38s %6d', t, v) }
    rej = scope.where(response_type: 'Access-Reject').count
    acc = scope.where(response_type: 'Access-Accept').count
    puts format('  %-38s %6s   of all authentications', 'reject rate', pct(rej, rej + acc).strip)
    puts

    # What VLAN did the Access-Accept actually assign? Regex rather than
    # YAML.load: cheaper, and we never deserialise stored attribute blobs.
    yamls = scope.where(response_type: 'Access-Accept').limit(sample).pluck(:response_attributes_yaml)
    if yamls.any?
      vl = Hash.new(0)
      yamls.each do |y|
        m = y.to_s[/Tunnel-Private-Group-Id[^0-9]{0,40}(\d{1,4})/, 1]
        vl[m || '(no VLAN in Accept)'] += 1
      end
      no_vlan = vl['(no VLAN in Accept)']
      puts format('Access-Accept VLAN assignment  (sample of %d):', yamls.size)
      puts format('  %-38s %6d   %s', 'Accepts carrying no VLAN', no_vlan, pct(no_vlan, yamls.size).strip)
      puts format('  %-38s %6d', 'distinct VLANs assigned', vl.keys.count { |k| k != '(no VLAN in Accept)' })
      puts
    end

    # The AP's belief vs the rXg's instruction, from independent sources: the
    # kernel MAC table says which VLAN each client was learned on, and the
    # Access-Accept says which VLAN was assigned. They cannot disagree unless
    # something resolved the wrong one, so a mismatch here is the bug observed
    # rather than reasoned from the code. Same MAC, two answers.
    ap_vlan = {}
    rows.each do |r|
      r[:f]['smv'].to_s.split(',').each do |pair|
        mac, vlan = pair.split(':', 7).then { |f| [f[0, 6].join(':'), f[6]] }
        next if mac.to_s.empty? || vlan.to_s.empty?

        ap_vlan[norm_mac(mac)] = [vlan.to_i, r[:ap].name, mac.downcase]
      end
    end
    if ap_vlan.any?
      # Newest Accept wins: a client re-authenticated onto a different VLAN is
      # a reassignment, not a disagreement, and the AP should hold the latest.
      # Bounded on purpose. response_attributes_yaml is a blob column and an
      # unqualified pluck over a busy window would drag the whole table into
      # this process; restricting to the MACs the APs actually reported keeps
      # it an indexed IN over ~1k values. If the stored MAC format ever stops
      # matching sdan's lowercase-colon form this returns nothing, which shows
      # up as a zero in the 'to compare' line below rather than as a silent
      # all-clear.
      rad_vlan = {}
      scope.where(response_type: 'Access-Accept')
           .where(calling_station_id_mac: ap_vlan.values.map { |v| v[2] }.uniq)
           .order(:created_at).limit(20_000)
           .pluck(:calling_station_id_mac, :response_attributes_yaml)
           .each do |mac, y|
        v = y.to_s[/Tunnel-Private-Group-Id[^0-9]{0,40}(\d{1,4})/, 1]
        rad_vlan[norm_mac(mac)] = v.to_i if v
      end
      vm_both = ap_vlan.keys & rad_vlan.keys
      vm_bad  = vm_both.reject { |m| ap_vlan[m][0] == rad_vlan[m] }
      puts 'AP VLAN vs RADIUS VLAN  (sdan MAC table vs Access-Accept):'
      puts format('  %-38s %6d   of %d AP-learned MAC(s)',
                  'MACs with a RADIUS Accept to compare', vm_both.size, ap_vlan.size)
      puts format('  %-38s %6d   %s   <<< wrong VLAN applied',
                  'MACs where the two disagree', vm_bad.size,
                  pct(vm_bad.size, vm_both.size).strip)
      if vm_bad.any?
        vm_who = accounts_for_macs(vm_bad.first(15))
        puts format('    %-19s %7s %8s  %-22s %s',
                    'MAC', 'AP SAYS', 'RADIUS', 'ACCOUNT', 'ON AP')
        vm_bad.first(15).each do |m|
          puts format('    %-19s %7d %8d  %-22s %s', ap_vlan[m][2],
                      ap_vlan[m][0], rad_vlan[m],
                      account_label(vm_who[m]).to_s[0, 22], ap_vlan[m][1].to_s[0, 26])
        end
      end
      puts
    end

    # Rejections. 500 rejects is a completely different problem depending on
    # whether it is 20 devices retrying or 400 residents locked out, so the
    # statistic that matters is distinct devices -- and specifically devices
    # that were rejected and never accepted in the window. A device that was
    # rejected once and then got in is a typo; one that never got in is a
    # resident who cannot use the network.
    if rej.positive?
      rej_scope = scope.where(response_type: 'Access-Reject').where.not(calling_station_id_mac: [nil, ''])
      rej_macs  = rej_scope.distinct.pluck(:calling_station_id_mac)
      acc_macs  = scope.where(response_type: 'Access-Accept')
                       .where.not(calling_station_id_mac: [nil, '']).distinct.pluck(:calling_station_id_mac)
      locked    = rej_macs - acc_macs
      seen      = (rej_macs | acc_macs).size

      puts 'Rejections:'
      puts format('  %-38s %6d', 'reject events', rej)
      puts format('  %-38s %6d   %.1f attempts each', 'distinct devices rejected',
                  rej_macs.size, rej_macs.size.zero? ? 0.0 : rej.to_f / rej_macs.size)
      puts format('  %-38s %6d   %s of devices seen this window',
                  'devices NEVER authenticated', locked.size, pct(locked.size, seen).strip)

      rej_scope.group(:ssid).count.sort_by { |_, v| -v }.first(6).each do |ssid, v|
        puts format('    %-36s %6d', (ssid.presence || '(no ssid)').to_s[0, 36], v)
      end

      top      = rej_scope.group(:calling_station_id_mac).count.sort_by { |_, v| -v }.first(8)
      interest = (top.map(&:first) + locked.first(15)).uniq
      # user_name is still worth showing -- it is the credential the device
      # actually presented, which is the useful half of a rejection.
      unm  = interest.any? ?
             rej_scope.where(calling_station_id_mac: interest)
                      .distinct.pluck(:calling_station_id_mac, :user_name)
                      .each_with_object({}) { |(m, u), h| h[m] ||= u if u.present? } : {}
      who  = interest.any? ? accounts_for_macs(interest) : {}

      if top.any?
        puts '  most persistent devices (one device retrying is not an outage):'
        puts format('    %-19s %8s  %-24s %s', 'MAC', 'ATTEMPTS', 'ACCOUNT', 'USERNAME TRIED')
        top.each do |mac, v|
          puts format('    %-19s %8d  %-24s %s', mac, v,
                      account_label(who[mac]).to_s[0, 24], (unm[mac] || '(none)').to_s[0, 28])
        end
      end

      if locked.any?
        puts '  devices that never authenticated (these are locked-out residents):'
        puts format('    %-19s %-24s %s', 'MAC', 'ACCOUNT', 'USERNAME TRIED')
        locked.first(15).each do |mac|
          puts format('    %-19s %-24s %s', mac,
                      account_label(who[mac]).to_s[0, 24], (unm[mac] || '(none)').to_s[0, 28])
        end
        puts format('    ... and %d more', locked.size - 15) if locked.size > 15
      end

      # Do rejects cluster anywhere? Same building lens as the AP table.
      by_ip = AccessPoint.where.not(ip: [nil, '']).pluck(:ip, :name).to_h
      bld   = Hash.new(0)
      rej_scope.group(:nas_ip_address).count.each do |ip, v|
        bld[building_of(by_ip[ip] || '')] += v
      end
      if bld.size > 1
        puts '  by building:'
        bld.sort_by { |_, v| -v }.first(8).each { |b, v| puts format('    %-36s %6d', b, v) }
      end
      puts format('  => %d device(s) could not authenticate at all in the last %.1fh.',
                  locked.size, hours)
      puts
    end

    stops = scope.where(status_type: 'Stop')
    st = stops.count
    if st.positive?
      quiet = stops.where('coalesce(acct_input_octets,0) + coalesce(acct_output_octets,0) <= 2048').count
      puts 'Session outcomes (accounting Stop):'
      puts format('  %-38s %6d', 'sessions ended', st)
      puts format('  %-38s %6d   %s -- moved under 2 KB total', 'sessions that passed ~nothing', quiet, pct(quiet, st).strip)
      stops.group(:acct_terminate_cause).count.sort_by { |_, v| -v }.first(6).each do |cause, v|
        puts format('    %-36s %6d', cause || '(none)', v)
      end
      puts

      # Per-AP, joined through nas_ip_address -> access_points.ip.
      by_ip  = AccessPoint.where.not(ip: [nil, '']).pluck(:ip, :name).to_h
      tot_ip = stops.group(:nas_ip_address).count
      bad_ip = stops.where('coalesce(acct_input_octets,0) + coalesce(acct_output_octets,0) <= 2048')
                    .group(:nas_ip_address).count
      worst_ip = bad_ip.select { |ip, v| v >= 3 && tot_ip[ip].to_i >= 5 }
                       .sort_by { |ip, v| -(100.0 * v / tot_ip[ip]) }.first(10)
      if worst_ip.any?
        puts 'Worst APs by quiet-session ratio  (>= 5 sessions, >= 3 quiet):'
        puts format('  %-30s %-12s %-7s %s', 'AP (via nas_ip)', 'QUIET/TOTAL', 'RATIO', 'BUILDING')
        worst_ip.each do |ip, v|
          nm = by_ip[ip] || ip
          puts format('  %-30s %5d/%-6d %-7s %s', nm.to_s[0, 30], v, tot_ip[ip],
                      pct(v, tot_ip[ip]).strip, building_of(nm))
        end
        puts
      end
    end
  end
end

# Distinct probe signatures, deduped. 226 APs emitting one of three identical
# lines is three facts, not 226.
# Memory-pressure correlation. If the broken APs cluster in the low-headroom
# buckets, the once-only startup fetch is being killed by OOM rather than
# failing on the network -- a different fix entirely.
def mem_bucket(kb)
  return '(not reported)' if kb.nil? || kb.to_s.empty?
  mb = kb.to_i / 1024
  case mb
  when 0...16   then 'a. < 16 MB free'
  when 16...32  then 'b. 16-32 MB free'
  when 32...64  then 'c. 32-64 MB free'
  when 64...128 then 'd. 64-128 MB free'
  else               'e. >= 128 MB free'
  end
end

mem_rows = rows.reject { |r| r[:f]['memavail'].nil? }
if mem_rows.any?
  cross_tab('By free memory  (MemAvailable at probe time):', 'HEADROOM',
            mem_rows.group_by { |r| mem_bucket(r[:f]['memavail']) }.sort_by { |k, _| k },
            width: 20)

  # AP uptime, distinct from client age: the auto-recovery job reboots an AP
  # the rXg believes offline but which answers ping, and PoE-cycles one that
  # does not. A fleet being churned by false-offline recoveries shows up here
  # as a cluster of short uptimes -- and every reboot scatters that AP's
  # clients onto neighbours, which is a roaming source in its own right.
  up_rows = rows.reject { |r| r[:f]['uptime'].nil? || r[:f]['uptime'].to_i.zero? }
  if up_rows.any?
    cross_tab('By AP uptime  (has the recovery job been rebooting these?):', 'UPTIME',
              up_rows.group_by { |r|
                case r[:f]['uptime'].to_i
                when 0...3600      then 'a. < 1 hour'
                when 3600...21600  then 'b. 1-6 hours'
                when 21600...86400 then 'c. 6-24 hours'
                when 86400...604800 then 'd. 1-7 days'
                else                    'e. > 7 days'
                end
              }.sort_by { |k, _| k }, width: 20)
  end

  age_rows = rows.reject { |r| r[:f]['clientage'].nil? || r[:f]['clientage'].to_i.zero? }
  if age_rows.any?
    cross_tab('By rg_wss_client process age  (when did this AP last re-fetch?):',
              'CLIENT AGE', age_rows.group_by { |r|
                secs = r[:f]['clientage'].to_i
                case secs
                when 0...3600      then 'a. < 1 hour'
                when 3600...86400  then 'b. 1-24 hours'
                when 86400...604800 then 'c. 1-7 days'
                else                    'd. > 7 days'
                end
              }.sort_by { |k, _| k }, width: 20)
  end

  # ---- Crash records -------------------------------------------------------
  # These outlive the reboot. A panic with panic_on_oom set takes the whole AP
  # down rather than killing one daemon, so each of these is an outage for
  # every resident on that AP, and none of them appear in the logread-based
  # OOM count below -- the reboot erased that evidence before we asked.
  crash_rows = rows.reject { |r| r[:f]['crashn'].nil? }
  if crash_rows.any?
    with_crash = crash_rows.select { |r| r[:f]['crashn'].to_i.positive? }
    puts 'Crash records on disk  (pstore/crashlog -- survives the reboot):'
    puts format('  %-40s %4d   of %d AP(s) probed',
                'APs carrying a crash record', with_crash.size, crash_rows.size)
    puts format('  %-40s %4d   whole-AP reboot, not a killed daemon',
                '..record mentions kernel panic',
                crash_rows.count { |r| r[:f]['crashpanic'].to_i.positive? })
    puts format('  %-40s %4d   the logread OOM count cannot see these',
                '..record mentions out of memory',
                crash_rows.count { |r| r[:f]['crashoom'].to_i.positive? })
    # Printed in full and unpadded: which path answered is the thing we did not
    # know, and truncating it to a column width would hide the answer.
    paths = crash_rows.map { |r| r[:f]['crashpath'] }.reject { |v| v.nil? || v == '-' }
    paths.tally.sort_by { |_, n| -n }.first(3).each do |path, n|
      puts format('    %d AP(s) at %s', n, path)
    end
    # The distinction that matters. A boot-copied record means this AP is
    # running on a boot that a crash caused, so its uptime IS the time since
    # the crash. A pstore-only record is history of unbounded age.
    booted = with_crash.select { |r| r[:f]['crashsrc'] == 'boot' }
    pstore_only = with_crash.select { |r| r[:f]['crashsrc'] == 'pstore' }
    puts format('  %-40s %4d   uptime = time since that crash',
                'current boot FOLLOWED a crash', booted.size)
    puts format('  %-40s %4d   older history, age not bounded',
                'pstore record only', pstore_only.size)
    if booted.any?
      ups = booted.map { |r| r[:f]['uptime'].to_i }.reject(&:zero?)
      if ups.any?
        puts format('  %-40s %6s   most recent %s, oldest %s',
                    '  ..median time since crash', fmt_dur(median(ups)),
                    fmt_dur(ups.min), fmt_dur(ups.max))
      end
      # Sorted by uptime ascending: the top of this list crashed most
      # recently, and is where a live reproduction is still possible.
      puts
      puts 'Most recent crash-caused boots  (shortest uptime first):'
      puts format('  %-30s %10s %6s %6s %s',
                  'AP', 'SINCE CRASH', 'OOM?', 'RSS', 'ZONE')
      booted.reject { |r| r[:f]['uptime'].to_i.zero? }
            .sort_by { |r| r[:f]['uptime'].to_i }.first(10).each do |r|
        rss = r[:f]['clirss'].to_i
        puts format('  %-30s %10s %6s %5s %s', r[:ap].name.to_s[0, 30],
                    fmt_dur(r[:f]['uptime'].to_i),
                    r[:f]['crashoom'].to_i.positive? ? 'yes' : '-',
                    rss.zero? ? '-' : "#{rss / 1024}M", r[:ap].zone)
      end
    end
    puts
  end

  # ---- rg_wss_client footprint ---------------------------------------------
  # The leak test. RSS on its own says nothing -- Python is not small. RSS
  # bucketed by how long the process has been alive is the discriminator: a
  # rising median across the age buckets is our leak, a flat one means the
  # memory that panicked that AP went somewhere a task list never shows.
  rss_rows = rows.reject { |r|
    r[:f]['clirss'].nil? || r[:f]['clirss'].to_i.zero? ||
      r[:f]['clientage'].to_i.zero?
  }
  if rss_rows.any?
    all_rss = rss_rows.map { |r| r[:f]['clirss'].to_i }
    puts 'rg_wss_client footprint  (VmRSS at probe time):'
    puts format('  %-40s %4d MB  median   (%d AP(s))',
                'resident set', median(all_rss) / 1024, rss_rows.size)
    puts format('  %-40s %4d MB  on %s',
                'largest', all_rss.max / 1024,
                rss_rows.max_by { |r| r[:f]['clirss'].to_i }[:ap].name)
    puts '  RSS by process age  (a rising median here is a leak):'
    [['a. < 1 hour', 0...3600], ['b. 1-24 hours', 3600...86_400],
     ['c. 1-7 days', 86_400...604_800], ['d. > 7 days', 604_800..]].each do |label, range|
      g = rss_rows.select { |r| range.cover?(r[:f]['clientage'].to_i) }
      next if g.empty?

      v = g.map { |r| r[:f]['clirss'].to_i }
      puts format('    %-20s %5d AP(s)  median %4d MB   max %4d MB',
                  label, g.size, median(v) / 1024, v.max / 1024)
    end
    puts
  end

  tmp_left = rows.count { |r| r[:f]['tmpleft'].to_i > 0 }
  oom_logs = rows.count { |r| r[:f]['oomlogs'].to_i > 0 }
  ufp_top  = mem_rows.reject { |r| r[:f]['ufprss'].to_i.zero? }
                     .sort_by { |r| -r[:f]['ufprss'].to_i }.first(5)
  puts 'Memory-pressure evidence:'
  puts format('  %-40s %4d   interrupted download left a .tmp behind', 'APs with leftover .ko.tmp', tmp_left)
  puts format('  %-40s %4d   OOM kill present in logread -- see crash records, a panic wipes this',
              'APs with OOM in logs', oom_logs)
  ufp_absent  = mem_rows.count { |r| r[:f]['ufpbin'].to_s == '0' }
  ufp_idle    = mem_rows.count { |r| r[:f]['ufpbin'].to_s == '1' && r[:f]['ufprss'].to_i.zero? }
  ufp_running = mem_rows.count { |r| r[:f]['ufprss'].to_i > 0 }
  puts format('  %-40s %4d   /usr/sbin/ufpd not present', 'APs without ufpd installed', ufp_absent)
  puts format('  %-40s %4d   installed but no process -- cannot be leaking', 'APs with ufpd installed, not running', ufp_idle)
  puts format('  %-40s %4d   leak would show here as growing RSS', 'APs with ufpd running', ufp_running)
  if ufp_top.any?
    puts '  highest ufp RSS (the leak is unbounded per refresh cycle):'
    ufp_top.each do |r|
      puts format('    %-28s %8.1f MB  free=%.0f MB  up=%.1f d  %s',
                  r[:ap].name.to_s[0, 28], r[:f]['ufprss'].to_i / 1024.0,
                  r[:f]['memavail'].to_i / 1024.0,
                  r[:f]['uptime'].to_i / 86400.0, r[:verdict])
    end
  end
  puts
end

if bad.any?
  puts 'Distinct probe signatures among failures:'
  bad.group_by { |r| r[:raw].to_s.sub(/ tgt=\S+ kern=\S+.*/, '').sub(/ (memavail|memtotal|uptime|ufprss|overlayfree|oomlogs)=\S+/, '') }
     .sort_by { |_, g| -g.size }.first(8).each do |sig, group|
    puts format('  %4d x  %s', group.size, sig.to_s[0, 88])
  end
  puts
end

if EXAMPLES > 0 && bad.any?
  puts "Example APs per verdict (up to #{EXAMPLES}; full detail in the TSV):"
  bad.group_by { |r| r[:verdict] }.sort_by { |_, g| -g.size }.each do |verdict, group|
    names = group.first(EXAMPLES).map { |r| "#{r[:ap].id}:#{r[:ap].name}" }
    more  = group.size > EXAMPLES ? ", +#{group.size - EXAMPLES} more" : ''
    puts format('  %-26s %s%s', verdict, names.join(', '), more)
  end
  puts
end

# ---- TX power: planned vs applied -------------------------------------------
# A power-plan commit writes the AP record; the AP applies whatever its
# firmware and regulatory domain let it. This compares the two per band, and
# sets each band's client signal beside it, so a plan that never landed, was
# clamped, or landed and starved a band shows up as a number.
#
# Planned comes from the rXg's own resolver (AP dBm, AP relative, zone dBm,
# zone relative -- the value pushed to the AP), with the board's ceiling from
# the same place. Both are guarded: older rXg builds lack them, and then only
# the applied side is reported. The probe's iwinfo txpowerlist maximum is NOT
# used as the ceiling -- on these QCA radios it lists the regulatory span
# (30 dBm) rather than what the radio will actually transmit.
TX_BANDS = [['2.4', '2', '2G', :to_2g_tx_power_ow_dbm, 'tx_power_24_plan'],
            ['5',   '5', '5G', :to_5g_tx_power_ow_dbm, 'tx_power_5_plan'],
            ['6',   '6', '6G', :to_6g_tx_power_ow_dbm, 'tx_power_6_plan']].freeze
TX_TOLERANCE_DB = 2

def ap_tx_plan(ap)
  TX_BANDS.to_h do |label, _, band, resolver, plan_col|
    planned = begin
      ap.respond_to?(resolver) ? ap.public_send(resolver) : nil
    rescue StandardError
      nil
    end
    ceiling = begin
      ap.respond_to?(:max_tx_power_ow_dbm) ? ap.max_tx_power_ow_dbm(band) : nil
    rescue StandardError
      nil
    end
    plan = ap.has_attribute?(plan_col) ? ap.read_attribute(plan_col).presence : nil
    [label, { planned: planned&.to_i, ceiling: ceiling&.to_i, plan: plan }]
  end
end

rows.each { |r| r[:tx] = ap_tx_plan(r[:ap]) if r[:f]['tp2a'] }
tx_rows = rows.select { |r| r[:tx] }
if tx_rows.any?
  puts 'TX power  (planned on the rXg vs applied on the AP, per band):'
  puts format('  %-5s %4s %8s %6s %6s %6s %6s %6s %6s  %s', 'BAND', 'APs', 'PLANNED',
              'MATCH', 'LOWER', 'HIGHER', 'STA', '<-70', '<-80', 'APPLIED dBm (count)')
  TX_BANDS.each do |label, k, _, _, _|
    g = tx_rows.select { |r| r[:f]["tp#{k}a"].to_i.positive? }
    next if g.empty?
    with_plan = g.reject { |r| r[:tx][label][:planned].nil? }
    diff = ->(r) { r[:f]["tp#{k}a"].to_i - r[:tx][label][:planned] }
    match  = with_plan.count { |r| diff.(r).abs <= TX_TOLERANCE_DB }
    lower  = with_plan.count { |r| diff.(r) < -TX_TOLERANCE_DB }
    higher = with_plan.count { |r| diff.(r) > TX_TOLERANCE_DB }
    sta  = g.sum { |r| r[:f]["s#{k}n"].to_i }
    weak = g.sum { |r| r[:f]["s#{k}w"].to_i }
    edge = g.sum { |r| r[:f]["s#{k}e"].to_i }
    applied = g.map { |r| r[:f]["tp#{k}a"].to_i }.tally.sort_by { |_, n| -n }.first(6)
               .map { |v, n| "#{v}x#{n}" }.join(' ')
    puts format('  %-5s %4d %8d %6d %6d %6d %6d %6s %6s  %s', label, g.size, with_plan.size,
                match, lower, higher, sta, pct(weak, sta), pct(edge, sta), applied)
  end
  puts "  PLANNED = APs whose rXg record resolves to a dBm; MATCH within #{TX_TOLERANCE_DB} dB."
  puts '  LOWER = applied below planned: clamped by board/regulatory, or the plan never applied.'

  # The 2.4/5 split is where a power plan most often hurts: drop 2.4 to keep
  # cells small and dual-band clients still associate on it at the edge.
  gaps = tx_rows.select { |r| r[:f]['tp2a'].to_i.positive? && r[:f]['tp5a'].to_i.positive? }
                .map { |r| r[:f]['tp5a'].to_i - r[:f]['tp2a'].to_i }
  if gaps.any?
    puts format('  5 GHz minus 2.4 GHz applied        median %+d dB   range %+d..%+d   (%d APs)',
                median(gaps), gaps.min, gaps.max, gaps.size)
  end
  over = tx_rows.flat_map { |r|
    TX_BANDS.filter_map { |label, k, _, _, _|
      c = r[:tx][label][:ceiling]
      p = r[:tx][label][:planned]
      [r, label, p, c] if c && p && p > c
    }
  }
  puts format('  %-38s %6d   plan exceeds the board ceiling the rXg knows',
              'planned above board ceiling', over.size) if over.any?
  puts

  # Applied 2.4 GHz power against what clients experience on that AP. If low
  # 2.4 power is driving the churn, the low rows carry the weak share and the
  # timeouts; if the rows look alike, power is not the lever.
  buckets = [['a. <= 6 dBm', 0..6], ['b. 7-10 dBm', 7..10], ['c. 11-15 dBm', 11..15],
             ['d. 16-20 dBm', 16..20], ['e. > 20 dBm', 21..99]]
  puts 'Client outcome by applied 2.4 GHz power:'
  puts format('  %-14s %4s %6s %7s %7s %9s %9s', 'TX 2.4', 'APs', 'STA2.4', '<-70',
              '<-80', 'TIMEOUT', 'REASSOC')
  buckets.each do |label, range|
    g = tx_rows.select { |r| range.cover?(r[:f]['tp2a'].to_i) }
    next if g.empty?
    sta = g.sum { |r| r[:f]['s2n'].to_i }
    el  = g.sum { |r| r[:f]['celig'].to_i }
    nm  = g.sum { |r| r[:f]['nmac'].to_i }
    puts format('  %-14s %4d %6d %7s %7s %9s %9s', label, g.size, sta,
                pct(g.sum { |r| r[:f]['s2w'].to_i }, sta),
                pct(g.sum { |r| r[:f]['s2e'].to_i }, sta),
                pct(g.sum { |r| r[:f]['inact'].to_i }, el),
                pct(g.sum { |r| r[:f]['reassoc'].to_i }, nm))
  end
  puts '  TIMEOUT is window-normalised (evictions / assocs with 300s of ring); ' \
       'REASSOC is same-MAC re-associations / distinct MACs.'
  puts

  mism = tx_rows.flat_map { |r|
    TX_BANDS.filter_map { |label, k, _, _, _|
      a = r[:f]["tp#{k}a"].to_i
      p = r[:tx][label][:planned]
      [r, label, p, a, r[:tx][label][:plan]] if a.positive? && p && (a - p).abs > TX_TOLERANCE_DB
    }
  }
  if EXAMPLES.positive? && mism.any?
    puts "Planned/applied mismatches (#{mism.size}; up to 10 largest):"
    puts format('  %-30s %-5s %8s %8s  %s', 'AP', 'BAND', 'PLANNED', 'APPLIED', 'PLAN')
    mism.sort_by { |_, _, p, a, _| -(a - p).abs }.first(10).each do |r, label, p, a, plan|
      puts format('  %-30s %-5s %8d %8d  %s', r[:ap].name.to_s[0, 30], label, p, a, plan || '-')
    end
    puts
  end
end

# ---- Switch VLAN drift ------------------------------------------------------
# Untunneled per-key VLANs on swconfig APs, checked against the switch chip's
# VLAN table. See the probe comment for the mechanism. Two shapes matter and
# they mean different things:
#   PARTIAL  some vlan tunnels programmed, some not -- the drift signature: the
#            ones added after the last config apply survived, older ones did not
#   ALL      none programmed -- either every tunnel predates the last apply, or
#            the platform was never programmed at all (no WAN port discovered);
#            the by-model split and enable_vlan tell those apart
sw_any  = rows.select { |r| r[:f]['swn'].to_i.positive? }
sw_rows = sw_any.select { |r| r[:f]['swpath'] == '1' || r[:f]['swprog'].to_i.positive? }
if sw_any.any?
  vt_rows   = sw_rows.select { |r| r[:f]['vtun'].to_i.positive? }
  miss_rows = vt_rows.select { |r| r[:f]['swmiss'].to_i.positive? }
  part_rows = miss_rows.select { |r| r[:f]['swok'].to_i.positive? }
  all_rows  = miss_rows.reject { |r| r[:f]['swok'].to_i.positive? }
  vt_t   = vt_rows.sum { |r| r[:f]['vtun'].to_i }
  miss_t = vt_rows.sum { |r| r[:f]['swmiss'].to_i }
  mac_t  = vt_rows.sum { |r| r[:f]['swmac'].to_i }

  puts 'Switch VLAN drift  (plain-VLAN sdan tunnels vs the switch chip VLAN table):'
  puts format('  %-38s %6d', 'APs with a swconfig switch', sw_any.size)
  puts format('  %-38s %6d   no WAN port named and no programming logged; ignored',
              '..switch not in the uplink path', sw_any.size - sw_rows.size)
  puts format('  %-38s %6d', '..switch in the uplink path', sw_rows.size)
  puts format('  %-38s %6d   client logged programming a switch VLAN',
              '..by the client\'s own log', sw_any.count { |r| r[:f]['swprog'].to_i.positive? })
  puts format('  %-38s %6d   client logged no WAN/CPU port for a switch',
              '..client declined (no WAN port)', sw_any.count { |r| r[:f]['swnowan'].to_i.positive? })
  puts format('  %-38s %6d', '..carrying plain-VLAN tunnels', vt_rows.size)
  puts format('  %-38s %6d', 'plain-VLAN tunnels', vt_t)
  puts format('  %-38s %6d   %s   <<< untunneled traffic dropped in the switch',
              '..VID on no switch', miss_t, pct(miss_t, vt_t).strip)
  puts format('  %-38s %6d   UCI switch_vlan section gone',
              '..VID absent from UCI', vt_rows.sum { |r| r[:f]['ucimiss'].to_i })
  puts format('  %-38s %6d   %s', 'APs with >= 1 dead VLAN', miss_rows.size,
              pct(miss_rows.size, vt_rows.size).strip)
  puts format('  %-38s %6d   some VLANs fine, older ones not -- drift',
              '..PARTIAL', part_rows.size)
  puts format('  %-38s %6d   none programmed', '..ALL', all_rows.size)
  puts format('  %-38s %6d   learned by sdan, on a dead VLAN now',
              'client MACs on a dead VLAN', mac_t)

  # The mechanism predicts drift only where the network config was rewritten
  # while the client was running. If dead VLANs also show up where it was not,
  # something else is removing them.
  aged = vt_rows.select { |r| r[:f]['netcfgage'].to_i >= 0 && r[:f]['clientage'].to_i.positive? }
  if aged.any?
    rewr, kept = aged.partition { |r| r[:f]['netcfgage'].to_i < r[:f]['clientage'].to_i }
    [['config rewritten under the client', rewr],
     ['config older than the client', kept]].each do |label, g|
      bad_n = g.count { |r| r[:f]['swmiss'].to_i.positive? }
      puts format('  %-38s %6d   %d with dead VLANs  %s', label, g.size, bad_n,
                  pct(bad_n, g.size).strip)
    end
  end

  # Does it line up with the client-visible symptom? Same-MAC reassociations
  # are what a phone that never gets a lease does; settled-silent stations are
  # the ones that stayed associated and passed nothing.
  [['APs with dead VLANs', miss_rows],
   ['APs without', vt_rows - miss_rows]].each do |label, g|
    next if g.empty?
    el = g.sum { |r| r[:f]['stelig'].to_i }
    si = g.sum { |r| r[:f]['stsil'].to_i }
    ra = g.sum { |r| r[:f]['reassoc'].to_i }
    nm = g.sum { |r| r[:f]['nmac'].to_i }
    puts format('  %-38s silent %d/%d %s   reassociating MACs %d/%d %s',
                label, si, el, pct(si, el).strip, ra, nm, pct(ra, nm).strip)
  end
  puts

  by = ->(title, label, key, width) do
    puts title
    puts format('  %-*s %5s %5s %5s %7s %6s  %s', width, label,
                'APs', 'DEAD', 'DEAD%', 'TUN', 'DEADT', 'ENABLE_VLAN')
    vt_rows.group_by(&key).sort_by { |_, g| -g.count { |r| r[:f]['swmiss'].to_i.positive? } }
           .each do |k, g|
      dead = g.count { |r| r[:f]['swmiss'].to_i.positive? }
      en = g.map { |r| r[:f]['swen'] }.tally.sort_by { |_, n| -n }.first(2)
            .map { |v, n| "#{v} x#{n}" }.join(', ')
      puts format('  %-*s %5d %5d %5s %7d %6d  %s', width, k.to_s[0, width], g.size,
                  dead, pct(dead, g.size), g.sum { |r| r[:f]['vtun'].to_i },
                  g.sum { |r| r[:f]['swmiss'].to_i }, en)
    end
    puts
  end
  if vt_rows.any?
    by.call('Dead VLANs by model:', 'MODEL', ->(r) { r[:ap].model.presence || '(unknown)' }, 26)
    by.call('Dead VLANs by rg_wss_client version:', 'CLIENT', ->(r) { r[:cv] }, 14)
    by.call('Dead VLANs by firmware:', 'FIRMWARE', ->(r) { r[:fw] }, 30)
  end

  if EXAMPLES.positive? && miss_rows.any?
    puts "Worst APs by dead VLANs (up to #{[EXAMPLES * 3, 10].max}):"
    miss_rows.sort_by { |r| [-r[:f]['swmac'].to_i, -r[:f]['swmiss'].to_i] }
             .first([EXAMPLES * 3, 10].max).each do |r|
      f = r[:f]
      puts format('  %-28s dead %3d/%-3d macs %-3d cfg %6ss client %7ss  %s',
                  r[:ap].name.to_s[0, 28], f['swmiss'].to_i, f['vtun'].to_i,
                  f['swmac'].to_i, f['netcfgage'], f['clientage'],
                  f['swml'].to_s[0, 40])
    end
    puts
  end
end

# ---- Missing VLANs vs what the rXg actually saw -----------------------------
# The AP's word that a VLAN is missing from its switch is a claim about the
# hardware; this checks it against two independent witnesses, per client:
#   the AP     bytes it has sent the station (stv), and
#   the rXg    the DHCP messages it logged for that MAC since the session began
#              (dhcp_messages, whose `via` names the interface each message
#              came in on).
# Groups: clients on a missing VLAN of an in-path switch (the suspects), clients
# on a programmed VLAN of the same APs (same hardware, same firmware), and
# clients on APs with no switch in the uplink path (baseline).
#
# Prediction if the switch drops a missing VID in both directions: suspects are
# INVISIBLE to the rXg -- their DISCOVER never arrives -- rather than "deaf",
# which needs the DISCOVER to arrive and the OFFER to be lost. Deaf suspects
# would mean upstream passes and only the return path dies.
#
# Only sessions that began inside the window are judged: a client that has
# held its lease for hours sends nothing and proves nothing. Every reconnect
# does send DHCP -- DISCOVER, or REQUEST from INIT-REBOOT -- so a recent
# session with no message at all is the real signal.
#   DHCP_MINUTES=60   window (default 60)
#   STARVED_BYTES=20480  AP->STA bytes under which a session >= 60s is starved
DHCP_MINUTES  = (ENV['DHCP_MINUTES'] || 60).to_i
STARVED_BYTES = (ENV['STARVED_BYTES'] || 20_480).to_i
DHCP_SLACK_S  = 30
DEAF_HAVING   = 20 # same threshold as the dashboard's Deaf Clients gauges
PLAIN_VLAN_UNITS = 4096 # vlan0-vlan4095 are tag == unit; QinQ units start here

# The VLAN tag a DHCP message arrived on, from dhcpd's "via vlan3439". For an
# ordinary VLAN the interface unit is the tag. vlans.tag cannot stand in for
# it: vlan_id points at the Vlan record, and one auto-increment record carries
# every per-unit tag under a single base tag. QinQ units, relay IPs and
# physical ports name no comparable tag, so they give nil.
via_tag = lambda do |via|
  unit = via.to_s[/\Avlan(\d+)\z/, 1]&.to_i
  unit if unit && unit < PLAIN_VLAN_UNITS
end

stations = rows.flat_map do |r|
  stv = r[:f]['stv'].to_s
  next [] if stv.empty? || stv == '-'
  in_path = r[:f]['swpath'] == '1' || r[:f]['swprog'].to_i.positive?
  stv.split(';').filter_map do |rec|
    mac, vlan, dead, ct, tb, rb = rec.split(',')
    next unless mac && ct
    group = if !in_path then :baseline
            elsif dead == '1' then :missing
            else :programmed
            end
    { r: r, mac: norm_mac(mac), vlan: vlan.to_i, group: group, ct: ct.to_i,
      tb: tb.to_i, rb: rb.to_i, at: r[:at] || Time.current }
  end
end

if stations.any? && defined?(DhcpMessage)
  window = DHCP_MINUTES * 60
  recent = stations.select { |s| s[:ct] <= window }
  msgs = Hash.new { |h, k| h[k] = [] }
  dhcp_live = nil
  begin
    dhcp_live = DhcpMessage.where('created_at > ?', 10.minutes.ago).exists?
    if recent.any?
      DhcpMessage.where(mac: mac_variants(recent.map { |s| s[:mac] }.uniq))
                 .where('created_at > ?', Time.current - window - 600)
                 .pluck(:mac, :message_type, :created_at, :via)
                 .each { |m, t, at, via| msgs[norm_mac(m)] << [t.to_s.upcase, at, via_tag.(via)] }
    end
  rescue StandardError => e
    puts "DHCP join skipped: #{e.class}: #{e.message}"
    recent = []
  end

  classify = lambda do |s|
    since = s[:at] - s[:ct] - DHCP_SLACK_S
    ev = msgs[s[:mac]].select { |_, at, _| at >= since }
    types = ev.map(&:first)
    if types.include?('ACK')
      tags = ev.select { |t, _, _| t == 'ACK' }.map(&:last).compact
      if tags.empty? then :acked_vlan_unknown
      elsif tags.include?(s[:vlan]) then :acked
      else :acked_other_vlan
      end
    elsif types.include?('NAK') then :nak
    elsif types.include?('OFFER') then :deaf
    elsif types.any? then :unanswered
    else :invisible
    end
  end
  recent.each { |s| s[:dhcp] = classify.(s) }

  groups = [[:missing, 'missing VLAN (in-path switch)'],
            [:programmed, 'programmed VLAN (same APs)'],
            [:baseline, 'no switch in path (baseline)']]
  puts "Missing VLANs vs the rXg's DHCP log  (sessions begun in the last #{DHCP_MINUTES}m):"
  puts '  (no DHCP messages logged in the last 10 minutes -- is DHCP logging on?)' unless dhcp_live
  puts format('  %-30s %5s %6s %6s %6s %6s %6s %6s', 'GROUP', 'STA', 'ACKED',
              'OTHERV', 'NAK', 'DEAF', 'UNANS', 'INVIS')
  groups.each do |key, label|
    g = recent.select { |s| s[:group] == key }
    next if g.empty?
    c = g.map { |s| s[:dhcp] }.tally
    c[:acked] = c[:acked].to_i + c.delete(:acked_vlan_unknown).to_i
    puts format('  %-30s %5d %6s %6s %6s %6s %6s %6s', label, g.size,
                *%i[acked acked_other_vlan nak deaf unanswered invisible]
                  .map { |k| "#{c[k].to_i}" })
  end
  puts '  ACKED lease granted; OTHERV acked on a different VLAN than the AP assigned;'
  puts '  DEAF offer sent, never requested; UNANS client spoke, rXg never offered;'
  puts '  INVIS rXg logged nothing from this MAC since the session began.'
  unknown = recent.count { |s| s[:dhcp] == :acked_vlan_unknown }
  if unknown.positive?
    puts "  #{unknown} ACKED came via a relay, port or QinQ interface; VLAN not compared."
  end
  puts

  puts "AP->client traffic by group  (sessions >= 60s; starved = under #{STARVED_BYTES / 1024} KB sent):"
  puts format('  %-30s %5s %8s %6s %10s', 'GROUP', 'STA', 'STARVED', 'RATE', 'MEDIAN TX')
  groups.each do |key, label|
    g = stations.select { |s| s[:group] == key && s[:ct] >= 60 }
    next if g.empty?
    st = g.count { |s| s[:tb] < STARVED_BYTES }
    puts format('  %-30s %5d %8d %6s %8d KB', label, g.size, st, pct(st, g.size),
                median(g.map { |s| s[:tb] }) / 1024)
  end
  puts

  # The dashboard's own notion of a deaf client, fleet-wide, placed against
  # the groups above. If missing VLANs were behind the deaf clients, they would
  # cluster in the first row; if they sit in the baseline, deafness is a
  # different fault (e.g. the ipq60xx switch-chip ARL mislearn).
  begin
    deaf_sql = "WITH recent AS (SELECT * FROM dhcp_messages " \
               "WHERE created_at > now() - interval '#{DHCP_MINUTES} minutes') " \
               'SELECT mac FROM (SELECT *, lag(message_type) OVER ' \
               '(PARTITION BY mac ORDER BY created_at) AS prev_message_type, ' \
               'created_at - lag(created_at) OVER (PARTITION BY mac ORDER BY created_at) AS delta ' \
               "FROM recent) s WHERE message_type != 'REQUEST' AND prev_message_type = 'OFFER' " \
               "AND delta < interval '1 minute' GROUP BY mac HAVING count(*) > #{DEAF_HAVING}"
    deaf = ActiveRecord::Base.connection.select_values(deaf_sql).map { |m| norm_mac(m) }.uniq
    where = stations.group_by { |s| s[:mac] }.transform_values { |ss| ss.first[:group] }
    placed = deaf.map { |m| where[m] || :not_associated }.tally
    puts "Deaf DHCP clients, last #{DHCP_MINUTES}m  (dashboard definition, > #{DEAF_HAVING} ignored offers):"
    puts format('  %-38s %6d', 'deaf clients on the rXg', deaf.size)
    (groups.map { |k, l| [k, l] } + [[:not_associated, 'not on any probed AP now']]).each do |k, l|
      puts format('  %-38s %6d', "..#{l}", placed[k].to_i)
    end
    puts
  rescue StandardError => e
    puts "Deaf-client query skipped: #{e.class}: #{e.message}"
    puts
  end
end

# ---- Cell overlap: are clients on their own unit's AP? ---------------------
# The question is whether devices sit on the unit above or below because the
# cells overlap through walls and floors. Measured from the APs' side: where
# each client sits relative to its own unit, how good its link is there, and,
# with PROBE_SECS, how loudly its own unit's AP hears it compared with the AP
# it is actually on.
#
# Probe signals are uplink -- what an AP hears from the client. With every AP
# at the same power the client hears the same ranking, shifted by the power
# difference, so the gaps between APs carry over and absolute levels do not.
# The join and probe thresholds compare exactly this uplink signal, which is
# why the threshold what-if needs no assumption about client transmit power,
# and why lowering AP power does not change which APs those thresholds admit.
#
# Matching a client to its own unit's AP prefers PMS room records: the
# client's own VLAN assignment (or its registered device's account) and the AP
# each link to a room, sub-rooms fold into their primary room, and the IDs are
# compared, with no text involved. Unit numbers parsed from room or AP names
# are used only where a side has no room link, because naming differs between
# properties; a numeric account login only with UNIT_FROM_LOGIN=1. The report
# prints how often AP names agree with the APs' own PMS rooms, how many matches
# each method made, and a self-check on the result (UNIT_FROM_NAME=0 turns
# AP-name parsing off).
#
# The above/below/same-floor split assumes units are numbered floor + two-digit
# position (227 = floor 2, unit 27). That is not true everywhere, so it is off
# unless UNIT_LAYOUT=floor.

DFS_CHANNELS    = (52..144)
JOIN_THRESHOLDS = [-75, -72, -70, -67].freeze
UNIT_LAYOUT     = ENV['UNIT_LAYOUT'].to_s
UNIT_FROM_NAME  = ENV['UNIT_FROM_NAME'] != '0'
UNIT_FROM_LOGIN = ENV['UNIT_FROM_LOGIN'] == '1'
VLAN_UNIT_MAX   = (ENV['VLAN_UNIT_MAX'] || 25).to_i # more devices than this = a shared VLAN
BLOCK_RANGES    = ENV['BLOCK_UNITS'].to_s.split(',').filter_map do |part|
  ends = part.strip.split('-', 2)
  lo, hi = ends.map { |x| Integer(x, 10) rescue nil }
  hi = lo if ends.size == 1
  (lo..hi) if lo && hi   # a malformed range is dropped, not guessed
end
RELATION_LABELS = {
  home: "own unit's AP", away: "another unit's AP",
  vertical: 'unit directly above/below', same_floor: 'same floor, within 2 units',
  near: 'diagonal (1 floor, 2 units)', other: 'other unit, further away',
  common: 'AP not linked to a unit'
}.freeze

def unit_key(text, pms: false)
  t = text.to_s
  # Name fallback: "...-RM227", or a trailing "-441b". The trailing form needs
  # a hyphen, so a model number like "labap3 ipq60xx 660a" is not a unit.
  m = if pms then t.match(/0*(\d{2,5})/)
      else t.match(/\bRM[\s-]*0*(\d{2,5})/i) || t.match(/(?:\A|-)0*(\d{3,5})[a-z]?\z/i)
      end
  m && m[1]
end

def floor_relation(home, serv)
  h, s = home.to_s, serv.to_s
  return :away unless h =~ /\A\d{3,}\z/ && s =~ /\A\d{3,}\z/
  hf, hp, sf, sp = h.to_i / 100, h.to_i % 100, s.to_i / 100, s.to_i % 100
  return :vertical if hp == sp && (hf - sf).abs == 1
  return :same_floor if hf == sf && (hp - sp).abs <= 2
  return :near if (hf - sf).abs <= 1 && (hp - sp).abs <= 2
  :other
end

def canon_room(room)
  room && (room.try(:primary_room_id).presence || room.id)
end

def dfs?(ch)
  DFS_CHANNELS.cover?(ch.to_i)
end

ov_num = ->(v) { v.nil? || v == '-' ? nil : v.to_f }
ov_stations = rows.flat_map do |r|
  r[:f]['stq'].to_s.split(';').filter_map do |rec|
    mac, band, vlan, sig, txr, rxr, rt, tf, tp, ct, iv = rec.split(',')
    next unless mac && iv
    { r: r, mac: mac, band: band.to_i, vlan: (vlan.nil? || vlan == '-' ? nil : vlan.to_i),
      sig: ov_num.(sig), txr: ov_num.(txr), rxr: ov_num.(rxr), rt: ov_num.(rt),
      tf: ov_num.(tf), tp: ov_num.(tp), ct: ov_num.(ct), iv: ov_num.(iv) }
  end
end

if ov_stations.any?
  # AP side: its PMS room (canonical id, plus the unit number in the room's
  # name), else the unit number in the AP's name.
  rooms = begin
    ids = rows.map { |r| r[:ap].try(:pms_room_id) }.compact.uniq
    ids.any? ? PmsRoom.where(id: ids).index_by(&:id) : {}
  rescue StandardError
    {}
  end
  ap_room, ap_key, ap_src = {}, {}, Hash.new(0)
  name_both, name_agree, name_disagree = 0, 0, []
  rows.each do |r|
    id = r[:ap].id
    room = rooms[r[:ap].try(:pms_room_id)]
    pk = room && unit_key(room.room, pms: true)
    nk = UNIT_FROM_NAME ? unit_key(r[:ap].name) : nil
    if pk && nk
      name_both += 1
      pk == nk ? name_agree += 1 : name_disagree << "#{r[:ap].name} (PMS room #{room.room})"
    end
    ap_room[id] = canon_room(room)
    ap_key[id] = pk || nk
    ap_src[room ? :pms : (nk ? :name : :none)] += 1
  end
  rows_by_room = rows.group_by { |r| ap_room[r[:ap].id] }.except(nil)
  rows_by_key  = rows.group_by { |r| ap_key[r[:ap].id] }.except(nil)
  ap_name_of   = rows.to_h { |r| [r[:ap].id, r[:ap].name] }

  # Client side, per device MAC -- never per VLAN tag: assignments are
  # per-device rows and include expired ones, so a tag's history says nothing
  # about who is on it now. Current assignment for this MAC (preferring one
  # whose tag is the VLAN the AP actually put it on): its PMS room, else its
  # account's rooms, else its PMS guest's rooms. Then the registered Device's
  # account. A numeric login counts only with UNIT_FROM_LOGIN=1.
  vtas = Hash.new { |h, k| h[k] = [] }
  dev_acct = {}
  begin
    variants = mac_variants(ov_stations.map { |s| s[:mac] }.uniq)
    q = VlanTagAssignment.where(mac: variants)
    q = q.where(expired: [false, nil]) if VlanTagAssignment.column_names.include?('expired')
    q.includes(:pms_room, account: :pms_rooms).order(updated_at: :desc).each do |v|
      vtas[norm_mac(v.mac)] << v
    end
    acct_ids = {}
    Device.where(mac: variants).where.not(account_id: nil).pluck(:mac, :account_id)
          .each { |m, a| acct_ids[norm_mac(m)] ||= a }
    accts = Account.where(id: acct_ids.values.uniq).includes(:pms_rooms).index_by(&:id)
    acct_ids.each { |m, a| dev_acct[m] = accts[a] if accts[a] }
  rescue StandardError => e
    puts "Client-to-unit mapping skipped: #{e.class}: #{e.message}"
  end
  login_key = lambda do |acct|
    l = acct.try(:login).to_s
    UNIT_FROM_LOGIN && l =~ /\A\D{0,12}0*(\d{3,4})[a-z]?\z/i ? $1 : nil
  end

  ov_stations.each do |s|
    nm = norm_mac(s[:mac])
    list = vtas.key?(nm) ? vtas[nm] : []
    v = list.find { |x| x.tag.to_i == s[:vlan].to_i } || list.first
    s[:vta_match] = v && v.tag.to_i == s[:vlan].to_i
    rms, src, key = [], nil, nil
    if v
      if (rm = v.try(:pms_room)) then rms, src = [rm], :vta_room
      elsif (ar = v.try(:account).try(:pms_rooms).to_a).any? then rms, src = ar, :vta_account_room
      elsif (gr = v.try(:pms_guest).try(:pms_rooms).to_a).any? then rms, src = gr, :vta_guest_room
      elsif (k = login_key.(v.try(:account))) then key, src = k, :login
      end
    end
    if src.nil? && (a = dev_acct[nm])
      if (ar = a.pms_rooms.to_a).any? then rms, src = ar, :device_account_room
      elsif (k = login_key.(a)) then key, src = k, :login
      end
    end
    s[:has_vta], s[:has_dev] = !v.nil?, dev_acct.key?(nm)
    s[:src]   = src
    s[:rooms] = rms.map { |x| canon_room(x) }.compact.uniq.presence
    s[:unit]  = key || rms.filter_map { |x| unit_key(x.room, pms: true) }.first

    # Own unit's AP(s): those sharing a PMS room with the client; unit
    # numbers are compared only for APs (or clients) with no room link.
    srv_id = s[:r][:ap].id
    s[:srv_unit] = ap_key[srv_id]
    by_id = (s[:rooms] || []).flat_map { |rid| rows_by_room[rid] || [] }
    by_key = s[:unit] ? (rows_by_key[s[:unit]] || []) : []
    by_key = by_key.select { |x| ap_room[x[:ap].id].nil? } if s[:rooms]
    s[:home_rows] = (by_id + by_key).uniq { |x| x[:ap].id }
    s[:basis] = if by_id.any? then (by_key.any? ? :mixed : :pms_id)
                elsif by_key.any? then :unit_number
                end
    home_ids = s[:home_rows].map { |x| x[:ap].id }
    s[:rel] =
      if home_ids.empty? then nil
      elsif home_ids.include?(srv_id) then :home
      elsif ap_room[srv_id].nil? && s[:srv_unit].nil? then :common
      elsif UNIT_LAYOUT == 'floor' then floor_relation(s[:unit], s[:srv_unit])
      else :away
      end
    s[:unit_label] = s[:unit] || (s[:rooms] && "room##{s[:rooms].first}")
    s[:unit_group] = s[:rooms]&.first || s[:unit]
    s[:home_tx5] = s[:home_rows].map { |x| x[:f]['tp5a'].to_i }.max
    s[:home_ch5] = s[:home_rows].map { |x| x[:f]['dfsch'].to_i }.first
  end
  mapped = ov_stations.select { |s| s[:rel] }
  linked = ov_stations.select { |s| s[:src] }
  src_counts = linked.map { |s| s[:src] }.tally
  basis = mapped.map { |s| s[:basis] }.tally
  with_vta = ov_stations.select { |s| s[:has_vta] }

  # Self-check: if the mapping is right, the AP holding most of a unit's
  # devices should usually be that unit's own AP. When it mostly is not, the
  # devices are clustered around somebody else's AP and the mapping -- not
  # the radio -- is what is wrong.
  by_unit = mapped.group_by { |s| s[:unit_group] }
  checks = by_unit.select { |_, g| g.size >= 2 }.map do |_, g|
    busiest, n = g.map { |s| s[:r][:ap].id }.tally.max_by { |_, c| c }
    [g, g.first[:home_rows].any? { |x| x[:ap].id == busiest }, busiest, n]
  end
  self_ok = checks.count { |c| c[1] }
  suspect = checks.size >= 3 && self_ok < checks.size / 2.0

  puts "Cell overlap  (#{ov_stations.size} station(s) reported by #{rows.count { |r| r[:f]['stq'].to_s != '-' && r[:f]['stq'].present? }} AP(s)):"
  puts format('  %-38s %6s   PMS room %d, name only %d', 'APs linked to a unit',
              "#{ap_src[:pms] + ap_src[:name]}/#{rows.size}", ap_src[:pms], ap_src[:name])
  if name_both.positive?
    puts format('  %-38s %6d   agree %d, disagree %d%s', 'AP name vs its own PMS room', name_both,
                name_agree, name_disagree.size,
                name_agree < name_both ? '   <<< name matches are suspect at this site' : '')
    name_disagree.first(EXAMPLES).each { |d| puts "    e.g. #{d}" }
  end
  puts format('  %-38s %6d   %s of stations; tag matches the AP\'s VLAN for %d',
              'stations with a current VLAN assignment', with_vta.size,
              pct(with_vta.size, ov_stations.size).strip, with_vta.count { |s| s[:vta_match] })
  puts format('  %-38s %6d   %s of stations', 'stations with a registered Device',
              ov_stations.count { |s| s[:has_dev] }, pct(ov_stations.count { |s| s[:has_dev] }, ov_stations.size).strip)
  puts format('  %-38s %6d   assignment room %d, assignment account %d, guest %d, device account %d, login %d',
              '..linked to a unit', linked.size, src_counts[:vta_room].to_i,
              src_counts[:vta_account_room].to_i, src_counts[:vta_guest_room].to_i,
              src_counts[:device_account_room].to_i, src_counts[:login].to_i)
  puts format('  %-38s %6d   %s   by PMS room %d, by unit number %d, both %d',
              "..own unit's AP found", mapped.size, pct(mapped.size, ov_stations.size),
              basis[:pms_id].to_i, basis[:unit_number].to_i, basis[:mixed].to_i)
  if checks.any?
    puts format('  %-38s %6s   %s', "self-check: unit's busiest AP is its own",
                "#{self_ok}/#{checks.size}",
                suspect ? '<<< under half: the mapping is wrong here, ignore the home shares below' : 'units with 2+ mapped devices')
    checks.reject { |c| c[1] }.sort_by { |c| -c[3] }.first(EXAMPLES).each do |g, _, busiest, n|
      puts "    e.g. unit #{g.first[:unit_label]}: busiest AP #{ap_name_of[busiest]} (#{n} of #{g.size} devices)"
    end
  end
  puts '  Unit-number matches depend on this site\'s naming; PMS room matches do not.'
  puts '  UNIT_FROM_LOGIN=1 also reads a unit from a numeric account login.' unless UNIT_FROM_LOGIN
  puts

  rel_table = lambda do |title, groups, total|
    retry_pct = lambda do |g|
      w = g.select { |s| s[:rt] && s[:tp]&.positive? }
      tp = w.sum { |s| s[:tp] }
      tp.positive? ? format('%5.1f%%', 100.0 * w.sum { |s| s[:rt] } / tp) : '    -'
    end
    med = ->(g, k) { (x = median(g.map { |s| s[k] }.compact)) ? x.round : '-' }
    puts title
    puts format('  %-30s %5s %6s %5s %6s %8s %8s %7s', 'CLIENT IS ON', 'STA', 'SHARE',
                '5GHZ', 'SIG', 'TX MBPS', 'RX MBPS', 'RETRY')
    groups.each do |label, g|
      next if g.empty?
      puts format('  %-30s %5d %6s %5s %6s %8s %8s %7s', label, g.size, pct(g.size, total),
                  pct(g.count { |s| s[:band] == 5 }, g.size), med.(g, :sig), med.(g, :txr),
                  med.(g, :rxr), retry_pct.(g))
    end
    puts '  SIG is what the serving AP hears from the client; TX is AP->client, RX client->AP;'
    puts '  RETRY is tx retries / tx packets. Medians.'
    puts
  end

  if mapped.any?
    rel_table.("Where clients are, relative to their own unit#{suspect ? '  (MAPPING SUSPECT -- see self-check)' : ''}:",
               RELATION_LABELS.map { |k, l| [l, mapped.select { |s| s[:rel] == k }] }, mapped.size)

    home_share = lambda do |g|
      format('%5d %6d  %s', g.size, g.count { |s| s[:rel] == :home }, pct(g.count { |s| s[:rel] == :home }, g.size))
    end
    puts "Share on their own unit's AP, by:"
    puts format('  %-38s %5s %6s  %s', '', 'STA', 'HOME', 'SHARE')
    [[2, '2.4 GHz'], [5, '5 GHz']].each do |b, l|
      g = mapped.select { |s| s[:band] == b }
      puts format('  %-38s %s', "client band #{l}", home_share.(g)) if g.any?
    end
    mapped.group_by { |s| s[:home_tx5] }.sort_by { |p, _| p.to_i }.each do |p, g|
      puts format('  %-38s %s', "own unit's AP at #{p} dBm on 5 GHz", home_share.(g))
    end
    [[true, 'on DFS'], [false, 'not on DFS']].each do |d, l|
      g = mapped.select { |s| s[:home_ch5].to_i.positive? && dfs?(s[:home_ch5]) == d }
      puts format('  %-38s %s', "own unit's AP 5 GHz #{l}", home_share.(g)) if g.any?
    end
    puts '  If clients whose own AP is on DFS are away much more often, DFS is costing'
    puts '  home attachment independently of transmit power.'
    puts

    puts "Per unit  (#{by_unit.size} unit(s) with at least one mapped device):"
    puts format('  %-22s %6s %8s %8s  %s', 'DEVICES IN UNIT', 'UNITS', 'DEVICES', 'ON HOME', 'UNITS WITH <=1 ON HOME')
    [['1', 1..1], ['2-4', 2..4], ['5+', 5..]].each do |l, range|
      us = by_unit.select { |_, g| range.cover?(g.size) }
      next if us.empty?
      devs = us.values.flatten
      few = us.count { |_, g| g.count { |s| s[:rel] == :home } <= 1 }
      puts format('  %-22s %6d %8d %8s  %d (%s)', l, us.size, devs.size,
                  pct(devs.count { |s| s[:rel] == :home }, devs.size), few, pct(few, us.size).strip)
    end
    puts
  end

  # Unit VLAN spread -- needs no client-to-unit mapping. Each unit has its own
  # VLAN, so the APs a VLAN's devices sit on show how one unit's devices spread
  # out. The VLAN's busiest AP stands in for "home"; that is an upper bound on
  # the true home share, and the self-check above says how often it is right.
  # Very large VLANs are shared ones (onboarding, IoT), not units, and are
  # left out.
  vl = ov_stations.select { |s| s[:vlan] }.group_by { |s| s[:vlan] }
  shared = vl.select { |_, g| g.size > VLAN_UNIT_MAX }
  unit_vlans = vl.reject { |k, _| shared.key?(k) }
  multi = unit_vlans.select { |_, g| g.size >= 2 }
  multi.each_value do |g|
    busiest = g.map { |s| s[:r][:ap].id }.tally.max_by { |_, c| c }.first
    bkey, broom = ap_key[busiest], ap_room[busiest]
    g.each do |s|
      sid = s[:r][:ap].id
      s[:vbusy] = busiest
      s[:vrel] =
        if sid == busiest then :busiest
        elsif (broom && broom == ap_room[sid]) || (bkey && bkey == ap_key[sid]) then :same_unit
        elsif ap_room[sid].nil? && ap_key[sid].nil? then :common
        elsif UNIT_LAYOUT == 'floor' then floor_relation(bkey, ap_key[sid])
        else :away
        end
    end
  end
  if multi.any?
    vs = multi.values.flatten
    spread = multi.values.map { |g| g.map { |s| s[:r][:ap].id }.uniq.size }
    on_busy = vs.count { |s| s[:vrel] == :busiest }
    puts 'Unit VLAN spread  (no client-to-unit mapping needed; one VLAN per unit):'
    puts format('  %-38s %6d   %d with 2+ devices; %d shared VLAN(s) over %d devices left out',
                'VLANs seen', vl.size, multi.size, shared.size, VLAN_UNIT_MAX)
    puts format('  %-38s %s', 'APs serving one VLAN (2+ devices)',
                [1, 2, 3].map { |n| "#{n}: #{spread.count(n)}" }.push("4+: #{spread.count { |n| n >= 4 }}").join('   '))
    in_unit = on_busy + vs.count { |s| s[:vrel] == :same_unit }
    puts format('  %-38s %6d   %s of %d   <<< upper bound on "on their own AP"',
                "devices on their VLAN's busiest AP", on_busy, pct(on_busy, vs.size).strip, vs.size)
    # A unit with two APs splits its own devices between them, so the busiest
    # AP alone understates such units; this is the figure to compare.
    puts format('  %-38s %6d   %s of %d', "..or on another AP in that AP's unit", in_unit,
                pct(in_unit, vs.size).strip, vs.size)
    puts
    vlabels = [[:busiest, "the VLAN's busiest AP"], [:same_unit, 'another AP in that unit']] +
              RELATION_LABELS.except(:home).map { |k, l| [k, l.sub("another unit's AP", 'another AP')] }
    rel_table.("Devices relative to their VLAN's busiest AP#{UNIT_LAYOUT == 'floor' ? ' (floor layout assumed)' : ''}:",
               vlabels.map { |k, l| [l, vs.select { |s| s[:vrel] == k }] }, vs.size)
    per_ap = ov_stations.select { |s| s[:vlan] && !shared.key?(s[:vlan]) }.group_by { |s| s[:r][:ap].id }
                        .select { |_, g| g.size >= 2 }.map { |_, g| g.map { |s| s[:vlan] }.uniq.size }
    puts format('  %-38s %s', 'unit VLANs per AP (APs with 2+ clients)',
                [1, 2, 3].map { |n| "#{n}: #{per_ap.count(n)}" }.push("4+: #{per_ap.count { |n| n >= 4 }}").join('   '))
    puts
  end

  # Probe overlap, per probing device. Associated phones rarely scan and use
  # random MACs when they do, so joining probes to associated clients finds
  # almost nothing; counting how many APs hear each probing device, and how
  # close the strongest two are, needs no join. Signals are uplink.
  # Gaps compare units, not APs: a unit's two APs hear its devices about
  # equally, which is not the ambiguity in question. Each unit counts at the
  # signal of its loudest AP; APs with no unit count on their own.
  unit_of_ap = ->(id) { ap_room[id] ? "r#{ap_room[id]}" : (ap_key[id] ? "k#{ap_key[id]}" : "a#{id}") }
  by_unit_sig = lambda do |aps|
    aps.each_with_object({}) do |(id, x), u|
      k = unit_of_ap.(id)
      u[k] = [x, id] if u[k].nil? || x > u[k][0]
    end
  end
  if PROBE_SECS.positive?
    heard = {}
    rows.each do |r|
      (r[:prb] || []).each do |p|
        b = p[:freq] < 3000 ? 2 : (p[:freq] < 5925 ? 5 : 6)
        slot = ((heard[norm_mac(p[:mac])] ||= {})[b] ||= {})
        slot[r[:ap].id] = p[:max] if slot[r[:ap].id].nil? || p[:max] > slot[r[:ap].id]
      end
    end
    prb_aps = rows.count { |r| r[:prb]&.any? }
    assoc = ov_stations.map { |s| norm_mac(s[:mac]) }.uniq
    puts "Overlap from probe requests  (#{PROBE_SECS}s capture, #{prb_aps}/#{rows.size} AP(s) returned data):"
    puts format('  %-38s %6d   %s', 'distinct devices heard probing', heard.size,
                "#{pct(heard.keys.count { |m| m[1].to_i(16) & 2 != 0 }, heard.size).strip} with a random (locally administered) MAC")
    puts format('  %-38s %6d   of %d associated', 'associated clients heard probing',
                assoc.count { |m| heard.key?(m) }, assoc.size)
    [[5, '5 GHz'], [2, '2.4 GHz']].each do |b, l|
      sets = heard.values.filter_map { |h| h[b]&.values&.sort&.reverse }
      next if sets.empty?
      usets = heard.values.filter_map { |h| h[b] && by_unit_sig.(h[b]).values.map(&:first).sort.reverse }
      multi_p = usets.select { |v| v.size >= 2 }
      puts "  #{l}: #{sets.size} device(s) heard, #{multi_p.size} by APs of 2+ units"
      if multi_p.any?
        gaps = multi_p.map { |v| v[0] - v[1] }
        puts format('    %-36s 0-2 dB: %s   3-5: %s   6-9: %s   10+: %s', "strongest unit's AP minus next unit's",
                    *[0..2, 3..5, 6..9, 10..].map { |rg| pct(gaps.count { |g| rg.cover?(g) }, gaps.size).strip })
        [3, 6].each do |db|
          w = multi_p.map { |v| v.count { |x| x >= v[0] - db } }
          puts format('    %-36s 1: %s   2: %s   3: %s   4+: %s', "units within #{db} dB of the strongest",
                      *[1, 2, 3].map { |n| pct(w.count(n), w.size).strip }, pct(w.count { |n| n >= 4 }, w.size).strip)
        end
      end
      puts "    APs that would admit it, by join threshold (all #{sets.size} heard):"
      JOIN_THRESHOLDS.each do |t|
        a = sets.map { |v| v.count { |x| x >= t } }
        puts format('      %-8s 0 APs: %s   1: %s   2: %s   3+: %s', "#{t} dBm",
                    *[0, 1, 2].map { |n| pct(a.count(n), a.size).strip }, pct(a.count { |n| n >= 3 }, a.size).strip)
      end
    end
    puts '  Each AP hears probes only on its own channel, and clients mostly listen rather'
    puts '  than probe on DFS channels, so APs on DFS are under-represented here.'
    puts

    # Join attempts the capture saw. hostapd checks the association RSSI
    # threshold before publishing "assoc", so an authentication with nothing
    # after it went no further: below ASSOC_T that is the threshold refusing
    # the station; at or above it, something else stopped the join.
    asc = rows.flat_map { |r| (r[:asc] || []).map { |a| a.merge(r: r) } }
    if asc.any?
      ever = rows.flat_map do |r|
        r[:f]['dpe'].to_s.split(';').select { |e| e.start_with?('C,') }.map { |e| e.split(',')[2] }
      end.to_set
      tot, ok = asc.sum { |a| a[:auths] }, asc.sum { |a| a[:ok] }
      low, high = asc.sum { |a| a[:low] }, asc.sum { |a| a[:high] }
      by_dev = asc.select { |a| a[:low].positive? }.group_by { |a| a[:mac] }
      repeat = by_dev.count { |_, g| g.sum { |a| a[:low] } >= 2 }
      puts "Join attempts during the capture  (association RSSI threshold #{ASSOC_T} dBm):"
      puts format('  %-42s %6d', 'authentications seen', tot)
      puts format('  %-42s %6d   %s', '..followed by a join request', ok, pct(ok, tot).strip)
      puts format('  %-42s %6d   %s   <<< refused by the threshold',
                  '..nothing after it, signal below threshold', low, pct(low, tot).strip)
      puts format('  %-42s %6d   %s   another cause (control)',
                  '..nothing after it, at or above threshold', high, pct(high, tot).strip)
      puts format('  %-42s %6d   capture ended before it resolved', '..still open', asc.sum { |a| a[:open] })
      puts format('  %-42s %6d   %d refused 2+ times; %d never connected in any log ring',
                  'devices refused below the threshold', by_dev.size, repeat,
                  by_dev.keys.count { |m| !ever.include?(m) })
      by_dev.sort_by { |_, g| -g.sum { |a| a[:low] } }.first(EXAMPLES * 2).each do |m, g|
        puts format('    e.g. %s  refused %dx  best %d dBm  on %s%s', m, g.sum { |a| a[:low] },
                    g.map { |a| a[:maxun] }.max, g.map { |a| a[:r][:ap].name }.uniq.join(', ')[0, 60],
                    ever.include?(m) ? '' : '  (never connected)')
      end
      puts '  A refused station shows as an authentication with no join request after it.'
      puts '  MBO phones then wait 30 s before retrying; others retry at once.'
      puts
    end
  end

  # Block test: the same spread, split by whether a unit VLAN's busiest AP is
  # in the block. Both halves come from one run, so time of day and load cancel
  # out; compare each half before and after the power change.
  if BLOCK_RANGES.any? && multi.any?
    in_block = lambda do |ap_id|
      k = ap_key[ap_id].to_s
      k.match?(/\A\d+\z/) && BLOCK_RANGES.any? { |rg| rg.cover?(k.to_i) }
    end
    block_rows = rows.select { |r| in_block.(r[:ap].id) }
    tally_s = ->(key) { block_rows.map { |r| r[:f][key].to_i }.tally.sort.map { |p, c| "#{p}x#{c}" }.join(' ') }
    puts "Block test  (units #{ENV['BLOCK_UNITS']}; #{block_rows.size} AP(s) in the block):"
    puts format('  %-38s 5 GHz %s   2.4 GHz %s', "block APs' applied power (dBm x APs)",
                tally_s.('tp5a'), tally_s.('tp2a'))
    puts
    [['in the block', true], ['outside the block (control)', false]].each do |label, want|
      g = multi.values.select { |vg| in_block.(vg.first[:vbusy]) == want }
      vsg = g.flatten
      next if vsg.empty?
      on = vsg.count { |s| s[:vrel] == :busiest }
      inu = on + vsg.count { |s| s[:vrel] == :same_unit }
      rel_table.("Units whose busiest AP is #{label}: #{g.size} VLAN(s), #{vsg.size} device(s), " \
                 "#{pct(on, vsg.size).strip} on the busiest AP, #{pct(inu, vsg.size).strip} in its unit",
                 vlabels.map { |k, l| [l, vsg.select { |s| s[:vrel] == k }] }, vsg.size)
    end
    if heard
      [['in the block', true], ['outside the block', false]].each do |label, want|
        gaps = heard.values.filter_map do |h|
          next unless h[5]
          u = by_unit_sig.(h[5])
          next unless u.size >= 2
          next unless in_block.(u.values.max_by(&:first)[1]) == want
          vals = u.values.map(&:first).sort.reverse
          vals[0] - vals[1]
        end
        next if gaps.empty?
        puts format('  5 GHz unit gap, strongest AP %-18s 0-2 dB: %s   3-5: %s   6-9: %s   10+: %s  (%d devices)',
                    label, *[0..2, 3..5, 6..9, 10..].map { |rg| pct(gaps.count { |x| rg.cover?(x) }, gaps.size).strip },
                    gaps.size)
      end
      puts
    end
  end

  # Dataplane readiness after association. A VLAN with no tunnel on an AP
  # (on 1.0.29 after its last client left; from 1.0.38, which keeps tunnels
  # resident, until its first client since rg_wss_client started) is not
  # forwarded until rg_wss_client builds one: sdan logs
  # "unconfigured" as the client's VLAN interface appears and "ingress hook
  # registered" when forwarding starts, and the gap between them is how long
  # that client had no dataplane. A phone whose DHCP goes unanswered may give
  # up on the AP and try another, so cold and warm associations are compared
  # on what the station did next. Moves are classed against the VLAN's busiest
  # AP, as in the unit VLAN spread.
  dp = []
  conn_by_mac = Hash.new { |h, k| h[k] = [] }
  rows.each do |r|
    evs = r[:f]['dpe'].to_s.split(';').map { |e| e.split(',') }
    next if evs.empty?
    ring = evs.find { |e| e[0] == 'E' }
    ring_end = ring && ring[2].to_i
    by = evs.group_by(&:first)
    cmds = (by['S'] || []).map { |e| e[1].to_i }
    us, hs, auths = by['U'] || [], by['H'] || [], by['A'] || []
    old_win = evs.reject { |x| %w[E A].include?(x[0]) }.last(OLD_DPE_CAP)
    old_cs = old_win.select { |x| x[0] == 'C' }.map(&:object_id)
    old_us = old_win.select { |x| x[0] == 'U' }
    old_hs = old_win.select { |x| x[0] == 'H' }
    age = r[:f]['clientage'].to_i
    started = ring_end && age.positive? ? ring_end - age : nil
    (by['C'] || []).each { |e| conn_by_mac[e[2]] << [e[1].to_i, r[:ap].id] }
    (by['C'] || []).each do |e|
      t, mac, ifc, vlan = e[1].to_i, e[2], e[3], e[4]
      a = { r: r, mac: mac, t: t, vlan: vlan.to_i, tunneled: !vlan.nil? && vlan.match?(/\A\d+\z/) }
      d = (by['D'] || []).find { |x| x[2] == mac && x[1].to_i >= t }
      a[:sess] = d && d[1].to_i - t
      a[:judged] = ring_end && ring_end - t >= 60
      a[:since_start] = started && t - started
      h = nil
      if a[:tunneled]
        vif = "#{ifc}-v#{vlan}"
        u = us.find { |x| x[2] == vif && (x[1].to_i - t).abs <= 3 }
        a[:cold] = !u.nil?
        if u
          # A later registration of the same VIF means it was destroyed and
          # recreated, so a hook after that belongs to the next station.
          nu = us.map { |x| x[3].to_f if x[2] == vif && x[3].to_f > u[3].to_f }.compact.min
          h = hs.find { |x| x[2] == vif && x[3].to_f >= u[3].to_f && (nu.nil? || x[3].to_f < nu) }
          h = nil if h && d && h[1].to_i > d[1].to_i
          a[:delay] = h && (h[3].to_f - u[3].to_f).round(2)
          a[:left_unfwd] = h.nil? && !d.nil?
          a[:cmd] = cmds.any? { |st| u[1].to_i.between?(st - 1, st + 15) }
        end
        uo = old_cs.include?(e.object_id) && old_us.find { |x| x[2] == vif && (x[1].to_i - t).abs <= 3 }
        if uo
          a[:cold_old] = true
          ho = old_hs.find { |x| x[2] == vif && x[3].to_f >= uo[3].to_f }
          a[:delay_old] = ho && (ho[3].to_f - uo[3].to_f).round(2)
        end
      end
      bound = [h, d].compact.map { |x| x[1].to_i }.min || ring_end
      ac = auths.select { |x| x[2] == mac && x[1].to_i >= t - 2 && (bound.nil? || x[1].to_i <= bound + 1) }
      a[:client_t] = ac.first && ac.first[1].to_i - t
      a[:client_codes] = ac.map { |x| x[3] }
      dp << a
    end
  end
  vbusy = multi.transform_values { |g| g.first[:vbusy] }
  dp.each do |a|
    a[:at_busiest] = vbusy[a[:vlan]] == a[:r][:ap].id
    next unless a[:sess]
    td = a[:t] + a[:sess]
    nxt = conn_by_mac[a[:mac]].reject { |_, id| id == a[:r][:ap].id }
                              .select { |tt, _| tt.between?(td - 2, td + 120) }.min_by(&:first)
    next unless nxt
    a[:moved] = nxt[1]
    b = vbusy[a[:vlan]]
    a[:moved_rel] = if b.nil? then :unknown
                    elsif nxt[1] == b then :busiest
                    elsif (ap_room[b] && ap_room[b] == ap_room[nxt[1]]) ||
                          (ap_key[b] && ap_key[b] == ap_key[nxt[1]]) then :same_unit
                    else :elsewhere
                    end
  end
  dp.group_by { |a| a[:r] }.each do |r, g|
    r[:dp_cold] = g.count { |a| a[:cold] }
    r[:dp_max] = g.filter_map { |a| a[:delay] }.max
    r[:dp_unfwd] = g.count { |a| a[:left_unfwd] }
  end

  tun = dp.select { |a| a[:tunneled] }
  if tun.any?
    cold = tun.select { |a| a[:cold] }
    warm = tun.reject { |a| a[:cold] }
    pn = ->(n, t) { t.positive? ? "#{n} (#{pct(n, t).strip})" : '-' }
    puts "Dataplane readiness after association  (#{dp.size} association(s) in the log rings):"
    puts format('  %-38s %6d   %d untunneled not counted', 'associations on a unit VLAN', tun.size, dp.size - tun.size)
    puts format('  %-38s %6d   %s   no tunnel for the VLAN on that AP yet', '..cold', cold.size, pct(cold.size, tun.size))
    puts format('  %-38s %6d   %s   tunnel already there', '..warm', warm.size, pct(warm.size, tun.size))
    dpm = cold.select { |a| a[:delay] }
    if dpm.any?
      ds = dpm.map { |a| a[:delay] }.sort
      q = ->(p) { ds[[(ds.size * p).ceil - 1, 0].max] }
      puts format('  %-38s median %.1fs   p90 %.1fs   max %.1fs   (%d forwarded)',
                  'cold: time until traffic forwarded', q.(0.5), q.(0.9), ds.last, dpm.size)
      [[true, 'during an rXg shell command'], [false, 'outside one']].each do |flag, l|
        g = dpm.select { |a| a[:cmd] == flag }.map { |a| a[:delay] }.sort
        next if g.empty?
        puts format('    %-36s %4d   median %.1fs   max %.1fs', l, g.size, g[g.size / 2], g.last)
      end
      # Right after rg_wss_client starts, every VLAN is cold and stations wait
      # on the rXg's first state push, so a capture taken just after a fleet
      # reload measures the reload. Kept apart from steady state.
      [[true, "within #{CLIENT_START_S / 60} min of rg_wss_client start"], [false, 'later']].each do |flag, l|
        g = dpm.select { |a| a[:since_start] && (a[:since_start] < CLIENT_START_S) == flag }
               .map { |a| a[:delay] }.sort
        next if g.empty?
        puts format('    %-36s %4d   median %.1fs   max %.1fs', l, g.size, g[g.size / 2], g.last)
      end
    end
    # The old pairing, over the old event window: the first hook on the VIF
    # name after "unconfigured", even one registered for a later station.
    # For comparison with readings taken before the pairing fix.
    cold_old = tun.select { |a| a[:cold_old] }
    olds = cold_old.filter_map { |a| a[:delay_old] }.sort
    if olds.any?
      qo = ->(p) { olds[[(olds.size * p).ceil - 1, 0].max] }
      puts format('  %-38s median %.1fs   p90 %.1fs   max %.1fs   (%d measured, %d not forwarding by the end of the ring)',
                  '..same, as measured before Oct 15', qo.(0.5), qo.(0.9), olds.last, olds.size,
                  cold_old.size - olds.size)
    end
    lu = cold.select { |a| a[:left_unfwd] }
    if lu.any?
      ss = lu.map { |a| a[:sess] }.sort
      puts format('  %-38s %6d   stayed median %ds, max %ds   <<< associated, never forwarded',
                  'cold: left before it was forwarded', lu.size, ss[ss.size / 2], ss.last)
    end
    pend = cold.reject { |a| a[:delay] || a[:left_unfwd] }
    puts format('  %-38s %6d', 'cold: not forwarded by end of the ring', pend.size) if pend.any?

    # What the station's own authorization did on the client, for the cold
    # ones that waited. 'a' is logged before add-tunnel runs, so a prompt 'a'
    # with a late hook points below rg_wss_client's decision.
    cause_of = lambda do |a|
      c = a[:client_codes]
      if c.include?('q') then 'tunnel config not known yet (waiting on the rXg state push)'
      elsif c.include?('z') then 'sdan dataplane down on the AP'
      elsif c.include?('n') then 'no VLAN from RADIUS or config'
      elsif c.include?('f') then 'tunnel apply failed'
      elsif c.empty? then 'no sta-authorized logged by rg_wss_client'
      elsif a[:client_t] > 2 then 'rg_wss_client handled it late'
      elsif a[:delay] then 'handled promptly, hook came later'
      elsif a[:left_unfwd] then 'handled promptly, station left before any hook'
      else 'handled promptly, no hook yet'
      end
    end
    slow = cold.reject { |a| a[:delay] && a[:delay] <= COLD_SLOW_S }
    if slow.any?
      puts format('  %-38s %6d   by what rg_wss_client logged for that station:',
                  "cold: over #{COLD_SLOW_S}s or never forwarded", slow.size)
      slow.group_by { |a| cause_of.(a) }.sort_by { |_, g| -g.size }.each do |c, g|
        puts format('    %-60s %4d', c, g.size)
      end
      slow.sort_by { |a| a[:delay] ? -a[:delay] : -(a[:sess] || 1e9) }.first(EXAMPLES).each do |a|
        what = if a[:delay] then format('forwarded after %.1fs', a[:delay])
               elsif a[:left_unfwd] then "left unforwarded after #{a[:sess]}s"
               else 'still not forwarded'
               end
        st = a[:since_start] ? "#{a[:since_start]}s after client start" : 'client start unknown'
        puts format('    e.g. %-28s %s vlan %-5d %s; %s; %s', a[:r][:ap].name.to_s[0, 28], a[:mac], a[:vlan],
                    what, cause_of.(a), st)
      end
    end
    fate = lambda do |g|
      j = g.select { |a| a[:judged] }
      left = j.select { |a| a[:sess] && a[:sess] <= 60 }
      moved = left.select { |a| a[:moved] }
      [j.size, left.size, moved.size, moved.count { |a| a[:moved_rel] == :elsewhere }]
    end
    fc, fw = fate.(cold), fate.(warm)
    puts format('  %-38s %16s %16s', 'what the station did next', 'COLD', 'WARM')
    puts format('  %-38s %16d %16d', 'associations with 60 s of ring after', fc[0], fw[0])
    puts format('  %-38s %16s %16s', 'left that AP within 60 s', pn.(fc[1], fc[0]), pn.(fw[1], fw[0]))
    puts format('  %-38s %16s %16s', '..and joined another AP within 2 min', pn.(fc[2], fc[0]), pn.(fw[2], fw[0]))
    puts format('  %-38s %16s %16s', "....outside its unit's busiest AP/unit", pn.(fc[3], fc[0]), pn.(fw[3], fw[0]))
    ch = cold.select { |a| a[:at_busiest] }
    puts format('  %-38s %6d   %d of them left within 60 s', "cold on the VLAN's busiest AP", ch.size,
                ch.count { |a| a[:judged] && a[:sess] && a[:sess] <= 60 })
    puts '  A cold association that leaves fast and lands elsewhere, much more often than a'
    puts '  warm one, is a device giving up on an AP whose dataplane was not ready.'
    puts
  end

  File.open(OUT_STA, 'w') do |io|
    io.puts %w[ap_id ap_name band mac vlan client_unit map_source vlan_assignment_tag_match
               ap_unit relation match_basis home_aps vlan_busiest_ap vlan_relation
               signal tx_mbps rx_mbps tx_retries tx_failed tx_packets connected_s inactive_ms
               home_ap_tx5 home_ap_ch5].join("\t")
    ov_stations.each do |s|
      io.puts [s[:r][:ap].id, s[:r][:ap].name, s[:band], s[:mac], s[:vlan], s[:unit_label],
               s[:src], s[:has_vta] ? s[:vta_match] : nil, s[:srv_unit], s[:rel], s[:basis],
               s[:home_rows].map { |x| x[:ap].name }.join(','),
               s[:vbusy] && ap_name_of[s[:vbusy]], s[:vrel],
               s[:sig], s[:txr], s[:rxr], s[:rt], s[:tf], s[:tp], s[:ct], s[:iv],
               s[:home_tx5], s[:home_ch5]].map { |x| x.to_s.tr("\t", ' ') }.join("\t")
    end
  end
  puts "Per-station detail (#{ov_stations.size} rows): #{OUT_STA}"
  puts
end

# ---- DFS ---------------------------------------------------------------------
# A radio on a DFS channel listens 60 s before transmitting after any restart,
# must leave within seconds when it detects radar, and cannot return for 30
# minutes. Clients scan DFS channels passively, and some devices skip them
# entirely, so an AP on DFS is also harder for its own unit's devices to find.
# Planned channel is the AP record's override; 0/blank means automatic.
dfs_rows = rows.select { |r| r[:f]['dfsch'].to_i.positive? }
if dfs_rows.any?
  on_dfs = dfs_rows.select { |r| dfs?(r[:f]['dfsch']) }
  radar_now = dfs_rows.select { |r| r[:f]['dfsu'].to_i.positive? }
  radar_logs = dfs_rows.select { |r| r[:f]['radar'].to_i.positive? }
  planned = dfs_rows.map { |r| [r, r[:ap].try(:channel_5_override).to_i] }
  off_plan = planned.select { |r, p| p.positive? && p != r[:f]['dfsch'].to_i }
  puts 'DFS  (5 GHz):'
  puts format('  %-38s %6s', 'radios on DFS channels (52-144)', "#{on_dfs.size}/#{dfs_rows.size}")
  newest = radar_now.map { |r| r[:f]['dfsmin'].to_i }.select { |v| v >= 0 }.min
  puts format('  %-38s %6d   %s', 'radar detected in the last 30 min', radar_now.size,
              newest ? "newest #{fmt_dur(newest)} ago" : '')
  puts format('  %-38s %6d   %d line(s)', "APs with 'radar' in their log ring", radar_logs.size,
              radar_logs.sum { |r| r[:f]['radar'].to_i })
  puts format('  %-38s %6d', 'channel set automatically', planned.count { |_, p| p.zero? })
  puts format('  %-38s %6d   <<< moved by radar, or a plan never applied', 'running a channel other than planned', off_plan.size)
  off_plan.first([EXAMPLES, 10].max).each do |r, p|
    puts format('    %-32s planned %-4d running %-4d %s', r[:ap].name.to_s[0, 32], p, r[:f]['dfsch'].to_i,
                dfs?(p) ? '(planned DFS)' : '')
  end
  puts
end

File.open(OUT, 'w') do |io|
  io.puts %w[ap_id name model firmware client_version zone target_kernel
             rxg_has_build verdict rc ctl ko kohash ctlhash ctl_size ko_size
             loaded mem_avail_kb mem_total_kb uptime_s ufp_rss_kb tmp_leftover
             overlay_free_kb oom_log_lines client_age_s ufpd_installed
             assoc_count inactivity_evictions unconfigured_drops
             vlan_ifaces sdan_tunnels orphan_vlans low_traffic_vlans
             reassociations band_steers log_ring_span_s
             eligible_assocs stations sig_ge60 sig_60_70 sig_70_80 sig_lt80
             sess_lt60s sess_60_300s sess_gt300s
             ch_2g bw_2g txpower_2g ch_5g bw_5g txpower_5g
             survey_active_2g_ms survey_busy_2g_ms
             survey_active_5g_ms survey_busy_5g_ms
             addport_failures tunnel_deletions chain_vlans chain_vlan_list
             peer_delete_timeouts auth_noack assoc_prunes
             peer_not_found rx_tid_failures tunnel_del_premature tunnel_del_ok
             sta_idle_lt1s sta_idle_1_60s sta_idle_gt60s
             rel_sched_gone rel_sched_move rel_cancelled rel_fired
             settled_ok_stations silent_stations sdan_macs sdan_macs_no_tunnel
             client_rss_kb crash_records crash_panic crash_oom
             crash_src crash_path
             sw_count sw_in_uplink sw_enable_vlan vlan_tunnels vlan_sw_ok vlan_sw_missing
             vlan_uci_missing vlan_missing_list macs_on_missing macs_on_missing_list
             netcfg_age_s sw_programmed_logs sw_nowan_logs
             tx24_planned tx24_applied tx24_ceiling tx24_iwinfo_max sta24 sta24_lt70 sta24_lt80 sta24_sig_sum
             tx5_planned tx5_applied tx5_ceiling tx5_iwinfo_max sta5 sta5_lt70 sta5_lt80 sta5_sig_sum
             tx6_planned tx6_applied tx6_ceiling tx6_iwinfo_max sta6 sta6_lt70 sta6_lt80 sta6_sig_sum
             tx24_plan tx5_plan tx6_plan stations
             client_macs evicted_macs
             dfs_unavailable dfs_newest_radar_s radar_log_lines probe_sightings
             station_quality channel_5g_by_freq cold_associations max_cold_delay_s
             cold_left_unforwarded].join("\t")
  rows.sort_by { |r| [r[:verdict], r[:ap].id] }.each do |r|
    f = r[:f]
    io.puts [r[:ap].id, r[:ap].name, r[:ap].model, r[:fw], r[:cv],
             r[:ap].zone, r[:pair], (r[:pair] ? r[:have] : nil), r[:verdict],
             f['rc'], f['ctl'], f['ko'], f['kohash'], f['ctlhash'],
             f['ctlsz'], f['kosz'], f['loaded'], f['memavail'], f['memtotal'],
             f['uptime'], f['ufprss'], f['tmpleft'], f['overlayfree'],
             f['oomlogs'], f['clientage'], f['ufpbin'],
             f['conn'], f['inact'], f['undrop'],
             f['vifs'], f['tuns'], f['orph'], f['vlow'],
             f['reassoc'], f['bandroam'], f['lspan'],
             f['celig'], f['nsta'], f['sg60'], f['sg70'], f['sg80'], f['sg81'],
             f['ct60'], f['ct300'], f['ctlong'],
             f['ch2'], f['bw2'], f['txp2'], f['ch5'], f['bw5'], f['txp5'],
             f['sa2'], f['sb2'], f['sa5'], f['sb5'],
             f['apfail'], f['tundel'], f['chain'], f['chainvl'],
             f['pdto'], f['noack'], f['prune'],
             f['peerfind'], f['rxtid'], f['tdprem'], f['tdok'],
             f['iact'], f['iamid'], f['iastale'],
             f['relgone'], f['relmove'], f['relcan'], f['relfire'],
             f['stelig'], f['stsil'], f['macn'], f['macdrop'],
             f['clirss'], f['crashn'], f['crashpanic'], f['crashoom'],
             f['crashsrc'], f['crashpath'],
             f['swn'], f['swpath'], f['swen'], f['vtun'], f['swok'], f['swmiss'],
             f['ucimiss'], f['swml'], f['swmac'], f['swmacl'], f['netcfgage'],
             f['swprog'], f['swnowan'],
             *TX_BANDS.flat_map { |label, k, _, _, _|
               [r[:tx]&.dig(label, :planned), f["tp#{k}a"], r[:tx]&.dig(label, :ceiling),
                f["tp#{k}m"], f["s#{k}n"], f["s#{k}w"], f["s#{k}e"], f["s#{k}s"]]
             },
             *TX_BANDS.map { |label, *| r[:tx]&.dig(label, :plan) },
             f['stv'],
             f['macs'], f['emacs'],
             f['dfsu'], f['dfsmin'], f['radar'], r[:prb]&.size,
             f['stq'], f['dfsch'], r[:dp_cold], r[:dp_max], r[:dp_unfwd]].map { |v| v.to_s.tr("\t", ' ') }.join("\t")
  end
end
puts "Per-AP detail (#{total} rows): #{OUT}"
puts "  e.g.  awk -F'\\t' '$9 ~ /FETCH/ {print $2, $4, $5}' #{OUT} | sort | uniq -c"
puts

# ---- Remediation (opt-in, off by default) ---------------------------------
#
# A miniloader restart re-runs the once-only sdan fetch, which is currently the
# only thing that recovers an AP in this state. The survey justifies it: APs
# whose client started within the last 24 hours were 44/44 healthy, while the
# bulk-restart window was 224/389 failed. The recovery works -- at a trickle.
#
# So this paces hard and deliberately. A fleet-wide restart is what caused the
# outage: rgminiloader adds MAC-derived jitter of only (MAC mod 60) + 1
# seconds, so 434 APs smear across just 60 seconds -- roughly 7 APs/second each
# pulling ~1.35 MB, which is what exhausted the web tier. The interval below
# stacks on top of that jitter rather than replacing it.
#
#   REMEDIATE=1             arm it (otherwise this section does nothing)
#   REMEDIATE_INTERVAL=10   seconds between APs (default 10)
#   REMEDIATE_MAX=20        cap how many to restart -- start small
#   REMEDIATE_VERIFY=1      re-probe the restarted APs when done
#   REMEDIATE_SETTLE=120    seconds to wait before verifying (default 120)
#
# Long runs hold the session open (225 APs at 10s is ~38 min), so for a full
# pass run it detached -- e.g. under screen, or with nohup and output to a file.

RESTART_CMD = '/etc/init.d/rgminiloader restart'

if ENV['REMEDIATE'].present?
  # Only the retryable classes. NO BUILD ON RXG cannot be fixed by refetching,
  # and a NO_REPLY AP never told us anything -- restarting it would be guessing.
  targets = bad.select { |r| r[:verdict].start_with?('FETCH FAILED', 'CORRUPT') }
               .map { |r| r[:ap] }
  targets = targets.first(ENV['REMEDIATE_MAX'].to_i) if ENV['REMEDIATE_MAX'].present?
  interval = (ENV['REMEDIATE_INTERVAL'] || 10).to_i

  if targets.empty?
    puts 'Remediation: nothing retryable to restart.'
  else
    puts "Remediation: restarting the miniloader on #{targets.size} AP(s), " \
         "one every #{interval}s (~#{(targets.size * interval / 60.0).ceil} min)."
    puts "  command: #{RESTART_CMD}"
    puts '  Each AP drops its WSS session briefly. The rXg reboots APs it sees'
    puts '  offline for ~10 minutes, so do not pace this so fast that many are'
    puts '  down at once.'
    puts

    done = 0
    failed = 0
    targets.each_with_index do |ap, i|
      begin
        # Fire and forget. The restart kills the very client that would deliver
        # the reply, so a missing result here is expected, not a failure -- which
        # is why this does not wait on exit_code like the probe does.
        CommandExecution.create!(source: ap, command: RESTART_CMD,
                                 created_by: 'sdan_fleet_check_remediate')
        done += 1
      rescue StandardError => e
        failed += 1
        warn "  restart dispatch failed for AP #{ap.id} (#{ap.name}): #{e.class}: #{e.message}"
      end
      warn "  restarted #{done}/#{targets.size}" \
           "#{failed.positive? ? " (#{failed} dispatch failure(s))" : ''}"
      sleep interval unless i == targets.size - 1
    end

    puts
    puts "Remediation dispatched: #{done} restart(s)" \
         "#{failed.positive? ? ", #{failed} dispatch failure(s)" : ''}."

    if ENV['REMEDIATE_VERIFY'].present?
      settle = (ENV['REMEDIATE_SETTLE'] || 120).to_i
      # The last AP restarted still has up to 60s of its own jitter to burn
      # before it even begins fetching, hence a settle window rather than an
      # immediate re-probe.
      puts "Waiting #{settle}s for the last AP to jitter, reconnect and fetch..."
      sleep settle

      vrows = probe_aps(AccessPoint.where(id: targets.map(&:id)).order(:id).to_a,
                        builds, TIMEOUT, label: 'Re-probing')
      now_ok = vrows.count { |r| r[:verdict] == 'OK' }
      puts
      puts "Verification: #{now_ok}/#{vrows.size} restarted AP(s) now running rxg_sdan"
      still = vrows.reject { |r| r[:verdict] == 'OK' }
      if still.any?
        puts "  still not running it (#{still.size}):"
        still.group_by { |r| r[:verdict] }.sort_by { |_, g| -g.size }.each do |v, g|
          puts format('    %-28s %4d', v, g.size)
        end
        puts '  A NO_REPLY here may just be an AP still reconnecting; re-run the'
        puts '  survey on its own in a few minutes before treating it as failed.'
      end
    else
      puts 'Re-run the survey (or add REMEDIATE_VERIFY=1) to confirm recovery.'
    end
    puts
  end
end