MARTe-Integrated-Components/Common/Client/go/wshub/trigger.go

package wshub

import (
	"encoding/binary"
	"encoding/json"
	"log"
	"math"
	"sort"
	"strconv"
	"strings"
	"sync"
	"time"

	"github.com/gorilla/websocket"
)

// Trigger FSM states, matching the C++ StreamHub TriggerEngine and the strings
// expected by the web SPA's "triggerState" handler.
const (
	trigIdle       = "idle"
	trigArmed      = "armed"
	trigCollecting = "collecting"
	trigTriggered  = "triggered"
)

// captureMarginSec is the extra delay past the post-trigger window before the
// capture is extracted, so the rings have received the last samples.
const captureMarginSec = 0.15

// captureStallSec is how long the stream may be silent before a collecting
// trigger gives up waiting for the rest of its window and delivers what it has.
const captureStallSec = 2.0

// autoRearmDelaySec is the pause between a completed capture and the automatic
// rearm in "normal" mode.
const autoRearmDelaySec = 0.2

// trigCapturePts caps the points sent per signal in a capture frame. A window
// of 60 s at 1 MSps is 60 M raw samples — ~960 MB per signal on the wire, which
// no client can take and which the send path would simply drop. Matches the C++
// StreamHub's kTrigCapturePts.
const trigCapturePts = 20000

// shortCaptureTol is the fraction of the window a capture may miss at its front
// before it is reported. One min/max bucket of slack, not a quality target.
const shortCaptureTol = 0.01

// maxTriggerWindowSec bounds the capture window, matching the longest option
// the web UI offers. It is not a resolution limit: retuneRings buckets the
// rings so any window fits the per-signal memory budget, at the cost of storing
// min/max pairs rather than every sample.
const maxTriggerWindowSec = 600.0

// trigConfig is the client-settable part of the trigger.
type trigConfig struct {
	signalKey  string // "src:sig" or "src:sig[i]"
	edge       string // "rising" | "falling" | "both"
	threshold  float64
	windowSec  float64
	prePercent float64
	mode       string  // "normal" | "single"
	holdoffSec float64 // rearm delay after a capture (double-trigger guard)
}

// triggerEngine implements the hub-side trigger FSM. Its methods are safe to
// call from the WebSocket read goroutines and from Hub.Run() concurrently.
type triggerEngine struct {
	mu  sync.Mutex
	cfg trigConfig

	// Parsed form of cfg.signalKey, refreshed by SetConfig.
	baseKey string // "src:sig"
	elemIdx int    // -1 when the key has no "[i]" suffix

	state   string
	stopped bool
	// sentState is the state carried by the last stateMsg handed out. The
	// armed→collecting transition happens inside feed(), on the ingest path,
	// so the hub cannot see it by sampling State() across a tick — by the time
	// the tick runs, ingest has already moved the FSM.
	sentState string
	// sentFill is the pre-fill fraction carried by the last stateMsg, so a
	// trigger that is armed but still filling can report progress.
	sentFill float64

	// How far back the trigger signal's ring reaches and how fast that is
	// growing (seconds of span per second of wall clock), refreshed by the hub.
	// bufKnown is false when there is no ring to measure, which disables the
	// fill gate rather than blocking the trigger on a measurement that will
	// never arrive; bufRateOK is false until two measurements exist.
	bufSpan   float64
	bufGrowth float64
	bufKnown  bool
	bufRateOK bool
	// bufCoverage is the maximum span (seconds) the ring can reach at its
	// current bucket and capacity — the gate must never demand more than this,
	// or a ring whose coverage is below the window can never satisfy it. 0 =
	// unknown (no measurable rate).
	bufCoverage float64
	// bufArchived is true when the disk history already spans the trigger
	// window, so a short capture's front can be back-filled from it.
	bufArchived bool
	// Reference point the growth is measured against.
	bufRefSpan, bufRefWall float64

	prevValue float64
	prevValid bool
	lastT     float64
	lastTOK   bool
	// lastFeedWall is the wall clock at the last feed(), used only to notice a
	// stalled stream — the window itself is measured on the sample clock.
	lastFeedWall float64

	trigTime   float64
	firedPre   float64
	firedPost  float64
	firedValid bool

	// The edge to fire on as soon as the FSM rearms, in sample time. Recorded
	// while a capture is still being collected or handed out, for edges late
	// enough that a capture of them would not overlap the one in flight.
	//
	// Without this the trigger is deaf from its own trigger point until the
	// capture has been harvested — a post-window plus captureMarginSec — and
	// then for the holdoff on top of that, and afterwards waits for a FRESH
	// edge. On a sparse pulse train that rounds the capture spacing up to a
	// whole pulse period: at the default 1 s window the blind stretch comes to
	// 1.15 s, so a 1 Hz train was caught at 0.5 Hz and a wider window lost whole
	// multiples. Remembering the edge instead makes the blind stretch exactly
	// the post-window it has to be, since the capture is built from the edge's
	// own timestamp and the ring still holds everything around it.
	pendingT     float64
	pendingValid bool

	rearmAt float64 // wall-clock seconds; 0 when no rearm is pending
}

