fix: enforce revoke disconnect, group pairing, and agent disable cascade

- Revoke paired device now force-closes the active WebSocket connection
  via pubsub (EventPairingRevoked → Server.DisconnectByPairing)
- Add group-level pairing policy ("pairing") to Discord, WhatsApp, and
  Zalo Personal channels (matching existing Telegram/Feishu pattern)
- Block inactive agents at resolver level (chat, cron, delegation all
  reject with "agent is inactive")
- Cascade: setting agent status to "inactive" auto-disables linked
  channel instances via EventAgentStatusChanged pubsub

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

cmd/gateway.go

@@ -10,6 +10,8 @@ import (
 	"path/filepath"
 	"syscall"
 
+	"github.com/google/uuid"
+
 	"github.com/nextlevelbuilder/goclaw/internal/agent"
 	"github.com/nextlevelbuilder/goclaw/internal/bootstrap"
 	"github.com/nextlevelbuilder/goclaw/internal/bus"
@@ -850,6 +852,61 @@ func runGateway() {
 		}
 	})
 
+	// Wire pairing revocation → force disconnect active WebSocket sessions.
+	msgBus.Subscribe(bus.TopicPairingRevoked, func(event bus.Event) {
+		if event.Name != bus.EventPairingRevoked {
+			return
+		}
+		payload, ok := event.Payload.(bus.PairingRevokedPayload)
+		if !ok {
+			return
+		}
+		go server.DisconnectByPairing(payload.SenderID, payload.Channel)
+	})
+
+	// Cascade: when an agent becomes inactive, disable its linked channel instances.
+	if managedStores != nil && managedStores.ChannelInstances != nil {
+		ciStore := managedStores.ChannelInstances
+		msgBus.Subscribe(bus.TopicAgentStatusChanged, func(event bus.Event) {
+			if event.Name != bus.EventAgentStatusChanged {
+				return
+			}
+			payload, ok := event.Payload.(bus.AgentStatusChangedPayload)
+			if !ok || payload.NewStatus != store.AgentStatusInactive {
+				return
+			}
+			go func() {
+				agentID, err := uuid.Parse(payload.AgentID)
+				if err != nil {
+					return
+				}
+				all, err := ciStore.ListAll(context.Background())
+				if err != nil {
+					slog.Warn("cascade disable: failed to list channel instances", "error", err)
+					return
+				}
+				disabled := 0
+				for _, inst := range all {
+					if inst.AgentID == agentID && inst.Enabled {
+						if err := ciStore.Update(context.Background(), inst.ID, map[string]any{"enabled": false}); err != nil {
+							slog.Warn("cascade disable: failed to disable channel instance", "name", inst.Name, "error", err)
+						} else {
+							disabled++
+						}
+					}
+				}
+				if disabled > 0 {
+					slog.Info("cascade disabled channel instances for inactive agent", "agent_id", payload.AgentID, "count", disabled)
+					// Trigger channel reload so disabled instances are stopped.
+					msgBus.Broadcast(bus.Event{
+						Name:    protocol.EventCacheInvalidate,
+						Payload: bus.CacheInvalidatePayload{Kind: bus.CacheKindChannelInstances},
+					})
+				}
+			}()
+		})
+	}
+
 	// Setup graceful shutdown
 	ctx, cancel := context.WithCancel(context.Background())
 	defer cancel()

cmd/gateway_methods.go

