Cyrene_v2/internal/services/channelbridge/flowgate.go

package channelbridge

import (
	"context"
	"encoding/json"
	"fmt"
	"log/slog"
	"sort"
	"strings"

	"git.yeij.top/AskaEth/Cyrene_v2/internal/channel/adapter"
	"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/model"
)

// gateFor returns the per-account flow gate for an adapter, building it lazily
// from the adapter's Config snapshot on first use. Gates are keyed by the adapter
// name so every account (platform:confId) keeps its own gate that survives adapter
// reconnects; **the throttling counters inside a gate are per conversation**
// (adapter.ConversationKey: platform+channelID+groupID/peerID), so two groups never
// squeeze each other (design/12 §6 策略三/四, 裁决 06.4.3).
func (b *Bridge) gateFor(a adapter.PlatformAdapter) *adapter.FlowGate {
	name := a.Name()
	b.gateMu.Lock()
	defer b.gateMu.Unlock()
	if g, ok := b.gates[name]; ok {
		return g
	}
	g := adapter.NewFlowGate(b.gateOptions(a))
	b.gates[name] = g
	return g
}

// gateOptions resolves the per-account flow-limits out of the adapter's Config
// snapshot into a FlowGate. An absent key (single-account env adapters) leaves the
// limit off, so existing channels run unchanged; a present key with value 0 follows
// the protective default (6 group/min, 50 cap) and a negative value disables it —
// the backward-compatible contract for CHANNELS_JSON accounts (design/12 §6 实现要点).
func (b *Bridge) gateOptions(a adapter.PlatformAdapter) adapter.FlowGateOptions {
	opts := adapter.FlowGateOptions{
		Review:   b.review,
		OnPruned: b.onRejected,
		Now:      b.now,
	}
	cfg := a.Config()
	if grp, ok := cfgInt(cfg, "groupMsgPerMin"); ok {
		opts.GroupMsgPerMin = flowOrDefault(grp, adapter.FlowLimitDefault)
	}
	if tot, ok := cfgInt(cfg, "msgPerMin"); ok && tot > 0 {
		opts.MsgPerMin = tot
	}
	if max, ok := cfgInt(cfg, "maxMessages"); ok {
		opts.MaxMessages = flowOrDefault(max, adapter.FlowCapDefault)
	}
	return opts
}

// flowOrDefault maps a configured limit value to an effective limit: a negative
// value disables the limit (0), zero follows the protective default, and a positive
// value pins it. This is the "old account without the field keeps prior behavior or
// gets the safe default" contract.
func flowOrDefault(v, def int) int {
	switch {
	case v < 0:
		return 0
	case v == 0:
		return def
	default:
		return v
	}
}

// cfgInt reads an integer field from an adapter config map, returning (0,false)
// when the key is absent or of an unexpected type (so an absent field stays "off").
func cfgInt(m map[string]any, key string) (int, bool) {
	if m == nil {
		return 0, false
	}
	switch v := m[key].(type) {
	case int:
		return v, true
	case int64:
		return int(v), true
	case float64:
		return int(v), true
	default:
		return 0, false
	}
}

const (
	// reviewMaxTokens bounds the review model output so a noisy queue listing never
	// produces a giant completion (design/12 §6 策略四 cost guard).
	reviewMaxTokens = 256
	// reviewSnippetLen is how many bytes of each message's content are shown to the
	// reviewer, enough to judge importance without leaking the whole queue.
	reviewSnippetLen = 80
	// reviewPrompt instructs the review model to pick which queued messages matter.
	// It only ever decides noise/importance — never who the sender is (project red
	// line: identity comes from actor metadata, not the LLM).
	reviewPrompt = "你是群聊消息去噪助手。下面列出若干进入队列的消息（每条带编号）。" +
		"请判断哪些消息没有保留价值（灌水、无关、低信息量），只返回你认为【有必要保留】的编号 JSON 数组。" +
		"只判断信息价值，不判断发送者身份；不要编造编号；只输出 JSON 数组。"
)

// NewLLMReviewer wraps a model.Adapter into an adapter.Reviewer (design/12 §6 策略四).
// It asks the LLM, via a light single-turn prompt, which of the queued messages are
// worth keeping and routes the answer back to the gate. The gate only invokes it
// when a pending set nears its cap, so the LLM is never called per-message.
func NewLLMReviewer(llm model.Adapter, log *slog.Logger) adapter.Reviewer {
	if log == nil {
		log = slog.Default()
	}
	return func(ctx context.Context, msgs []adapter.UnifiedMessage) ([]int, error) {
		resp, err := llm.Chat(ctx, model.Request{
			Messages: []model.Message{
				model.TextMessage(model.RoleSystem, reviewPrompt),
				model.TextMessage(model.RoleUser, buildReviewContent(msgs)),
			},
			Purpose:   "noise_review",
			MaxTokens: reviewMaxTokens,
		}, model.Options{})
		if err != nil {
			return nil, fmt.Errorf("flowgate: review model: %w", err)
		}
		keep, err := parseKeepIndices(resp.Content, len(msgs))
		if err != nil {
			log.Warn("flow gate review output unparsable, falling back to FIFO",
				"error", err)
			return nil, err
		}
		return keep, nil
	}
}

// buildReviewContent renders the queued messages as a numbered, truncated listing so
// the reviewer can judge importance cheaply. No identity or LLM judgement is made
// here — only a compact, index-labelled text summary.
func buildReviewContent(msgs []adapter.UnifiedMessage) string {
	var b strings.Builder
	for i, m := range msgs {
		name := senderDisplayName(m, m.Sender)
		if name == "" {
			name = "(未标注)"
		}
		fmt.Fprintf(&b, "[%d] %s: %s\n", i, name, truncateText(m.Content, reviewSnippetLen))
	}
	return b.String()
}

// truncateText shortens s to at most n bytes (by rune count) for display.
func truncateText(s string, n int) string {
	if n <= 0 {
		return ""
	}
	runes := []rune(s)
	if len(runes) <= n {
		return s
	}
	return string(runes[:n]) + "…"
}

// parseKeepIndices decodes a JSON array of indices from the review model and keeps
// only in-range, de-duplicated values in ascending order. Out-of-range or duplicate
// indices are dropped so a hallucinated number can never evict a real message.
func parseKeepIndices(s string, bound int) ([]int, error) {
	var idx []int
	if err := json.Unmarshal([]byte(s), &idx); err != nil {
		return nil, err
	}
	seen := make(map[int]bool, len(idx))
	out := make([]int, 0, len(idx))
	for _, i := range idx {
		if i >= 0 && i < bound && !seen[i] {
			seen[i] = true
			out = append(out, i)
		}
	}
	sort.Ints(out)
	return out, nil
}