func newTriggerEngine() *triggerEngine {
	return &triggerEngine{
		cfg:     trigConfig{edge: "rising", windowSec: 1, prePercent: 20, mode: "normal", holdoffSec: autoRearmDelaySec},
		elemIdx: -1,
		state:   trigIdle,
	}
}

// parseSignalKey splits "src:sig[3]" into ("src:sig", 3). A key without an
// element suffix yields an index of -1.
func parseSignalKey(key string) (string, int) {
	if !strings.HasSuffix(key, "]") {
		return key, -1
	}
	open := strings.LastIndexByte(key, '[')
	if open < 0 {
		return key, -1
	}
	idx, err := strconv.Atoi(key[open+1 : len(key)-1])
	if err != nil || idx < 0 {
		return key, -1
	}
	return key[:open], idx
}

func (te *triggerEngine) SetConfig(cfg trigConfig) {
	te.mu.Lock()
	defer te.mu.Unlock()
	// Clamp to the bounds the web UI offers.
	if cfg.windowSec < 1e-4 {
		cfg.windowSec = 1e-4
	}
	if cfg.windowSec > maxTriggerWindowSec {
		cfg.windowSec = maxTriggerWindowSec
	}
	if cfg.prePercent < 0 {
		cfg.prePercent = 0
	}
	if cfg.prePercent > 100 {
		cfg.prePercent = 100
	}
	if cfg.holdoffSec < 0 {
		cfg.holdoffSec = 0
	}
	if cfg.holdoffSec > 60 {
		cfg.holdoffSec = 60
	}
	te.cfg = cfg
	base, idx := parseSignalKey(cfg.signalKey)
	if base != te.baseKey {
		// The buffer measurement belongs to the old signal's ring.
		te.bufKnown, te.bufRateOK = false, false
		te.bufCoverage, te.bufArchived = 0, false
	}
	te.baseKey, te.elemIdx = base, idx
	te.prevValid = false
	te.prevValue = 0
	// An edge held over from the old configuration would be latched against the
	// new window, whose fill the gate has not vouched for.
	te.pendingValid = false
}

func (te *triggerEngine) Config() trigConfig {
	te.mu.Lock()
	defer te.mu.Unlock()
	return te.cfg
}

// Arm starts a fresh acquisition. It is the user's own arm, so it discards any
// edge remembered during the previous capture: the user asked for the next
// event, not for one that has already been and gone.
func (te *triggerEngine) Arm() {
	te.mu.Lock()
	te.state = trigArmed
	te.prevValid = false
	te.prevValue = 0
	te.pendingValid = false
	te.rearmAt = 0
	te.mu.Unlock()
}

// rearm is the automatic arm at the end of a capture. Unlike Arm it honours an
// edge that arrived while the capture was being collected, firing on it at once
// rather than waiting for the next one — see pendingT. It also keeps the level
// tracked through the dead time, so the first sample after rearming is compared
// against its real predecessor instead of being spent seeding one.
func (te *triggerEngine) rearm() {
	te.mu.Lock()
	te.rearmAt = 0
	if te.pendingValid {
		t := te.pendingT
		te.pendingValid = false
		te.latchWindowLocked(t)
	} else {
		te.state = trigArmed
	}
	te.mu.Unlock()
}

