feat: Phase F-2 — socket.io inbound + wakeup gate + token refresh

End-to-end Fabric inbound→Plexum→Fabric outbound now works against a live Fabric stack: alice posts in bt2-clean (Fabric REST) → guild emits message.created over socket.io → plugin's wakeup gate decides dispatch → notifications/plexum/channel/inbound to host → Plexum agent runs (echo provider) → outbound `send` tool posts via Fabric REST → fabrictester reply visible in channel internal/socketio/ (~280 LOC + 2 tests): - Minimal Engine.IO v4 + Socket.IO v5 client over websocket - WebSocket-only transport (skip polling upgrade dance) - AuthFunc callback re-evaluated on every (re)connect — fixes the stale-JWT-on-reconnect bug openclaw plugin documented for the JS client's single-shot auth, which the available Go socket.io library (zishang520) doesn't address either - PING/PONG per server-supplied interval - Caller-driven reconnect: Connect returns on close, supervisor re-dials with fresh token internal/tokens/ (~95 LOC + 9 tests): - Per-agent session cache with 8min TTL (matches openclaw's TOKEN_TTL_MS); guild tokens are ~15min so 8min keeps a margin - Invalidate forces re-login (used by inbound when CONNECT auth fires) - GuildToken helper picks the per-guild JWT from the cached session; if the guild is missing from the cache, invalidate + retry once internal/inbound/ (~290 LOC): - Supervisor: one socket.io conn per (agent, guild); reconnect with fresh token on drop; ChannelSyncInterval (60s) polling + push channel.joined/channel.left handlers - Wakeup gate: dm channels deliver any non-self message; other x_types require wakeup=true (record-only for non-wake non-dm deferred — Plexum has no history-injection equivalent in v1) - Self-author filter on selfUserId from cached session - Per-(agent,msgId) dedup bounded to 5000 entries - Per-channel serial queue with 5s idle drain so concurrent inbounds on the same channel run one-at-a-time (matches openclaw plugin) - Emits notifications/plexum/channel/inbound with session_id = "s_fab_<fabric_channel_id>" for stable per-channel session continuity cmd/plexum-fabric-channel-plugin: - Wires inbound supervisor at Init; runs in a background goroutine for the plugin's lifetime - Replaces F-1's sessions map with tokens.Cache (same warm-sessions behavior, now backed by TTL) - hostLogHandler: bridges slog records from inbound supervisor to HostAPI.Log notifications F-2 deferred to F-3+: - record-only history injection (Plexum v1 has no equivalent) - tools.ts port (15 MCP tools — channel/canvas/sub-discussion family) - presence-sync, command-sync, attachments, coalesce parity Tests: 22 (5 identity + 6 config + 9 tokens + 2 socketio). Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
2026-05-31 15:29:01 +01:00
parent f8d43ae70e
commit 0efcdfd342
8 changed files with 1162 additions and 36 deletions
--- a/internal/inbound/inbound.go
+++ b/internal/inbound/inbound.go
@@ -0,0 +1,438 @@
+// Package inbound is the Fabric → Plexum message pump. For each
+// (agent, guild) it maintains a long-lived socket.io connection to the
+// guild's /realtime namespace, joins every channel the agent has
+// membership in (resyncing periodically + on push events), and dispatches
+// matching message.created events through a per-channel serial queue.
+//
+// Wakeup gate matches openclaw's plugin: dispatch iff
+//   - m.xType == "dm" (any non-self message), OR
+//   - m.wakeup == true.
+//
+// Other inbounds (non-dm with wakeup != true) are dropped silently
+// (Plexum has no "record-only history injection" equivalent in v1;
+// future phase can add it).
+package inbound
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"log/slog"
+	"sync"
+	"time"
+
+	"git.hangman-lab.top/hzhang/Plexum-fabric-channel-plugin/internal/config"
+	"git.hangman-lab.top/hzhang/Plexum-fabric-channel-plugin/internal/fabric"
+	"git.hangman-lab.top/hzhang/Plexum-fabric-channel-plugin/internal/socketio"
+	"git.hangman-lab.top/hzhang/Plexum-fabric-channel-plugin/internal/tokens"
+)
+
+// ChannelSyncInterval is how often the supervisor diffs joined channels
+// against /api/channels to catch newly-membership-bestowed channels
+// that arrived without a push event. Matches openclaw's 60s.
+const ChannelSyncInterval = 60 * time.Second
+
+// ReconnectBackoff is the wait between reconnect attempts when a
+// socket drops + Connect returns. Keep small for snappier recovery;
+// exponential backoff is a future improvement.
+const ReconnectBackoff = 3 * time.Second
+
+// Notifier pushes one inbound message to the Plexum host. The plugin
+// main wires this to HostAPI.EmitNotification.
+type Notifier func(channelName, message, sessionID string)
+
+// Supervisor owns the per-(agent, guild) socket.io connections. Run
+// blocks until ctx is cancelled.
+type Supervisor struct {
+	Client       *fabric.Client
+	Tokens       *tokens.Cache
+	Bindings     []config.FabricBinding
+	ByFabric     config.ByFabricChannel // (guild_node, channel_id) → binding
+	AgentIDs     []string               // unique agents to bring up
+	Notify       Notifier
+	Logger       *slog.Logger
+
+	// Deduplicate (agentID, messageId) → already dispatched. Bounded
+	// to ~5000 entries (matches openclaw).
+	dedupMu sync.Mutex
+	seen    map[string]struct{}
+
+	// Per-channel serial chain: each channel id maps to a chan of
+	// queued task funcs. Channel goroutine drains them in order.
+	chainMu  sync.Mutex
+	chains   map[string]chan func()
+}
+
+// New constructs a Supervisor with deduped agentIDs derived from bindings.
+func New(client *fabric.Client, tokenCache *tokens.Cache, bindings []config.FabricBinding,
+	notify Notifier, logger *slog.Logger) *Supervisor {
+	agentSet := map[string]struct{}{}
+	for _, b := range bindings {
+		agentSet[b.AgentID] = struct{}{}
+	}
+	agents := make([]string, 0, len(agentSet))
+	for a := range agentSet {
+		agents = append(agents, a)
+	}
+	return &Supervisor{
+		Client: client, Tokens: tokenCache,
+		Bindings: bindings, ByFabric: config.Index(bindings),
+		AgentIDs: agents, Notify: notify, Logger: logger,
+		seen:   map[string]struct{}{},
+		chains: map[string]chan func(){},
+	}
+}
+
+// Run blocks: spawns one goroutine per (agent, guild) socket.io
+// supervisor. Each supervisor reconnects on drop until ctx cancels.
+// Returns when ctx is cancelled and all supervisors have exited.
+func (s *Supervisor) Run(ctx context.Context) error {
+	var wg sync.WaitGroup
+	for _, agentID := range s.AgentIDs {
+		sess, err := s.Tokens.Get(ctx, agentID)
+		if err != nil {
+			s.Logger.Warn("inbound: skip agent (no session)", "agent", agentID, "err", err)
+			continue
+		}
+		for _, g := range sess.Guilds {
+			if g.Endpoint == "" {
+				continue
+			}
+			wg.Add(1)
+			go func(agentID string, guild fabric.GuildInfo) {
+				defer wg.Done()
+				s.runAgentGuild(ctx, agentID, guild)
+			}(agentID, g)
+		}
+	}
+	wg.Wait()
+	return nil
+}
+
+// runAgentGuild keeps one socket.io connection alive for (agent, guild)
+// until ctx cancels. Reconnects with fresh auth on every drop.
+func (s *Supervisor) runAgentGuild(ctx context.Context, agentID string, guild fabric.GuildInfo) {
+	logger := s.Logger.With("agent", agentID, "guild", guild.NodeID)
+	for {
+		if err := ctx.Err(); err != nil {
+			return
+		}
+		err := s.connectOnce(ctx, agentID, guild, logger)
+		if ctx.Err() != nil {
+			return
+		}
+		errStr := "(nil)"
+		if err != nil {
+			errStr = err.Error()
+		}
+		logger.Warn("inbound: socket connection ended; reconnecting", "err", errStr)
+		select {
+		case <-time.After(ReconnectBackoff):
+		case <-ctx.Done():
+			return
+		}
+	}
+}
+
+// connectOnce opens one socket connection, runs the read loop, and
+// returns the terminating error (nil on graceful close).
+func (s *Supervisor) connectOnce(ctx context.Context, agentID string, guild fabric.GuildInfo, logger *slog.Logger) error {
+	authFn := func(authCtx context.Context) (map[string]any, error) {
+		// Always fetch a fresh token before the CONNECT — fixes the
+		// stale-JWT-on-reconnect bug openclaw documented.
+		s.Tokens.Invalidate(agentID)
+		tok, err := s.Tokens.GuildToken(authCtx, agentID, guild.NodeID)
+		if err != nil {
+			return nil, err
+		}
+		return map[string]any{"token": tok}, nil
+	}
+	cli, err := socketio.New(guild.Endpoint, "/socket.io/", "/realtime", authFn)
+	if err != nil {
+		return err
+	}
+
+	// Tracked subscriptions for this connection. Cleared on each
+	// reconnect (server forgets join_channel state).
+	joined := map[string]struct{}{}
+	var joinedMu sync.Mutex
+
+	// Initial sync + periodic resync runs in a goroutine; it reads
+	// channels from REST and emits join_channel for new + leave_channel
+	// for gone.
+	syncCtx, cancelSync := context.WithCancel(ctx)
+	defer cancelSync()
+	syncOnce := func(kind string) {
+		tok, err := s.Tokens.GuildToken(syncCtx, agentID, guild.NodeID)
+		if err != nil {
+			logger.Warn("inbound: sync token", "err", err)
+			return
+		}
+		channels, err := s.Client.ListChannels(syncCtx, guild.Endpoint, tok, guild.NodeID)
+		if err != nil {
+			logger.Warn("inbound: list channels", "err", err)
+			return
+		}
+		current := map[string]struct{}{}
+		for _, c := range channels {
+			current[c.ID] = struct{}{}
+		}
+		var added, removed int
+		joinedMu.Lock()
+		for id := range current {
+			if _, ok := joined[id]; !ok {
+				_ = cli.Emit(syncCtx, "join_channel", map[string]any{"channelId": id})
+				joined[id] = struct{}{}
+				added++
+			}
+		}
+		for id := range joined {
+			if _, ok := current[id]; !ok {
+				_ = cli.Emit(syncCtx, "leave_channel", map[string]any{"channelId": id})
+				delete(joined, id)
+				removed++
+			}
+		}
+		size := len(joined)
+		joinedMu.Unlock()
+		if kind == "initial" {
+			logger.Info("inbound: channels joined", "n", size)
+		} else if added > 0 || removed > 0 {
+			logger.Info("inbound: channels resync", "added", added, "removed", removed, "now", size)
+		}
+	}
+
+	// Event handlers. Registered BEFORE Connect so we don't miss the
+	// first events the server pushes right after handshake.
+	cli.On("connect", func([]json.RawMessage) { /* no-op; server CONNECT ack */ })
+
+	cli.On("channel.joined", func(args []json.RawMessage) {
+		var evt struct {
+			ChannelID string `json:"channelId"`
+		}
+		if len(args) == 0 {
+			return
+		}
+		_ = json.Unmarshal(args[0], &evt)
+		if evt.ChannelID == "" {
+			return
+		}
+		joinedMu.Lock()
+		if _, already := joined[evt.ChannelID]; already {
+			joinedMu.Unlock()
+			return
+		}
+		joined[evt.ChannelID] = struct{}{}
+		joinedMu.Unlock()
+		_ = cli.Emit(syncCtx, "join_channel", map[string]any{"channelId": evt.ChannelID})
+		logger.Info("inbound: channel.joined push", "channel", evt.ChannelID)
+	})
+
+	cli.On("channel.left", func(args []json.RawMessage) {
+		var evt struct {
+			ChannelID string `json:"channelId"`
+		}
+		if len(args) == 0 {
+			return
+		}
+		_ = json.Unmarshal(args[0], &evt)
+		if evt.ChannelID == "" {
+			return
+		}
+		joinedMu.Lock()
+		if _, present := joined[evt.ChannelID]; !present {
+			joinedMu.Unlock()
+			return
+		}
+		delete(joined, evt.ChannelID)
+		joinedMu.Unlock()
+		_ = cli.Emit(syncCtx, "leave_channel", map[string]any{"channelId": evt.ChannelID})
+		logger.Info("inbound: channel.left push", "channel", evt.ChannelID)
+	})
+
+	// Capture self user id from current cached session for the
+	// self-author filter.
+	sess := s.Tokens.Peek(agentID)
+	var selfUserID string
+	if sess != nil {
+		selfUserID = sess.User.ID
+	}
+
+	cli.On("message.created", func(args []json.RawMessage) {
+		var m FabricMessage
+		if len(args) == 0 {
+			return
+		}
+		if err := json.Unmarshal(args[0], &m); err != nil {
+			logger.Warn("inbound: bad message.created", "err", err)
+			return
+		}
+		s.dispatch(agentID, guild.NodeID, selfUserID, &m, logger)
+	})
+
+	// Periodic resync goroutine.
+	syncTimer := time.NewTicker(ChannelSyncInterval)
+	defer syncTimer.Stop()
+	go func() {
+		// Wait for connect to complete; we kick the initial sync from
+		// within Connect's startup. Use the ticker for the periodic.
+		for {
+			select {
+			case <-syncCtx.Done():
+				return
+			case <-syncTimer.C:
+				syncOnce("resync")
+			}
+		}
+	}()
+
+	// On connect, fire the initial sync. The socketio client doesn't
+	// fire a Go-side "connect" event before Connect returns; we
+	// schedule the initial sync to run shortly after Connect starts.
+	go func() {
+		time.Sleep(100 * time.Millisecond)
+		select {
+		case <-syncCtx.Done():
+			return
+		default:
+		}
+		syncOnce("initial")
+	}()
+
+	return cli.Connect(ctx)
+}
+
+// dispatch applies self-author filter, dedup, channel-binding lookup,
+// wakeup gate, then notifies Plexum.
+func (s *Supervisor) dispatch(agentID, guildNodeID, selfUserID string, m *FabricMessage, logger *slog.Logger) {
+	if m.ChannelID == "" {
+		return
+	}
+	// Self-author filter: skip messages this user wrote.
+	if m.AuthorUserID != "" && m.AuthorUserID == selfUserID {
+		return
+	}
+	// Dedup per (agent, messageId).
+	key := agentID + ":" + m.MessageID
+	s.dedupMu.Lock()
+	if _, dup := s.seen[key]; dup {
+		s.dedupMu.Unlock()
+		return
+	}
+	s.seen[key] = struct{}{}
+	if len(s.seen) > 5000 {
+		s.seen = map[string]struct{}{} // simple bounded reset
+	}
+	s.dedupMu.Unlock()
+
+	// Channel-binding lookup. We only deliver messages for channels
+	// the operator has bound to a Plexum agent in channels/*.json.
+	binding, ok := s.ByFabric[config.Key(guildNodeID, m.ChannelID)]
+	if !ok {
+		return
+	}
+	// The binding must point at THIS agent (a guild may have several
+	// agents; an inbound on agent A's socket for a channel bound to
+	// agent B is ignored — agent B has their own socket for it).
+	if binding.AgentID != agentID {
+		return
+	}
+
+	// Wakeup gate (decision #36-equivalent on the openclaw side):
+	//   dm channel → always deliver
+	//   other      → wakeup must be true
+	if m.XType != "dm" && !m.Wakeup {
+		logger.Debug("inbound: skip (no wakeup)",
+			"channel", m.ChannelID, "agent", agentID, "msg", m.MessageID)
+		return
+	}
+
+	logger.Info("inbound: dispatch",
+		"channel", m.ChannelID, "agent", agentID, "msg", m.MessageID, "xtype", m.XType)
+
+	// Per-channel serial queue. Concurrent inbounds on the same channel
+	// must run one at a time so Plexum's state machine sees clean turns.
+	sessionID := "s_fab_" + m.ChannelID
+	body := m.Content
+	plexumChannel := binding.PlexumChannelName
+	s.enqueueChannel(m.ChannelID, func() {
+		s.Notify(plexumChannel, body, sessionID)
+	})
+}
+
+// enqueueChannel chains a task onto the per-channel queue. If no chain
+// exists, spawn the drain goroutine. Drain goroutine exits when the
+// channel goes idle for >5s to avoid leaking goroutines per channel.
+func (s *Supervisor) enqueueChannel(channelID string, task func()) {
+	s.chainMu.Lock()
+	ch, ok := s.chains[channelID]
+	if !ok {
+		ch = make(chan func(), 32)
+		s.chains[channelID] = ch
+		go s.drainChannel(channelID, ch)
+	}
+	s.chainMu.Unlock()
+	select {
+	case ch <- task:
+	default:
+		// Backlog full (32+) — log + drop. Shouldn't happen unless
+		// the bound agent is hopelessly behind.
+		s.Logger.Warn("inbound: per-channel queue full; dropping",
+			"channel", channelID)
+	}
+}
+
+func (s *Supervisor) drainChannel(channelID string, ch chan func()) {
+	idle := time.NewTimer(5 * time.Second)
+	defer idle.Stop()
+	for {
+		select {
+		case task, open := <-ch:
+			if !open {
+				return
+			}
+			if !idle.Stop() {
+				<-idle.C
+			}
+			task()
+			idle.Reset(5 * time.Second)
+		case <-idle.C:
+			s.chainMu.Lock()
+			delete(s.chains, channelID)
+			s.chainMu.Unlock()
+			return
+		}
+	}
+}
+
+// FabricMessage is the wire shape of Fabric's message.created event.
+// Mirror of FabricMessage type in openclaw's inbound.ts.
+type FabricMessage struct {
+	MessageID    string             `json:"messageId"`
+	Seq          int                `json:"seq"`
+	Content      string             `json:"content"`
+	AuthorUserID string             `json:"authorUserId,omitempty"`
+	CreatedAt    string             `json:"createdAt,omitempty"`
+	ChannelID    string             `json:"channelId,omitempty"`
+	Attachments  []FabricAttachment `json:"attachments,omitempty"`
+	Wakeup       bool               `json:"wakeup,omitempty"`
+	// XType matches the Fabric channel's x-type field. 'dm' bypasses
+	// the wakeup gate.
+	XType string `json:"xType,omitempty"`
+}
+
+// FabricAttachment is the wire shape of one attachment ref. v1
+// (Phase F-2) doesn't download these; F-8 will.
+type FabricAttachment struct {
+	URL      string `json:"url"`
+	Name     string `json:"name,omitempty"`
+	MimeType string `json:"mimeType,omitempty"`
+}
+
+// Equality check used for tests.
+func (a FabricAttachment) Equal(b FabricAttachment) bool {
+	return a.URL == b.URL && a.Name == b.Name && a.MimeType == b.MimeType
+}
+
+// Sanity: ensure FabricMessage stays serializable.
+var _ = fmt.Sprint