Cyrene_v2/internal/services/channelbridge/channelbridge.go

// Package channelbridge wires the registered IM channel adapters into the Cyrene
// runtime: it connects each configured adapter, installs the inbound message
// handler, and routes every normalized inbound message through the
// channel→session→actor→loop→reply pipeline (DevDocs/design/08 §0-§2).
//
// The bridge is deliberately generic over adapter.PlatformAdapter. It owns the
// per-platform Connect lifecycle and the inbound→reply path; the concrete
// adapters themselves (and their configuration, sourced from env) are assembled
// in cmd/cyrene/wire.go and handed in via Add. It never reads message text to
// decide identity — that is ResolveActor's job (actor metadata + owner whitelist,
// a project red line).
package channelbridge

import (
	"context"
	"errors"
	"log/slog"
	"math/rand"
	"net/http"
	"sync"
	"time"

	"git.yeij.top/AskaEth/Cyrene_v2/internal/channel/adapter"
	"git.yeij.top/AskaEth/Cyrene_v2/internal/channel/session"
	"git.yeij.top/AskaEth/Cyrene_v2/internal/logger"
	"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/history"
	"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/loop"
	"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/model"
	"git.yeij.top/AskaEth/Cyrene_v2/internal/user"
)

// webhookAdapter is the optional capability a channel adapter may expose to
// source an HTTP inbound callback. *webhook.Adapter and *kook.Adapter (webhook
// mode) implement it. The bridge reports these so the gateway can mount them.
type webhookAdapter interface {
	// InboundHandler returns the http.Handler the gateway mounts at the route to
	// receive external callbacks.
	InboundHandler() http.Handler
	// WebhookPath returns the route the gateway should mount the handler at. It
	// returns "" when the adapter is not currently exposing a webhook, letting the
	// bridge skip it (e.g. a websocket-mode KOOK).
	WebhookPath() string
}

// Knob constants. Compile-time so no bare magic number appears in the run path
// (project rule §10).
const (
	// connectTimeout bounds a single adapter Connect so a hung dial cannot stall
	// boot (a failed connect logs a warning and is skipped, never fatal).
	connectTimeout = 10 * time.Second
	// maxConcurrent bounds how many inbound messages run the agent loop at once,
	// so a burst of channel messages cannot spawn unbounded goroutines.
	maxConcurrent = 16
)

// ErrNilLoop is returned when the bridge is built without a chat runner.
var ErrNilLoop = errors.New("channelbridge: nil loop runner")

// ErrNilSessions is returned when the bridge is built without a session store.
var ErrNilSessions = errors.New("channelbridge: nil session store")

// Chatter runs a single agent turn. *loop.Loop is the production implementation;
// tests inject a stub.
type Chatter interface {
	Run(ctx context.Context, sessionID string, messages []model.Message) (*loop.Result, error)
}

// SessionStore resolves the channel↔session binding for an inbound message.
// *session.SessionManager is the production implementation.
type SessionStore interface {
	GetOrCreate(ctx context.Context, msg adapter.UnifiedMessage, owner *user.OwnerProfile) (session.Session, error)
}

// Options carries the bridge's injected dependencies. Loop and Sessions are
// required; Owner may be nil (then no account is recognized as owner, the safe
// default-deny posture from design/11 §0). History and OnPersisted are optional
// (design/16 §4.4): when a history store is wired, owner-private messages are
// routed into the main web session and tagged with the channel platform, and every
// persisted message is handed to OnPersisted so the gateway can broadcast a
// `message_persisted` (web chat) or `channel_persisted` (IM adapter log) event;
// when both are absent the bridge runs exactly as before (isolated sess-xxx, no
// chat_messages writes).
type Options struct {
	Log         *slog.Logger
	Loop        Chatter
	Sessions    SessionStore
	Owner       *user.OwnerProfile
	History     history.HistoryStore
	OnPersisted func(history.Message)
	// OnRejected, when non-nil, is called after a message is dropped by the
	// allow/deny black-or-white lists (design/16 §4.6 §6), so the composition root
	// can route it into an audit trail. It is invoked with the rejected message and
	// a short reason token ("group_denied", "user_not_allowed", ...). Nil = log only.
	//
	// It is also wired as the flow gate's OnPruned audit hook (design/12 §6 策略三/四),
	// so a message dropped by the rate limiter or the review prune reaches the same
	// audit trail with reason "rate_limited", "review_pruned" or "capped".
	OnRejected func(adapter.UnifiedMessage, string)
	// Review is the optional review model (design/12 §6 策略四) that prunes low-value
	// queued messages. Nil falls back to FIFO eviction at the total cap. Use
	// NewLLMReviewer to wrap a model.Adapter into this signature.
	Review adapter.Reviewer
	// Light is the optional light-response judge for non-owner group/stranger
	// messages (design/12 §6 策略二/五 + design/18 §3). When nil the bridge uses its
	// built-in rule-based judge and fails-closed to silence, so no chat message is sent
	// to an LLM just to decide whether to bother the owner. It is consulted when the
	// rule judge is inconclusive and the participation engine (design/18) has either
	// been triggered (role-name mention / probability) or is in an ACTIVE window. Use
	// NewLLMLightJudge to wrap a model.Adapter into a cost-controlled judge.
	Light LightJudge
	// TriggerWords are the current persona's role-name reminder words (design/18 §2).
	// A group message containing any one routes into participation PROBING even without
	// a direct @. They never imply identity (that stays structural). Empty keeps the
	// bridge on the legacy design/12 §6 light judgement.
	TriggerWords []string
	// PersonaPrompt is the current persona's identity/system-prompt block handed to the
	// participation judge so it can weigh the role's tone. It carries no owner secret.
	PersonaPrompt string
	// Engage is the participation-engine config (probability, activity window, cooldown,
	// context size). When its Enabled/Probability are zero AND TriggerWords is empty the
	// bridge keeps the legacy light judgement and never consults the engine. Use
	// DefaultEngageConfig for the design/18 defaults.
	Engage EngageConfig
	// EngageFor is an optional per-conversation override (design/18 §6 per-group knobs).
	// Nil keeps the base Engage config for every conversation (P2a).
	EngageFor func(platform, groupID string) *EngageConfig
	// EngageRand is the injectable [0,1) source for the probability trigger. Nil uses a
	// default math/rand source; tests inject a fixed source for determinism.
	EngageRand func() float64
	// QuietStartHour and QuietEndHour bound the do-not-disturb window (local clock
	// hours), shared with the backthink service so 策略五 reuses the same definition.
	// Equal non-zero hours disable the window; both zero pick the backthink defaults
	// (23-07). This is the price of keeping one source of truth across services.
	QuietStartHour int
	QuietEndHour   int
	// Now is the wall-clock source for quiet-hour judgement. Nil uses time.Now,
	// which makes time-dependent behaviour injectable in tests.
	Now func() time.Time
	// Factory rebuilds a platform's adapters from its effective account config
	// (channels.json → env fallback). When non-nil the bridge supports hot reload;
	// nil keeps the bridge boot-only (Reload returns ErrNilFactory).
	Factory PlatformFactory
}