func (te *triggerEngine) Disarm() {
	te.mu.Lock()
	te.state = trigIdle
	te.stopped = false
	te.prevValid = false
	te.prevValue = 0
	te.firedValid = false
	te.pendingValid = false
	te.rearmAt = 0
	te.mu.Unlock()
}

func (te *triggerEngine) SetStopped(v bool) {
	te.mu.Lock()
	te.stopped = v
	if v {
		te.rearmAt = 0
	}
	te.mu.Unlock()
}

func (te *triggerEngine) Stopped() bool {
	te.mu.Lock()
	defer te.mu.Unlock()
	return te.stopped
}

func (te *triggerEngine) State() string {
	te.mu.Lock()
	defer te.mu.Unlock()
	return te.state
}

// Active reports whether a trigger signal is configured. The rings must stay
// populated from that moment on: a capture reaches back over the pre-trigger
// window, so waiting until the trigger arms would leave that window empty.
func (te *triggerEngine) Active() bool {
	te.mu.Lock()
	defer te.mu.Unlock()
	return te.baseKey != ""
}

// baseSignalKey is the configured trigger signal without its "[i]" suffix, or
// "" when no trigger signal is set.
func (te *triggerEngine) baseSignalKey() string {
	te.mu.Lock()
	defer te.mu.Unlock()
	return te.baseKey
}

// bufGrowthIntervalSec is the shortest baseline the span growth is measured
// over. The hub refreshes 30 times a second and the span moves in steps as
// batches land, so a shorter baseline measures the batching, not the trend.
const bufGrowthIntervalSec = 0.5

// bufGrowthSmooth is the weight of a new growth measurement in the running
// estimate.
const bufGrowthSmooth = 0.5

// setBuffered records how far back the trigger signal's ring reaches, at wall
// clock now, and derives how fast that is growing. coverage is the maximum
// span (seconds) the ring can reach at its current bucket/capacity; archived
// says the disk history already spans the trigger window. Pass known=false when
// there is no ring to measure.
func (te *triggerEngine) setBuffered(span, coverage float64, archived, known bool, now float64) {
	te.mu.Lock()
	defer te.mu.Unlock()
	if !known {
		te.bufKnown, te.bufRateOK = false, false
		te.bufCoverage, te.bufArchived = 0, false
		return
	}
	if !te.bufKnown {
		te.bufKnown = true
		te.bufRefSpan, te.bufRefWall = span, now
	}
	te.bufSpan = span
	te.bufCoverage = coverage
	te.bufArchived = archived
	dt := now - te.bufRefWall
	if dt < bufGrowthIntervalSec {
		return
	}
	g := (span - te.bufRefSpan) / dt
	// A ring that is not full grows one second of span per second; one that is
	// full grows by whatever its incoming samples free up. Neither can exceed 1,
	// and a shrinking ring is simply not growing.
	if g < 0 {
		g = 0
	} else if g > 1 {
		g = 1
	}
	if te.bufRateOK {
		g = te.bufGrowth + bufGrowthSmooth*(g-te.bufGrowth)
	}
	te.bufGrowth, te.bufRateOK = g, true
	te.bufRefSpan, te.bufRefWall = span, now
}

