felhom-agent/internal/poke/poke.go

// Package poke is the agent-plane immediate-sync LISTENER (Direction-2a,
// SPIKE-immediate-sync-transport-2026-07-16). It receives a CONTENTLESS UDP "sync now" poke
// relayed hub → ep0 forced-command → wg0-origin → here, and fires ONE immediate desired-state
// cycle (the hub control loop's out-of-band report trigger). This collapses a user-triggered
// agent-plane config change (e.g. a pbsdr descriptor) from the 15-min report cycle to the
// spike-measured ~0.42 s path — while the 15-min cycle remains the guarantee (a lost poke is
// harmless by construction).
//
// The security posture is the spike's, unchanged:
//   - BIND CONFINEMENT: the socket binds EXCLUSIVELY to the box's WireGuard /32 (10.77.0.x from
//     registered.json). Never 0.0.0.0, never the LAN interface — a datagram to the LAN address
//     reaches no socket. The only way in is over the tunnel, and WireGuard refuses to encrypt to a
//     /32 no registered peer owns (spike P1 EKEYREJECTED), so a poke can only originate from ep0.
//   - CONTENTLESS: the payload is ignored ENTIRELY — any datagram means only "tick now". A forged
//     or replayed poke costs at most one extra debounced tick; it carries no config, no auth
//     handshake, no version. There is nothing in a poke to trust.
//   - DEBOUNCE: a burst of pokes within DebounceWindow yields ≤1 extra tick (leading-edge). The
//     downstream trigger channel is itself cap-1 coalescing, so a tick already running collapses
//     further pokes into one follow-up — never N queued.
package poke

import (
	"context"
	"fmt"
	"log/slog"
	"net"
	"net/netip"
	"sync"
	"time"
)

// Port is the FIXED UDP port the listener binds on the box's WG address and the hub/ep0
// forced-command targets. It is a shared cross-repo contract (documented in REUSE.md and the ep0
// felhom-poke forced-command): change it in one place → change it in all three.
const Port = 51822

// DebounceWindow coalesces a burst of pokes into a single immediate tick (leading-edge). Two
// distinct operator changes inside one window still converge in one tick because desired-state is
// level-triggered (the tick fetches the LATEST state, not a delta).
const DebounceWindow = 2 * time.Second

// rebindInterval is how long the listener waits before re-attempting a bind after WG is
// unregistered/down or a serve loop ends.
const rebindInterval = 5 * time.Second

// readBufSize bounds a single datagram read. A poke is contentless; the buffer only needs to be
// large enough to drain whatever a (harmless) sender emits.
const readBufSize = 512

// Listener binds the poke socket and fires the report trigger on each (debounced) poke.
type Listener struct {
	// resolve returns the box's current WG address (ok=false until registered). Called before each
	// bind attempt so a late registration / rare re-address is picked up on the next cycle.
	resolve func() (netip.Addr, bool)
	// trigger is the non-blocking nudge into the hub control loop's out-of-band report channel.
	trigger func()
	port    int
	window  time.Duration
	logger  *slog.Logger

	mu       sync.Mutex
	lastFire time.Time
	local    netip.AddrPort // set once bound (for tests / logging)
	boundCh  chan struct{}  // closed once the first bind succeeds
	now      func() time.Time
}

// NewListener builds a listener. port<=0 uses the fixed Port. A nil trigger is a no-op (safe).
func NewListener(resolve func() (netip.Addr, bool), trigger func(), port int, logger *slog.Logger) *Listener {
	if logger == nil {
		logger = slog.Default()
	}
	if trigger == nil {
		trigger = func() {}
	}
	if port <= 0 {
		port = Port
	}
	return &Listener{
		resolve: resolve, trigger: trigger, port: port,
		window:  DebounceWindow,
		logger:  logger,
		boundCh: make(chan struct{}),
		now:     time.Now,
	}
}

// onPoke applies the leading-edge debounce and fires the trigger at most once per window. Returns
// true iff it fired (the test hook for the debounce red-proof).
func (l *Listener) onPoke() bool {
	l.mu.Lock()
	now := l.now()
	if !l.lastFire.IsZero() && now.Sub(l.lastFire) < l.window {
		l.mu.Unlock()
		return false
	}
	l.lastFire = now
	l.mu.Unlock()
	l.trigger()
	return true
}

// Run resolves the WG address and serves pokes until ctx is cancelled, rebinding across a WG
// down/re-address. It NEVER returns nil early (the daemon treats a returned goroutine as a reason
// to exit) — only ctx cancellation ends it.
func (l *Listener) Run(ctx context.Context) error {
	for {
		if err := ctx.Err(); err != nil {
			return err
		}
		addr, ok := l.resolve()
		if !ok {
			l.logger.Debug("poke: no WG address yet (box not registered) — will retry")
			if !l.sleep(ctx, rebindInterval) {
				return ctx.Err()
			}
			continue
		}
		if err := l.serve(ctx, addr); err != nil && ctx.Err() == nil {
			l.logger.Warn("poke: listener stopped; will rebind", "err", err, "wg_addr", addr.String())
			if !l.sleep(ctx, rebindInterval) {
				return ctx.Err()
			}
		}
	}
}

// serve binds EXCLUSIVELY to addr:port and delivers each datagram to onPoke until ctx or a read
// error ends it.
func (l *Listener) serve(ctx context.Context, addr netip.Addr) error {
	bind := netip.AddrPortFrom(addr, uint16(l.port)).String()
	var lc net.ListenConfig
	pc, err := lc.ListenPacket(ctx, "udp", bind)
	if err != nil {
		return fmt.Errorf("bind %s: %w", bind, err)
	}
	defer pc.Close()

	l.mu.Lock()
	if ap, ok := netip.ParseAddrPort(pc.LocalAddr().String()); ok == nil {
		l.local = ap
	}
	select {
	case <-l.boundCh: // already closed
	default:
		close(l.boundCh)
	}
	l.mu.Unlock()
	l.logger.Info("poke: listening for hub sync-pokes (WG-confined, contentless)", "addr", pc.LocalAddr().String())

	// Unblock ReadFrom on shutdown.
	go func() {
		<-ctx.Done()
		pc.Close()
	}()

	buf := make([]byte, readBufSize)
	for {
		_, from, err := pc.ReadFrom(buf)
		if err != nil {
			if ctx.Err() != nil {
				return ctx.Err()
			}
			return fmt.Errorf("read: %w", err)
		}
		// Contentless: the payload is discarded. ANY datagram = "tick now".
		if l.onPoke() {
			l.logger.Info("poke received → triggering an immediate desired-state cycle", "from", from.String())
		} else {
			l.logger.Debug("poke received but coalesced within the debounce window", "from", from.String())
		}
	}
}

// BoundAddrPort waits (bounded by ctx) until the socket is bound and returns its address. Test
// helper; production never needs it.
func (l *Listener) BoundAddrPort(ctx context.Context) (netip.AddrPort, bool) {
	select {
	case <-l.boundCh:
		l.mu.Lock()
		defer l.mu.Unlock()
		return l.local, true
	case <-ctx.Done():
		return netip.AddrPort{}, false
	}
}

func (l *Listener) sleep(ctx context.Context, d time.Duration) bool {
	t := time.NewTimer(d)
	defer t.Stop()
	select {
	case <-ctx.Done():
		return false
	case <-t.C:
		return true
	}
}