@@ -29,7 +29,7 @@ func registerAllMethods(server *gateway.Server, agents *agent.Router, sessStore
 
 	// Phase 2: Pairing (store created externally, shared with channel manager).
 	// OnApprove callback is set later by the caller after channel manager is created.
-	pairingMethods := methods.NewPairingMethods(pairingStore)
+	pairingMethods := methods.NewPairingMethods(pairingStore, msgBus)
 	pairingMethods.Register(router)
 
 	// Phase 2: Usage (queries SessionStore for real token data)

internal/agent/resolver.go

@@ -89,6 +89,10 @@ func NewManagedResolver(deps ResolverDeps) ResolverFunc {
 			return nil, fmt.Errorf("agent not found: %s", agentKey)
 		}
 
+		if ag.Status != store.AgentStatusActive {
+			return nil, fmt.Errorf("agent %s is inactive", agentKey)
+		}
+
 		// Resolve provider
 		provider, err := deps.ProviderReg.Get(ag.Provider)
 		if err != nil {

internal/bus/types.go

@@ -72,8 +72,29 @@ const (
 	TopicCacheMCP              = "cache:mcp"
 	TopicChannelStreaming      = "channel-streaming"
 	TopicConfigChanged         = "config:changed"
+	TopicPairingRevoked        = "pairing:revoked"
+	TopicAgentStatusChanged    = "agent:status_changed"
 )
 
+// EventPairingRevoked is the event name broadcast when a paired device is revoked.
+const EventPairingRevoked = "pairing.revoked"
+
+// PairingRevokedPayload identifies the revoked device.
+type PairingRevokedPayload struct {
+	SenderID string `json:"sender_id"`
+	Channel  string `json:"channel"`
+}
+
+// EventAgentStatusChanged is broadcast when an agent's status changes (e.g., active → inactive).
+const EventAgentStatusChanged = "agent.status_changed"
+
+// AgentStatusChangedPayload carries agent status transition info for cascade operations.
+type AgentStatusChangedPayload struct {
+	AgentID   string `json:"agent_id"`
+	OldStatus string `json:"old_status"`
+	NewStatus string `json:"new_status"`
+}
+
 // CacheInvalidatePayload signals cache layers to evict stale entries.
 // Used with protocol.EventCacheInvalidate events.
 type CacheInvalidatePayload struct {

internal/channels/discord/discord.go

@@ -30,6 +30,7 @@ type Channel struct {
 	typingCtrls     sync.Map // channelID string → *typing.Controller
 	pairingService  store.PairingStore
 	pairingDebounce sync.Map // senderID → time.Time
+	approvedGroups  sync.Map // chatID → true (in-memory cache for paired groups)
 	groupHistory    *channels.PendingHistory
 	historyLimit    int
 }
@@ -240,7 +241,7 @@ func (c *Channel) handleMessage(_ *discordgo.Session, m *discordgo.MessageCreate
 			return
 		}
 	} else {
-		if !c.CheckPolicy("group", "", c.config.GroupPolicy, senderID) {
+		if !c.checkGroupPolicy(senderID, channelID) {
 			slog.Debug("discord group message rejected by policy",
 				"user_id", senderID,
 				"username", senderName,
@@ -366,6 +367,37 @@ func (c *Channel) handleMessage(_ *discordgo.Session, m *discordgo.MessageCreate
 	}
 }
 
+// checkGroupPolicy evaluates the group policy for a sender, with pairing support.
+func (c *Channel) checkGroupPolicy(senderID, channelID string) bool {
+	groupPolicy := c.config.GroupPolicy
+	if groupPolicy == "" {
+		groupPolicy = "open"
+	}
+
+	switch groupPolicy {
+	case "disabled":
+		return false
+	case "allowlist":
+		return c.IsAllowed(senderID)
+	case "pairing":
+		if c.IsAllowed(senderID) {
+			return true
+		}
+		if _, cached := c.approvedGroups.Load(channelID); cached {
+			return true
+		}
+		groupSenderID := fmt.Sprintf("group:%s", channelID)
+		if c.pairingService != nil && c.pairingService.IsPaired(groupSenderID, c.Name()) {
+			c.approvedGroups.Store(channelID, true)
+			return true
+		}
+		c.sendPairingReply(groupSenderID, channelID)
+		return false
+	default: // "open"
+		return true
+	}
+}
+
 // checkDMPolicy evaluates the DM policy for a sender, handling pairing flow.
 func (c *Channel) checkDMPolicy(senderID, channelID string) bool {
 	dmPolicy := c.config.DMPolicy

internal/channels/whatsapp/whatsapp.go

@@ -32,6 +32,7 @@ type Channel struct {
 	cancel          context.CancelFunc
 	pairingService  store.PairingStore
 	pairingDebounce sync.Map // senderID → time.Time
+	approvedGroups  sync.Map // chatID → true (in-memory cache for paired groups)
 }
 
 // New creates a new WhatsApp channel from config.
@@ -222,7 +223,7 @@ func (c *Channel) handleIncomingMessage(msg map[string]interface{}) {
 			return
 		}
 	} else {
-		if !c.CheckPolicy("group", "", c.config.GroupPolicy, senderID) {
+		if !c.checkGroupPolicy(senderID, chatID) {
 			slog.Debug("whatsapp group message rejected by policy", "sender_id", senderID)
 			return
 		}
@@ -266,6 +267,37 @@ func (c *Channel) handleIncomingMessage(msg map[string]interface{}) {
 	c.HandleMessage(senderID, chatID, content, media, metadata, peerKind)
 }
 
+// checkGroupPolicy evaluates the group policy for a sender, with pairing support.
+func (c *Channel) checkGroupPolicy(senderID, chatID string) bool {
+	groupPolicy := c.config.GroupPolicy
+	if groupPolicy == "" {
+		groupPolicy = "open"
+	}
+
+	switch groupPolicy {
+	case "disabled":
+		return false
+	case "allowlist":
+		return c.IsAllowed(senderID)
+	case "pairing":
+		if c.IsAllowed(senderID) {
+			return true
+		}
+		if _, cached := c.approvedGroups.Load(chatID); cached {
+			return true
+		}
+		groupSenderID := fmt.Sprintf("group:%s", chatID)
+		if c.pairingService != nil && c.pairingService.IsPaired(groupSenderID, c.Name()) {
+			c.approvedGroups.Store(chatID, true)
+			return true
+		}
+		c.sendPairingReply(groupSenderID, chatID)
+		return false
+	default: // "open"
+		return true
+	}
+}
+
 // checkDMPolicy evaluates the DM policy for a sender, handling pairing flow.
 func (c *Channel) checkDMPolicy(senderID, chatID string) bool {
 	dmPolicy := c.config.DMPolicy

internal/channels/zalo/personal/channel.go

@@ -35,6 +35,7 @@ type Channel struct {
 	config          config.ZaloPersonalConfig
 	pairingService  store.PairingStore
 	pairingDebounce sync.Map // senderID -> time.Time
+	approvedGroups  sync.Map // groupID → true (in-memory cache for paired groups)
 	typingCtrls     sync.Map // threadID → *typing.Controller
 
 	mu       sync.RWMutex // protects sess and listener

internal/channels/zalo/personal/policy.go

@@ -100,6 +100,21 @@ func (c *Channel) checkGroupPolicy(senderID, groupID string, mentions []*protoco
 			slog.Debug("zalo_personal group message rejected by allowlist", "group_id", groupID)
 			return false
 		}
+
+	case "pairing":
+		if c.IsAllowed(groupID) {
+			// pass — allowlist bypass
+		} else if _, cached := c.approvedGroups.Load(groupID); cached {
+			// pass — already approved
+		} else {
+			groupSenderID := fmt.Sprintf("group:%s", groupID)
+			if c.pairingService != nil && c.pairingService.IsPaired(groupSenderID, c.Name()) {
+				c.approvedGroups.Store(groupID, true)
+			} else {
+				c.sendPairingReply(groupSenderID, groupID)
+				return false
+			}
+		}
 	}
 
 	// @mention gating: only process group messages that @mention the bot.

internal/gateway/client.go

@@ -29,8 +29,10 @@ type Client struct {
 	remoteAddr  string    // peer IP (extracted from proxy headers or RemoteAddr)
 
 	// Browser pairing state
-	pairingCode    string // 8-char code if pending approval
-	pairingPending bool   // true while waiting for admin approval
+	pairingCode     string // 8-char code if pending approval
+	pairingPending  bool   // true while waiting for admin approval
+	pairedSenderID  string // senderID used for browser pairing auth (for revocation lookup)
+	pairedChannel   string // channel used for pairing auth (e.g., "browser")
 }
 
 func NewClient(conn *websocket.Conn, server *Server, remoteIP string) *Client {

internal/gateway/methods/pairing.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"encoding/json"
 
+	"github.com/nextlevelbuilder/goclaw/internal/bus"
 	"github.com/nextlevelbuilder/goclaw/internal/gateway"
 	"github.com/nextlevelbuilder/goclaw/internal/store"
 	"github.com/nextlevelbuilder/goclaw/pkg/protocol"
@@ -16,11 +17,12 @@ type PairingApproveCallback func(ctx context.Context, channel, chatID string)
 // PairingMethods handles device.pair.request, device.pair.approve, device.pair.list, device.pair.revoke.
 type PairingMethods struct {
 	service   store.PairingStore
+	msgBus    *bus.MessageBus
 	onApprove PairingApproveCallback
 }
 
-func NewPairingMethods(service store.PairingStore) *PairingMethods {
-	return &PairingMethods{service: service}
+func NewPairingMethods(service store.PairingStore, msgBus *bus.MessageBus) *PairingMethods {
+	return &PairingMethods{service: service, msgBus: msgBus}
 }
 
 // SetOnApprove sets a callback that fires after a pairing is approved.
@@ -154,6 +156,17 @@ func (m *PairingMethods) handleRevoke(_ context.Context, client *gateway.Client,
 		return
 	}
 
+	// Broadcast revocation so the server can force-disconnect the active session.
+	if m.msgBus != nil {
+		m.msgBus.Broadcast(bus.Event{
+			Name: bus.EventPairingRevoked,
+			Payload: bus.PairingRevokedPayload{
+				SenderID: params.SenderID,
+				Channel:  params.Channel,
+			},
+		})
+	}
+
 	client.SendResponse(protocol.NewOKResponse(req.ID, map[string]interface{}{
 		"revoked": true,
 	}))

internal/gateway/router.go

@@ -114,6 +114,8 @@ func (r *MethodRouter) handleConnect(ctx context.Context, client *Client, req *p
 		client.role = permissions.RoleOperator
 		client.authenticated = true
 		client.userID = params.UserID
+		client.pairedSenderID = params.SenderID
+		client.pairedChannel = "browser"
 		slog.Info("browser pairing authenticated", "sender_id", params.SenderID, "client", client.id)
 		r.sendConnectResponse(client, req.ID)
 		return

internal/gateway/server.go

@@ -335,6 +335,25 @@ func (s *Server) BroadcastEvent(event protocol.EventFrame) {
 	}
 }
 
+// DisconnectByPairing force-closes WebSocket connections authenticated via the
+// given pairing senderID and channel. Called after revoking a paired device so
+// that the revoked client cannot continue operating with its old role.
+func (s *Server) DisconnectByPairing(senderID, channel string) {
+	s.mu.RLock()
+	var targets []*Client
+	for _, c := range s.clients {
+		if c.pairedSenderID == senderID && c.pairedChannel == channel {
+			targets = append(targets, c)
+		}
+	}
+	s.mu.RUnlock()
+
+	for _, c := range targets {
+		slog.Info("disconnecting revoked paired device", "client", c.id, "sender_id", senderID, "channel", channel)
+		c.conn.Close()
+	}
+}
+
 func (s *Server) registerClient(c *Client) {
 	s.mu.Lock()
 	defer s.mu.Unlock()

internal/http/agents.go

@@ -247,6 +247,20 @@ func (h *AgentsHandler) handleUpdate(w http.ResponseWriter, r *http.Request) {
 	h.emitCacheInvalidate(bus.CacheKindAgent, ag.AgentKey)
 	h.emitCacheInvalidate(bus.CacheKindBootstrap, id.String())
 
+	// Cascade: if status changed, broadcast so channel instances and cron jobs react.
+	if newStatus, ok := updates["status"].(string); ok && newStatus != ag.Status {
+		if h.msgBus != nil {
+			h.msgBus.Broadcast(bus.Event{
+				Name: bus.EventAgentStatusChanged,
+				Payload: bus.AgentStatusChangedPayload{
+					AgentID:   id.String(),
+					OldStatus: ag.Status,
+					NewStatus: newStatus,
+				},
+			})
+		}
+	}
+
 	writeJSON(w, http.StatusOK, map[string]string{"ok": "true"})
 }