// fillNeedLocked is how far back the buffer must reach before an edge may be
// accepted, so that the capture is still whole when it is harvested a
// post-window later.
//
// What has to hold at harvest time is that the buffer spans the whole window:
// its newest sample is then trigTime+post, so anything less has lost the front
// of the capture. The buffer keeps filling while the post-window is collected,
// though, so the shortfall it may start with is exactly what it will make up in
// that time — measured, not assumed:
//
//	need = windowSec − growth × postSec, floored at the pre-trigger window
//
// A ring that is still filling grows a second per second, which reduces this to
// the pre-trigger window: everything after the trigger is yet to be recorded
// anyway. A full one grows only as fast as its incoming samples free space —
// re-bucketing to a longer window replaces dense old samples with sparse new
// ones — and it is that case, growth well below 1, where firing on the
//
// Two escapes keep an armed trigger from staying deaf forever:
//
//   - archived  — the disk history already spans the window, so the front of a
//     capture can be back-filled from it; the ring only needs to
//     hold the pre-window worth of recent data.
//   - coverage  — never demand more than the ring can physically reach. If its
//     coverage saturates below the window (a measured source rate
//     that over-estimates the true one), the gate opens once the
//     ring is full anyway and a short capture is delivered instead
//     of deafness.
func (te *triggerEngine) fillNeedLocked() float64 {
	pre := te.cfg.windowSec * te.cfg.prePercent / 100
	growth := 0.0 // until measured, assume the buffer will not fill on its own
	if te.bufRateOK {
		growth = te.bufGrowth
	}
	need := te.cfg.windowSec - growth*(te.cfg.windowSec-pre)
	if need < pre {
		need = pre
	}
	if te.bufArchived {
		// The archive back-fills the front; the ring holds the post-trigger
		// window live, so the pre-window is all it needs to have reached.
		return pre
	}
	if te.bufCoverage > 0 && need > te.bufCoverage {
		need = te.bufCoverage
	}
	return need
}

// fillLocked is how much of that requirement is met, as a fraction in [0, 1].
// It is 1 whenever the gate does not apply: nothing needed, or no ring to
// measure.
func (te *triggerEngine) fillLocked() float64 {
	need := te.fillNeedLocked()
	if need <= 0 || !te.bufKnown || te.bufSpan >= need*(1-shortCaptureTol) {
		return 1
	}
	if te.bufSpan <= 0 {
		return 0
	}
	return te.bufSpan / need
}

// latchWindowLocked freezes the pre/post split at fire time so later config
// edits do not change how the capture is rendered.
func (te *triggerEngine) latchWindowLocked(t float64) {
	te.state = trigCollecting
	te.trigTime = t
	te.firedPre = te.cfg.windowSec * te.cfg.prePercent / 100
	te.firedPost = te.cfg.windowSec - te.firedPre
	te.firedValid = true
	te.rearmAt = 0
}

// Force fires the trigger immediately at the most recent sample time (falling
// back to the current wall clock when no sample has been seen yet).
func (te *triggerEngine) Force() {
	te.mu.Lock()
	defer te.mu.Unlock()
	if te.state == trigCollecting {
		return
	}
	t := float64(time.Now().UnixNano()) / 1e9
	if te.lastTOK {
		t = te.lastT
	}
	te.latchWindowLocked(t)
}

