AskaEth
d81635672e
面板保存渠道配置后 Bridge.Reload 只重建了桥接器自己的适配器表，能力注册表里的条目仍指向保存前的旧实例；而管理面的渠道列表、每账号连接状态、open/close 的目标解析全部读注册表 —— 于是卡片永远「未连接」，点「连接」驱动的是旧实例（按旧 URL 拨号）。 生产日志实证：保存后 bridge 侧新实例连上 ws://localhost:3001 且心跳正常，而点连接时被驱动的适配器拨的是旧地址 ws://localhost:10301。 改法：注册表新增 upsert 语义的 ReplaceChannel（覆盖 Value、保留运行期 Policy；Register 遇重名会 ErrDuplicate，热重载必然同名）；桥接器经窄端口在适配器上线后同步，账号被移除时删除条目；组合根注入 a.caps 包装。同步失败只记日志、不中断 Reload；未接线时保持旧行为并只 WARN 一次。 验收：go build ./... + go vet 全绿；registry/channelbridge/gateway 三包 test 全绿；主会话亲自变异（摘掉 Policy 沿用 ⇒ TestReplaceChannelSwapsValueAndKeepsPolicy 红）。linux/darwin 交叉编译通过。
273 lines
13 KiB
Go
273 lines
13 KiB
Go
package main
|
||
|
||
import (
|
||
"context"
|
||
"fmt"
|
||
"log/slog"
|
||
"path/filepath"
|
||
"strings"
|
||
"time"
|
||
|
||
"git.yeij.top/AskaEth/Cyrene_v2/internal/capability/registry"
|
||
"git.yeij.top/AskaEth/Cyrene_v2/internal/channel/adapter"
|
||
"git.yeij.top/AskaEth/Cyrene_v2/internal/config"
|
||
"git.yeij.top/AskaEth/Cyrene_v2/internal/config/runtimeconfig"
|
||
"git.yeij.top/AskaEth/Cyrene_v2/internal/logger"
|
||
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/documents"
|
||
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/model"
|
||
"git.yeij.top/AskaEth/Cyrene_v2/internal/services/channelbridge"
|
||
"git.yeij.top/AskaEth/Cyrene_v2/internal/user"
|
||
"git.yeij.top/AskaEth/Cyrene_v2/pkg/plugins/sdk"
|
||
)
|
||
|
||
// ownerRegistrar 实现 gateway.OwnerRegistrar：它把一个渠道账号 id 持久化进主人
|
||
// 白名单（configs/users/owner.yaml），**并**热更新渠道 bridge 持有的活主人档案，
|
||
// 因此微信扫码登录的确认会被立即认成主人——无需重启，也无需手工改配置。path 与
|
||
// 活档案都在接线时捕获：启动加载的 *user.OwnerProfile 就是 bridge 判定 actor 时
|
||
// 依据的同一个指针，所以在它上面调 RegisterAccount 对 ResolveActor 立即可见，
|
||
// 无需触碰 bridge。
|
||
type ownerRegistrar struct {
|
||
path string
|
||
owner *user.OwnerProfile
|
||
log *slog.Logger
|
||
}
|
||
|
||
func (r *ownerRegistrar) RegisterChannelOwner(_ context.Context, platform, accountID string) error {
|
||
if strings.TrimSpace(accountID) == "" {
|
||
return nil
|
||
}
|
||
if err := user.RegisterOwnerAccount(r.path, platform, accountID); err != nil {
|
||
return fmt.Errorf("register owner %s:%s: %w", platform, accountID, err)
|
||
}
|
||
// 热生效：把这次登记扩到 bridge 已经指向的那份活档案上。若档案在启动时加载
|
||
// 失败（r.owner 为 nil），上面的落盘依然写进磁盘、重启后即被识别；此处打日志
|
||
// 是为了让运维知道热路径当时不可用。
|
||
if r.owner != nil {
|
||
r.owner.RegisterAccount(platform, accountID)
|
||
} else {
|
||
r.log.Warn("owner registrar: no live profile, whitelist persisted to disk only (restart to take effect)",
|
||
"platform", platform, "accountId", accountID)
|
||
}
|
||
r.log.Info("owner whitelist registered", "platform", platform, "accountId", accountID)
|
||
return nil
|
||
}
|
||
|
||
// LiveOwnerProfile 是 ownerRegistrar 对 gateway.liveOwnerProfile 可选扩展的实现（裁决
|
||
// 10.4.5）：管理面板 PUT /admin/owner-profile 保存后，gateway 需要拿到**同一个**活档案指针
|
||
// 做热替换——渠道 bridge 判定 actor 用的就是这个指针，所以热替换无需重启即生效。启动时档案
|
||
// 加载失败则为 nil，gateway 会退回「真实落盘 + 提示需重启」，绝不假装热生效。
|
||
func (r *ownerRegistrar) LiveOwnerProfile() *user.OwnerProfile { return r.owner }
|
||
|
||
// channelCaps 把能力注册表适配成 channelbridge.ChannelRegistry（热重载同步注册表的
|
||
// 窄端口）。替换时写入的形状与 wireChannels 启动注册**逐字相同**
|
||
// （Name/Kind/Description/Value）：Description 取适配器名、Value 取适配器本身 ——
|
||
// adminListChannels 把 Value 断言成 sdk.Channel、面板按注册名精确匹配，
|
||
// 形状一旦不一致，渠道就会从面板上消失。
|
||
type channelCaps struct{ caps *registry.Registry }
|
||
|
||
func (c channelCaps) ReplaceChannel(ch sdk.Channel) error {
|
||
if ch == nil {
|
||
return fmt.Errorf("wire channel registry: %w", registry.ErrInvalidArgument)
|
||
}
|
||
if err := c.caps.ReplaceChannel(ch); err != nil {
|
||
return fmt.Errorf("wire channel registry %q: %w", ch.Name(), err)
|
||
}
|
||
return nil
|
||
}
|
||
|
||
func (c channelCaps) RemoveChannel(name string) error {
|
||
if err := c.caps.Remove(name); err != nil {
|
||
return fmt.Errorf("wire channel registry %q: %w", name, err)
|
||
}
|
||
return nil
|
||
}
|
||
|
||
// wireChannels 依环境配置构造四个 IM 渠道适配器，把它们注册进能力注册表（管理
|
||
// API 据此枚举），并接进渠道 bridge——让已配置的适配器连上，把入站消息按
|
||
// session→actor→loop→reply 路由。缺凭据的适配器保持未连接（只记日志，绝不作
|
||
// 致命错误）：拿空配置去连要么失败，要么（weixin-ilink）触发阻塞式扫码登录。
|
||
func (a *app) wireChannels() error {
|
||
// 渠道配置可在运行时编辑：channels.json 存在且有效时以它为准，否则回落到 env
|
||
// 里的渠道块（design/04 §3.1）。工厂与配置读取读的是同一个文件，因此面板保存 +
|
||
// 热重载能让启动口径与运行口径保持一致。
|
||
channelsPath := runtimeconfig.DefaultChannelsPath(resolveDataDir(a.cfg))
|
||
factory := &channelFactory{path: channelsPath, fallback: channelAccountsFromEnv(), log: a.log}
|
||
adapters := buildChannelAdapters(factory)
|
||
for _, ch := range adapters {
|
||
capp := registry.Capability{
|
||
Name: ch.Name(),
|
||
Kind: registry.KindChannel,
|
||
Description: ch.Name(),
|
||
Value: ch,
|
||
}
|
||
if err := a.caps.Register(capp); err != nil {
|
||
a.log.Warn("channel register skipped", "channel", ch.Name(), "error", err)
|
||
}
|
||
}
|
||
|
||
// 入站附件的目录口径与文档摄取层（wire_runtime.go 的 docResolver）同源，
|
||
// 两处必须指向同一棵目录树，否则附件会落在一个地方、抽取去另一个地方找。
|
||
docDataDir := resolveDataDir(a.cfg)
|
||
bridge, berr := channelbridge.New(channelbridge.Options{
|
||
Log: logger.WithSource(a.log, "services/channelbridge"),
|
||
Loop: a.loop,
|
||
Sessions: a.sessions,
|
||
Owner: a.owner,
|
||
History: a.history,
|
||
QuietStartHour: a.cfg.Background.QuietStartHour,
|
||
QuietEndHour: a.cfg.Background.QuietEndHour,
|
||
// 入站附件摄取（design/24 §3；切片 M1）：IM 收到的受支持格式文件落进与
|
||
// Web 上传**同一个**存储核心（目录口径与 wire_runtime.go 的文档摄取层
|
||
// 同源：`<DataDir>/uploads/documents`），元数据也写进同一张 `documents`
|
||
// 表——`documents` 表的写入路径因此全仓只有一条（S7 的引用计数与 GC 有
|
||
// 单一事实来源）。未配置数据目录时写入面为 nil，桥接器走「未接线」的
|
||
// 诚实路径（模型知道有附件读不到），绝不假装读到了。
|
||
DocumentStore: documents.StoreConfig{
|
||
DataDir: docDataDir,
|
||
UploadsDir: filepath.Join(docDataDir, "uploads", "documents"),
|
||
Meta: documents.MetadataWriterFor(a.dbPath, a.log),
|
||
Log: a.log,
|
||
},
|
||
// EventStore 是渠道事件留存面（design/26 §4/§5；切片 E3+E4）：非消息事件
|
||
// （含未知类型与解析失败）由桥接层**无条件**写进它 —— T2/T3/unknown 逐条落行、
|
||
// T1 只累加计数；「进不进上下文」另由传入前审查决定，两者互不决定。
|
||
// 通道由 channelEventStore() 归一：未接线时交出**真 nil 接口**，桥接层据此走
|
||
// 内存计数降级（绝不因为缺存储而中断入站）。
|
||
EventStore: a.channelEventStore(),
|
||
// 可选的复审模型在接近单适配器上限时剔除低价值排队消息
|
||
// （design/12 §6 策略四）。适配器为 nil 时，闸门回落到 FIFO 驱逐。
|
||
Review: reviewFor(a.modelAdapter, logger.WithSource(a.log, "services/channelbridge/review")),
|
||
// Light 是 design/18 §3 的参与判定器：对触发命中的群聊/陌生人消息做一次
|
||
// 成本受控的轻量 LLM 决策（角色名提及 / 概率 / 活跃窗口）。只有真正触发
|
||
// 才把消息喂进群聊上下文。
|
||
Light: engageLightFor(a.modelAdapter, logger.WithSource(a.log, "services/channelbridge/engage")),
|
||
TriggerWords: a.defaultTriggerWords(),
|
||
PersonaPrompt: a.defaultInstructions(),
|
||
Engage: channelbridge.DefaultEngageConfig(),
|
||
// EngageFor 留空（P2a）；按群覆盖随 P2c 到来。
|
||
// Factory 让面板保存时热替换某平台的适配器而无需重启（design/04 §3.1）。
|
||
Factory: factory,
|
||
// Channels 让热重载把新适配器写回能力注册表：管理面板（渠道卡片、账号在线
|
||
// 状态、open/close 的目标解析）读的就是注册表，不同步它就只能看到旧实例，
|
||
// 于是卡片永远「未连接」、点「连接」驱动的是旧 URL 的那把。
|
||
Channels: channelCaps{caps: a.caps},
|
||
// ChannelPersona 从工厂构建所用的同一份配置里解析账号的渠道级角色
|
||
// （design/22 §2 方案 b）；bridge 按已连接的适配器缓存它，并在每次重载时刷新。
|
||
ChannelPersona: factory.Persona,
|
||
// VoiceSTT 把进程内语音链注入入站流水线（切片 IM-ASR；design/06 §1.1）：
|
||
// IM 语音在没有平台自带转写时由它在**流水线最前端**归一成文本（写回
|
||
// msg.Content），因此丢空判据 / 最小打扰判定 / 落库 / 模型可见消息看到的是同
|
||
// 一份文本。它在「静默档位」处**不会被当成成功**——语音链末端总是成功，所以
|
||
// bridge 用的是 `STTOutcome`（带档位与被掩盖的真实原因），并据此做临时熔断。
|
||
// a.voice 同时服务 Web 侧 WS 的 voice_input（同一条降级链，不另建一条）。
|
||
VoiceSTT: a.voice,
|
||
VoiceSTTTimeout: time.Duration(a.cfg.Voice.VoiceSTTTimeoutSec) * time.Second,
|
||
// VoiceTranscoder 把本地转码引擎交给入站语音副本链（design/25 §7.1 第 ④ 步；
|
||
// 裁决 11）：**不可播格式（微信/QQ 的 AMR）必须先转成可播副本再落盘**，否则
|
||
// 那条语音只能不落副本（`category=transcode_not_wired` 的诚实 WARN），客户端
|
||
// 便拿不到可回放的字节。注入的是 `wireTranscode` 装配的**同一个**引擎实例
|
||
// （与工具面的 `transcode` 工具同源），不是第二条降级链。
|
||
VoiceTranscoder: a.transcoder,
|
||
})
|
||
if berr != nil {
|
||
return fmt.Errorf("channel bridge: %w", berr)
|
||
}
|
||
for _, ch := range adapters {
|
||
bridge.Add(ch)
|
||
}
|
||
bridge.Start()
|
||
a.channelBridge = bridge
|
||
// 保留具体类型：gateway 只看到 ChannelConfigStore 接口，而启动播种管理员白名单
|
||
// 需要 concrete runtimeChannels 的方法。
|
||
rc := &runtimeChannels{
|
||
path: channelsPath,
|
||
fallback: factory.fallback,
|
||
bridge: bridge,
|
||
owner: a.owner,
|
||
log: a.log,
|
||
}
|
||
// 启动即灌一次每渠道管理员白名单（裁决 06.4.2）：不必等面板打开或热重载，
|
||
// channels.json 里声明的 adminIds 从第一帧入站消息起就参与判主。
|
||
rc.applyAdminIDs(rc.loadAccounts())
|
||
a.channels = rc
|
||
a.log.Info("channel adapters registered", "count", len(adapters))
|
||
return nil
|
||
}
|
||
|
||
// runtimeChannels 是「可运行时编辑的渠道配置」面向 gateway 的视图。Load 重读
|
||
// channels.json（文件优先 → env 兜底），Reload 委托给 bridge 的热替换，因此面板的
|
||
// 保存/重载路径驱动的正是 bridge 启动时的同一份来源。
|
||
type runtimeChannels struct {
|
||
path string
|
||
fallback map[string][]config.ChannelAccount
|
||
bridge *channelbridge.Bridge
|
||
// owner 是渠道 bridge 判 actor 用的**同一个**活档案指针：读配置时把每渠道管理员
|
||
// 白名单灌进它的 adminIDs 层（裁决 06.4.2），面板保存后无需重启即生效。
|
||
owner *user.OwnerProfile
|
||
log *slog.Logger
|
||
}
|
||
|
||
func (r *runtimeChannels) Path() string { return r.path }
|
||
|
||
func (r *runtimeChannels) Load() map[string][]config.ChannelAccount {
|
||
accounts := r.loadAccounts()
|
||
r.applyAdminIDs(accounts)
|
||
return accounts
|
||
}
|
||
|
||
// loadAccounts 只负责「读配置」：文件优先、env 兜底（design/04 §3.1）。
|
||
func (r *runtimeChannels) loadAccounts() map[string][]config.ChannelAccount {
|
||
accounts, warn := runtimeconfig.LoadChannels(r.path, r.fallback)
|
||
if warn != nil {
|
||
r.log.Warn("channels.json load fell back to env", "error", warn)
|
||
}
|
||
return accounts
|
||
}
|
||
|
||
// applyAdminIDs 把各平台声明的 adminIds 整体替换进活主人档案（裁决 06.4.2）。整体替换
|
||
// 语义保证热重载后已删除的管理员 id 立即失效；owner.yaml 的账号白名单不受影响。
|
||
func (r *runtimeChannels) applyAdminIDs(accounts map[string][]config.ChannelAccount) {
|
||
if r.owner == nil {
|
||
return
|
||
}
|
||
for platform, ids := range runtimeconfig.AdminIDs(accounts) {
|
||
r.owner.SetPlatformAdmins(platform, ids)
|
||
}
|
||
}
|
||
|
||
func (r *runtimeChannels) Reload(platform string) error {
|
||
// 先灌管理员白名单再重载适配器：面板一次保存（adminIds + 凭据）两条路径同时生效。
|
||
r.applyAdminIDs(r.loadAccounts())
|
||
return r.bridge.Reload(platform)
|
||
}
|
||
|
||
// reviewFor 构建渠道 bridge 的复审模型（design/12 §6 策略四）；无模型适配器可用时
|
||
// 返回 nil，让闸门回落到 FIFO 驱逐——绝不在适配器为 nil 时调用复审器（那会在
|
||
// .Chat 上 panic）。
|
||
func reviewFor(llm model.Adapter, log *slog.Logger) adapter.Reviewer {
|
||
if llm == nil {
|
||
return nil
|
||
}
|
||
return channelbridge.NewLLMReviewer(llm, log)
|
||
}
|
||
|
||
// engageLightFor 构建 design/18 §3 的参与判定器；无模型适配器可用时返回 nil，让
|
||
// bridge 的参与引擎朝「沉默」方向 fail-closed（触发命中但无判定器的消息绝不作答）。
|
||
func engageLightFor(llm model.Adapter, log *slog.Logger) channelbridge.LightJudge {
|
||
if llm == nil {
|
||
return nil
|
||
}
|
||
return channelbridge.NewLLMLightJudge(llm, log)
|
||
}
|
||
|
||
// defaultTriggerWords 返回默认人格的角色名提醒词（design/18 §2），让渠道 bridge
|
||
// 知道哪些群聊消息会引发 PROBING。
|
||
func (a *app) defaultTriggerWords() []string {
|
||
if c, ok := a.personas.DefaultCharacter(); ok {
|
||
if p, ok := a.personas.Persona(c.PersonaID); ok {
|
||
return append([]string(nil), p.TriggerWords...)
|
||
}
|
||
}
|
||
return nil
|
||
}
|