package main import ( "context" "errors" "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) // owner 车道结构化旋钮(design/18 §11「乙 = C」)的落盘面:主人在群里说「少说点」 // 这类诉求被写成 $DATA_DIR/config/owner_lane.json(可读可改,面板后续照此写入)。 // 它只承载「说不说 / 多频繁」,不承载任何身份/权限信息。 ownerLane := runtimeconfig.NewOwnerLaneStore( runtimeconfig.DefaultOwnerLanePath(docDataDir), logger.WithSource(a.log, "services/channelbridge/ownerlane")) // 静默窗口合并(M23)的落盘面:$DATA_DIR/config/coalesce.json(零新 env 键, // design/13 §2.1)。它只承载**调优**(窗口/绝对上限/单批上限/给不给陌生人), // **不承载身份**,也**翻不开**主人私聊的 direct 车道(那是 actor 元数据决定的类型)。 coalesceStore := runtimeconfig.NewCoalesceStore( runtimeconfig.DefaultCoalescePath(docDataDir), logger.WithSource(a.log, "services/channelbridge/coalesce")) 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 的文档摄取层 // 同源:`/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 到来。 // OwnerLaneFor / OwnerLaneSet 是 owner 车道结构化旋钮(design/18 §11「乙 = C」) // 的读取/落盘接缝:读接口按 (platform, groupID) 返回该群生效策略(缺省 = 不对她说 // 就不开口),写接口把「少说点」这类结构化诉求落进 owner_lane.json,也是面板后续 // 写入的那一份。它只影响说不说/多频繁,绝不触碰权限与 actor 元数据。 OwnerLaneFor: func(platform, groupID string) *channelbridge.OwnerLanePolicy { p := ownerLane.PolicyFor(platform, groupID) return &channelbridge.OwnerLanePolicy{ ReplyToUnaddressed: p.ReplyToUnaddressed, MinReplyGapSec: p.MinReplyGapSec, } }, OwnerLaneSet: func(platform, groupID string, p channelbridge.OwnerLanePolicy) error { return ownerLane.Set(platform, groupID, runtimeconfig.OwnerLanePolicy{ ReplyToUnaddressed: p.ReplyToUnaddressed, MinReplyGapSec: p.MinReplyGapSec, }) }, // Coalesce 是**静默窗口合并**(M23)的注入面:按车道键取生效窗口参数。 // 车道类型不在这里决定 —— 主人私聊恒为 direct 车道(无窗口),由 actor 元数据 // 判定,配置无权翻转;这里只给 batch 车道(群聊/陌生人私聊)的窗口取值。 Coalesce: channelbridge.CoalesceOptions{ For: func(laneKey string) *runtimeconfig.CoalescePolicy { p := coalesceStore.PolicyFor(laneKey) return &p }, }, // 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, coalesce: coalesceStore, 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 // coalesce 是静默窗口合并(M23)的**同一个** store 实例:bridge 每次发车前从它 // 读生效值,因此面板保存后下一次发车即生效(热生效),无需重启。 coalesce *runtimeconfig.CoalesceStore log *slog.Logger } func (r *runtimeChannels) Path() string { return r.path } // CoalescePath 返回静默窗口配置文件(coalesce.json)的绝对落点,供面板如实回显。 func (r *runtimeChannels) CoalescePath() string { if r.coalesce == nil { return "" } return r.coalesce.Path() } // CoalescePresent 报告静默窗口配置文件是否已存在(面板据此显示「还没配置过」)。 func (r *runtimeChannels) CoalescePresent() bool { if r.coalesce == nil { return false } return r.coalesce.Present() } // CoalesceSnapshot 返回盘上那份静默窗口配置(未 clamp 的原值),供面板回显。 func (r *runtimeChannels) CoalesceSnapshot() runtimeconfig.CoalesceFile { if r.coalesce == nil { return runtimeconfig.CoalesceFile{} } return r.coalesce.Snapshot() } // ApplyCoalesce 应用一份三态补丁:内存与落盘在同一个临界区,桥下一次发车即读到新值 // (store 与 bridge 是**同一个**实例,因此不需要任何重启或重载)。 func (r *runtimeChannels) ApplyCoalesce(patch runtimeconfig.CoalescePatch) (bool, error) { if r.coalesce == nil { return false, errors.New("channel coalesce config store not wired") } wrote, err := r.coalesce.Apply(patch) if err != nil { return wrote, fmt.Errorf("apply coalesce patch: %w", err) } if wrote { r.log.Info("channel coalesce config saved (hot: the bridge reads it on the next departure)", "path", r.CoalescePath()) } return wrote, nil } 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 }