// feed passes a batch of full-resolution samples for one signal to the FSM.
// key is the fully-prefixed "src:sig" name; nElem is the signal's element count
// so that an "[i]"-suffixed configuration can select a single column out of the
// flattened element-major batch.
func (te *triggerEngine) feed(key string, nElem int, t, v []float64) {
	if len(t) == 0 || len(t) != len(v) {
		return
	}
	te.mu.Lock()
	defer te.mu.Unlock()
	if key != te.baseKey {
		return
	}
	te.lastT = t[len(t)-1]
	te.lastTOK = true
	te.lastFeedWall = float64(time.Now().UnixNano()) / 1e9

	// A capture in flight does not stop the comparator; it only changes what an
	// edge does. See pendingT.
	inFlight := te.state == trigCollecting || te.state == trigTriggered
	if te.state != trigArmed && !inFlight {
		return
	}
	step, start := 1, 0
	if te.elemIdx >= 0 && nElem > 1 {
		if te.elemIdx >= nElem {
			return
		}
		step, start = nElem, te.elemIdx
	}
	// Hold off while the buffer does not reach back far enough. Firing now would
	// deliver a capture whose front is simply missing — the ring never held it —
	// which is what made the first shot after a window change come back short.
	// Track the level meanwhile, so the first edge once the buffer is deep
	// enough is still measured against the right previous sample.
	if !inFlight && te.fillLocked() < 1 {
		for i := start; i < len(v); i += step {
			te.prevValue, te.prevValid = v[i], true
		}
		return
	}
	// The earliest trigger point a new capture may take. The one in flight owns
	// everything up to the end of its own post-window, and the holdoff — a guard
	// against re-triggering on the ringing of the SAME event — is measured from
	// its trigger point too, so the two overlap rather than add.
	notBefore := math.Inf(1)
	if inFlight && te.firedValid {
		notBefore = te.trigTime + math.Max(te.firedPost, te.cfg.holdoffSec)
	}
	thr := te.cfg.threshold
	for i := start; i < len(t); i += step {
		if !te.prevValid {
			te.prevValue = v[i]
			te.prevValid = true
			continue
		}
		up := te.prevValue < thr && v[i] >= thr
		down := te.prevValue > thr && v[i] <= thr
		te.prevValue = v[i]
		fired := false
		switch te.cfg.edge {
		case "falling":
			fired = down
		case "both":
			fired = up || down
		default:
			fired = up
		}
		if !fired {
			continue
		}
		if !inFlight {
			te.latchWindowLocked(t[i])
			return
		}
		// Keep the FIRST qualifying edge and go on tracking the level: a later
		// one would be no more use, and stopping here would leave prevValue
		// stale by the time the FSM rearms.
		if !te.pendingValid && t[i] >= notBefore {
			te.pendingT, te.pendingValid = t[i], true
		}
	}
}

// dueCapture reports whether a collecting trigger's post-window has elapsed and
// returns the latched window.
//
// The window is measured on the sample clock, not the wall clock: trigTime is a
// sample timestamp, and a stream whose timestamps lag real time (a busy
// producer, a buffered link) would otherwise be cut short by exactly that lag —
// an 8 s lag turned a 60 s window into a 36 s capture. Waiting for the samples
// themselves also means the ring really holds the window by the time it is read.
func (te *triggerEngine) dueCapture(nowSec float64) (trigTime, pre, post float64, ok bool) {
	te.mu.Lock()
	defer te.mu.Unlock()
	if te.state != trigCollecting || !te.firedValid {
		return 0, 0, 0, false
	}
	deadline := te.trigTime + te.firedPost + captureMarginSec
	switch {
	case te.lastTOK && te.lastT >= deadline:
	// The samples have covered the window.
	case !te.lastTOK && nowSec >= deadline:
		// No sample ever seen, so trigTime came from the wall clock (Force).
	case te.lastFeedWall > 0 && nowSec-te.lastFeedWall >= captureStallSec:
		// The stream has dried up; deliver what was collected rather than
		// leaving the client stuck in "collecting" forever.
	default:
		return 0, 0, 0, false
	}
	return te.trigTime, te.firedPre, te.firedPost, true
}

// markTriggered completes a capture and schedules the automatic rearm when the
// engine runs in "normal" mode.
func (te *triggerEngine) markTriggered(nowSec float64) {
	te.mu.Lock()
	if te.state == trigCollecting {
		te.state = trigTriggered
		if te.cfg.mode != "single" && !te.stopped {
			te.rearmAt = nowSec + te.cfg.holdoffSec
		}
	}
	te.mu.Unlock()
}

// dueRearm reports whether a pending automatic rearm has come due, consuming it.
func (te *triggerEngine) dueRearm(nowSec float64) bool {
	te.mu.Lock()
	defer te.mu.Unlock()
	if te.state != trigTriggered || te.rearmAt == 0 || nowSec < te.rearmAt {
		return false
	}
	te.rearmAt = 0
	return !te.stopped
}

// stateUnsent reports whether the FSM has moved since the last stateMsg was
// built, i.e. whether clients still have to be told.
func (te *triggerEngine) stateUnsent() bool {
	te.mu.Lock()
	defer te.mu.Unlock()
	if te.state != te.sentState {
		return true
	}
	// An armed trigger waiting for its buffer is otherwise indistinguishable
	// from one that is ignoring edges, so the filling itself is news. Coarse
	// steps only: this is checked 30 times a second.
	if te.state == trigArmed {
		f := te.fillLocked()
		return math.Abs(f-te.sentFill) >= 0.02 || (f >= 1 && te.sentFill < 1)
	}
	return false
}