// Bridge owns the channel adapter lifecycle and the inbound message pipeline. It
// is safe for concurrent use.
type Bridge struct {
	log         *slog.Logger
	run         Chatter
	sessions    SessionStore
	owner       *user.OwnerProfile
	history     history.HistoryStore
	onPersisted func(history.Message)
	onRejected  func(adapter.UnifiedMessage, string)
	review      adapter.Reviewer

	light          LightJudge
	quietStartHour int
	quietEndHour   int
	now            func() time.Time

	// Participation-engine state (design/18): per-conversation in-memory state machine
	// + the persona reminder words / engage knobs that gate it, and the injectable
	// randomness source for the probability trigger.
	triggerWords  []string
	personaPrompt string
	engage        EngageConfig
	engageFor     func(platform, groupID string) *EngageConfig
	engageRand    func() float64
	// bubbleGapFn is the injectable delay between IM outbound bubbles (0.5-2s),
	// defaulting to the package bubbleGap. Tests inject a fixed short gap so the
	// multi-bubble delay is asserted deterministically without sleeping a real 0.5-2s.
	bubbleGapFn  func() time.Duration
	engageMu     sync.Mutex
	engageGroups map[string]*engageGroupState

	mu          sync.RWMutex
	adapters    map[string]adapter.PlatformAdapter
	connected   map[string]bool
	readCancels map[string]context.CancelFunc
	started     bool
	closed      bool
	factory     PlatformFactory
	// proactiveCursor is the rotation cursor for the proactive-push no-signal
	// fallback (channelbridge/proactive.go), so the default channel is not always
	// the first whitelisted owner channel. Guarded by mu.
	proactiveCursor int

	// gateMu guards the per-adapter flow gates (design/12 §6 策略三/四). They are
	// built lazily on the first inbound message per account and are independent of
	// the adapter/connected lifecycle, so the same gate survives reconnects.
	gateMu sync.Mutex
	gates  map[string]*adapter.FlowGate

	ctx    context.Context
	cancel context.CancelFunc
	wg     sync.WaitGroup

	sem chan struct{}
}

// New builds a Bridge from opts. It returns an error when a required dependency
// is missing so the caller fails fast rather than silently dropping messages.
func New(opts Options) (*Bridge, error) {
	if opts.Loop == nil {
		return nil, ErrNilLoop
	}
	if opts.Sessions == nil {
		return nil, ErrNilSessions
	}
	log := opts.Log
	if log == nil {
		log = logger.New(logger.Options{Out: nil})
	}
	// Quiet hours default to the same window the backthink service uses, so the two
	// services never disagree about the do-not-disturb window (design/12 §6 策略五).
	qStart, qEnd := opts.QuietStartHour, opts.QuietEndHour
	if qStart == 0 && qEnd == 0 {
		qStart, qEnd = defaultQuietStartHour, defaultQuietEndHour
	}
	now := opts.Now
	if now == nil {
		now = time.Now
	}
	engageRand := opts.EngageRand
	if engageRand == nil {
		engageRand = func() float64 { return rand.Float64() }
	}
	return &Bridge{
		log:            logger.WithSource(log, "services/channelbridge"),
		run:            opts.Loop,
		sessions:       opts.Sessions,
		owner:          opts.Owner,
		history:        opts.History,
		onPersisted:    opts.OnPersisted,
		onRejected:     opts.OnRejected,
		review:         opts.Review,
		light:          opts.Light,
		quietStartHour: qStart,
		quietEndHour:   qEnd,
		now:            now,
		triggerWords:   opts.TriggerWords,
		personaPrompt:  opts.PersonaPrompt,
		engage:         opts.Engage,
		engageFor:      opts.EngageFor,
		engageRand:     engageRand,
		bubbleGapFn:    bubbleGap,
		engageGroups:   make(map[string]*engageGroupState),
		adapters:       make(map[string]adapter.PlatformAdapter),
		connected:      make(map[string]bool),
		readCancels:    make(map[string]context.CancelFunc),
		gates:          make(map[string]*adapter.FlowGate),
		factory:        opts.Factory,
		cancel:         func() {},
		sem:            make(chan struct{}, maxConcurrent),
	}, nil
}