// stateMsg builds the JSON "triggerState" broadcast for the current FSM state.
func (te *triggerEngine) stateMsg() []byte {
	te.mu.Lock()
	te.sentState = te.state
	te.sentFill = te.fillLocked()
	m := map[string]any{
		"type":    "triggerState",
		"state":   te.state,
		"mode":    te.cfg.mode,
		"stopped": te.stopped,
	}
	if te.state == trigArmed && te.sentFill < 1 {
		// Armed but holding off: the buffer does not yet reach back far enough
		// to deliver the window, so edges are being ignored on purpose.
		m["bufferFill"] = te.sentFill
		m["bufferNeedSec"] = te.fillNeedLocked()
	}
	if te.firedValid {
		// The window latched at fire time. Clients draw the filling capture on
		// this axis before the v2 frame arrives, and config edits between arm
		// and fire would otherwise leave them inferring the wrong window from
		// their own copy of the config.
		m["trigTime"] = te.trigTime
		m["preSec"] = te.firedPre
		m["postSec"] = te.firedPost
	}
	te.mu.Unlock()
	msg, _ := json.Marshal(m)
	return msg
}

/* ─── Hub integration ─────────────────────────────────────────────────────── */

// broadcastTriggerState pushes the current FSM state to every client.
func (h *Hub) broadcastTriggerState() {
	h.broadcast(h.trigger.stateMsg())
}

// handleTriggerCommand processes a trigger-related browser message. It returns
// false when the message type is not a trigger command.
func (h *Hub) handleTriggerCommand(t string, env map[string]interface{}) bool {
	switch t {
	case "setTrigger":
		cfg := h.trigger.Config()
		if s, ok := env["signal"].(string); ok {
			cfg.signalKey = s
		}
		if s, ok := env["edge"].(string); ok {
			cfg.edge = s
		}
		if s, ok := env["mode"].(string); ok {
			cfg.mode = s
		}
		if f, ok := env["threshold"].(float64); ok {
			cfg.threshold = f
		}
		if f, ok := env["windowSec"].(float64); ok {
			cfg.windowSec = f
		}
		if f, ok := env["prePercent"].(float64); ok {
			cfg.prePercent = f
		}
		if f, ok := env["holdoffSec"].(float64); ok {
			cfg.holdoffSec = f
		}
		h.trigger.SetConfig(cfg)
	case "arm", "rearm":
		h.trigger.Arm()
	case "disarm":
		h.trigger.Disarm()
	case "trigStop":
		stopped := !h.trigger.Stopped()
		if b, ok := env["stopped"].(bool); ok {
			stopped = b
		}
		h.trigger.SetStopped(stopped)
	case "forceTrigger":
		h.trigger.Force()
	default:
		return false
	}
	// Measure the buffer now rather than waiting for the next tick: ingest runs
	// on the source goroutine and a 1 MSps stream crosses the threshold many
	// times within one 33 ms tick, so an arm serviced here would otherwise fire
	// on a stale (or missing) measurement before the gate ever saw the new
	// configuration.
	h.refreshTriggerFill()
	h.broadcastTriggerState()
	return true
}

// refreshTriggerFill tells the FSM how far back the trigger signal's ring
// reaches, which is what lets an armed trigger hold off until a capture taken
// now would come back whole.
//
// The ring is the right yardstick even though a short capture is back-filled
// from the archive: the archive is sized for the same window and starts over
// whenever that window changes, so it holds no more of the stretch being waited
// for than the ring does. It can only add to what the capture finds.
//
// Called both from the push tick and from the client goroutine handling a
// trigger command; all the state it derives lives in the engine, behind the
// engine's lock.
func (h *Hub) refreshTriggerFill() {
	if h.trigger == nil {
		return
	}
	now := float64(time.Now().UnixNano()) / 1e9
	key := h.trigger.baseSignalKey()
	var rb *sigRing
	if key != "" {
		rb = h.getRing(key)
	}
	if rb == nil {
		// Nothing to measure. Do not gate on a signal the hub does not carry:
		// that would leave the trigger armed forever, which is worse than a
		// short capture.
		h.trigger.setBuffered(0, 0, false, false, now)
		return
	}
	_, span := rb.stats()
	// Maximum span the ring can ever reach at its current bucket/capacity, in
	// seconds. The gate must never demand more than this, or a ring whose
	// coverage is below the window (a measured source rate that over-estimates
	// the true one) can never satisfy it.
	coverage := 0.0
	if rate := rb.sourceRate(); rate > 0 {
		coverage = float64(ringCoverage(rb.bucketSize(), rb.capacity())) / rate
	}
	// If the disk archive already spans the trigger window, the front of a
	// short capture can be back-filled from it, so the ring need not cover the
	// whole window on its own.
	archived := h.hist.coversWindow(key, h.trigger.Config().windowSec)
	h.trigger.setBuffered(span, coverage, archived, true, now)
}

// triggerTick services the trigger FSM; called from Hub.Run() on every push tick.
func (h *Hub) triggerTick() {
	nowSec := float64(time.Now().UnixNano()) / 1e9

	h.retuneRings(nowSec)
	h.openPendingHistoryFiles(nowSec)
	h.refreshTriggerFill()

	if trigTime, pre, post, ok := h.trigger.dueCapture(nowSec); ok {
		if msg := h.buildTriggerCapture(trigTime, pre, post); msg != nil {
			dropped := 0
			for c := range h.clients {
				select {
				case c.send <- wsMessage{websocket.BinaryMessage, msg}:
				default:
					dropped++
				}
			}
			// A dropped capture is invisible to the user — the trigger fires,
			// the state goes to "triggered" and no waveform ever arrives — so
			// say so rather than leaving it to be guessed at.
			if dropped > 0 {
				log.Printf("wshub: trigger capture (%d B) dropped for %d client(s): send queue full",
					len(msg), dropped)
			}
		}
		h.trigger.markTriggered(nowSec)
		// A capture is only zoomable for as long as its samples still exist at
		// full resolution somewhere, and the rings roll past the window within
		// seconds of it being taken. Lift the window out of the archive into a
		// file of its own, where nothing overwrites it until the next trigger.
		h.hist.captureRange(trigTime-pre, trigTime+post)
	} else if h.trigger.dueRearm(nowSec) {
		h.trigger.rearm()
	}

	if h.trigger.stateUnsent() {
		h.broadcastTriggerState()
	}
}

// backfillCaptureHead prepends the front of [t0, t1] that the ring no longer
// holds, read from the disk archive. It returns its input unchanged when the
// ring already reaches t0, when history is off, or when the archive has nothing
// for that range.
//
// The rings are sized for the window, but they only have to *become* that long:
// they are min/max buckets that cover the configured window once they have
// rolled over completely at the current bucket, which takes as long as the
// window itself. Widen the window and arm, and the first captures ask for more
// history than the ring has ever stored — the frame then starts late and the
// user sees a blank front half. The archive is written straight through, at the
// geometry its file was created with, so unless that file was re-sized too it
// has kept the stretch the ring is still converging on.
func (h *Hub) backfillCaptureHead(key string, t0, t1 float64, st, sv []float64) ([]float64, []float64) {
	window := t1 - t0
	if !h.hist.enabled() || window <= 0 {
		return st, sv
	}
	gapEnd := t1
	if len(st) > 0 {
		gapEnd = st[0]
	}
	gap := gapEnd - t0
	if gap <= shortCaptureTol*window {
		return st, sv
	}
	// Budget the read by the share of the window being back-filled. The frame is
	// decimated to trigCapturePts either way, so a bigger read would buy nothing
	// but disk seeks — on the hub's own goroutine, between two push ticks.
	maxOut := int(float64(trigCapturePts)*gap/window) + 2
	ht, hv := h.hist.readRange(key, t0, gapEnd, maxOut)
	if len(ht) == 0 {
		return st, sv
	}
	// Drop anything at or past the ring's first sample: the two sources overlap
	// around the join, and the frame's timestamps must stay ascending.
	n := len(ht)
	if len(st) > 0 {
		n = sort.SearchFloat64s(ht, st[0])
	}
	if n == 0 {
		return st, sv
	}
	outT := make([]float64, 0, n+len(st))
	outV := make([]float64, 0, n+len(sv))
	outT = append(append(outT, ht[:n]...), st...)
	outV = append(append(outV, hv[:n]...), sv...)
	return outT, outV
}

// buildTriggerCapture extracts [trigTime-pre, trigTime+post] from every ring
// buffer and encodes the version-2 binary capture frame:
//
//	[u8 2][f64 trigTime][f64 preSec][f64 postSec][u32 nSig]
//	{[u16 keyLen][fullKey][u32 N][t f64×N][v f64×N]}
func (h *Hub) buildTriggerCapture(trigTime, pre, post float64) []byte {
	t0, t1 := trigTime-pre, trigTime+post

	type sigSlice struct {
		key  string
		t, v []float64
	}
	h.ringsMu.RLock()
	keys := make([]string, 0, len(h.rings))
	rings := make([]*sigRing, 0, len(h.rings))
	for k, rb := range h.rings {
		keys = append(keys, k)
		rings = append(rings, rb)
	}
	h.ringsMu.RUnlock()

	slices := make([]sigSlice, 0, len(keys))
	held := make(map[string]sigData, len(keys))
	total := 1 + 8 + 8 + 8 + 4
	for i, k := range keys {
		st, sv := rings[i].slice(t0, t1)
		st, sv = h.backfillCaptureHead(k, t0, t1, st, sv)
		if len(st) == 0 {
			continue
		}
		// Neither the ring nor the archive reached t0. Nothing can recover that
		// data, so name it rather than leaving the user to wonder why the front
		// of their window is blank.
		if lost := st[0] - t0; lost > shortCaptureTol*(t1-t0) {
			cnt, span := rings[i].stats()
			log.Printf("wshub: capture %s is short by %.2f s of %.2f s: ring holds %.2f s (%d pts, min/max over %d)",
				k, lost, t1-t0, span, cnt, rings[i].bucketSize())
		}
		// Take the second half of the double buffer here, before the frame is
		// decimated: the client gets 20 000 points to draw, but a zoom into
		// them has to come back with the underlying samples, and the rings will
		// have rolled past them by the time it is asked for.
		held[k] = sigData{T: st, V: sv}
		// Decimate before framing: a long window at a high sample rate is
		// hundreds of megabytes raw, which the send path would silently drop.
		// The min/max envelope keeps every peak in the window, so a glitch is
		// still on screen at the zoomed-out view that first shows it.
		st, sv = minMaxDecimate(st, sv, trigCapturePts)
		slices = append(slices, sigSlice{key: k, t: st, v: sv})
		total += 2 + len(k) + 4 + len(st)*16
	}
	if len(slices) == 0 {
		return nil
	}
	// Swap only now that the capture is known good. A shot that yielded nothing
	// must leave the previous window on screen rather than blanking it.
	h.capture.publish(t0, t1, held)

	buf := make([]byte, total)
	buf[0] = 2
	off := 1
	binary.LittleEndian.PutUint64(buf[off:], math.Float64bits(trigTime))
	off += 8
	binary.LittleEndian.PutUint64(buf[off:], math.Float64bits(pre))
	off += 8
	binary.LittleEndian.PutUint64(buf[off:], math.Float64bits(post))
	off += 8
	binary.LittleEndian.PutUint32(buf[off:], uint32(len(slices)))
	off += 4
	for _, s := range slices {
		binary.LittleEndian.PutUint16(buf[off:], uint16(len(s.key)))
		off += 2
		copy(buf[off:], s.key)
		off += len(s.key)
		binary.LittleEndian.PutUint32(buf[off:], uint32(len(s.t)))
		off += 4
		off = writeFloat64s(buf, off, s.t)
		off = writeFloat64s(buf, off, s.v)
	}
	return buf
}