Cyrene_v2/internal/services/channelbridge/proactive.go

package channelbridge

import (
	"context"
	"fmt"
	"strings"

	"git.yeij.top/AskaEth/Cyrene_v2/internal/channel/adapter"
	"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/history"
	"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/turnreply"
	"git.yeij.top/AskaEth/Cyrene_v2/internal/user"
)

// proactiveTarget 是为一次主动推送解析出的主人 IM 地址
// （平台 + 账号 id）。账号 id 在每个适配器上都兼作私聊对端。
type proactiveTarget struct {
	platform  string
	accountID string
}

// proactiveRecentCount 是主动推送目标选择扫描的最近主会话消息条数，
// 用于找出「主人最近在哪里说话」，使推送落在她
// 实际在用的渠道上。编译期常量，运行路径上不会出现
// 魔法数字。
const proactiveRecentCount = 20

// SendProactive 通过该平台已注册的适配器把 msg 推送到主人的 IM 渠道。
// 它是 backthink 的 IMAdapter 接缝（design/12 §3 回退），
// 且只投递到一个主人渠道，由 proactiveTarget 中的单目标选择
// 决定：先看她显式的 [渠道:...] 选择，其次是她最近
// 使用的平台，最后是白名单默认（轮转，以免总是同一个渠道）。
// WeChat 是一等候选（它在白名单里，且连接后
// 最近活跃/回退选择可能选中它）。
//
// 推送经共享的 IM 气泡派发器（sendIMBubbles）出口，因此
// 多段消息会被拆成气泡、角色扮演/旁白被剥离、
// 相邻发送间隔 0.5-2s ——与回复行为一致，也是
// WeChat/其他 IM 主动推送不再以一整块未拆分的 <p> 到达的原因。
//
// 当主人档案没有白名单账号、或该账号没有已连接的
// 适配器支撑时，它返回明确错误，便于调用方记录原因，
// 而不是静默丢弃推送。
//
// 成功投递之后它**同时**做两件事：把这条主动消息镜像回主会话（恰好一次；气泡恒落的
// 根），并让实时通道能凭**结构化判据**认出「这条是她主动说的话」—— 后者是切片 P0b：
// `auto` 提示在主人有已连接 IM 渠道时优先走 IM，因此**实际最可能发生的路径**恰恰是这
// 一条；它过去只发普通 `message_persisted`，于是挂在 `proactive_message` 上的浏览器
// 通知永不触发（主人裁定：该不该提醒与走哪条路由无关）。
func (b *Bridge) SendProactive(ctx context.Context, msg adapter.UnifiedMessage) error {
	if err := ctx.Err(); err != nil {
		return err
	}
	target, ok := b.proactiveTarget(ctx, msg)
	if !ok {
		return fmt.Errorf("channelbridge: no connected owner IM channel for proactive push")
	}
	ad := b.adapterForPlatform(target.platform)
	if ad == nil {
		return fmt.Errorf("channelbridge: no adapter for owner platform %q", target.platform)
	}
	out := msg
	out.Platform = target.platform
	out.ChannelType = adapter.ChannelDirect
	// 主动推送的文本与回合回复走**同一份**段落包裹规整（同一函数、同一口径）：
	// 半包 <p> 的推送在 Web 历史里同样会塌成一整段，在 IM 侧同样只会得到
	// 一个气泡（splitBubbles 只认完整 <p> 块）。
	out.Content = turnreply.NormalizeParagraphs(msg.Content)
	// 主人的账号 id 在每个适配器上都作私聊对端：telegram/qq 读
	// ChannelID，kook/weixin 读 SenderID（ChannelID 作回退）。两者
	// 都设置，无论适配器偏好哪个字段，私聊消息都可投达，
	// 且不会重复发送。
	out.ChannelID = target.accountID
	out.SenderID = target.accountID
	// 推送经共享的 IM 气泡派发器出口，因此多段主动消息
	// 会被拆成气泡、角色扮演/旁白被剥离，并采用
	// 与回复相同的 0.5-2s 间隔——WeChat 主动推送不再作为
	// 一整条未拆分消息发送。切分吃的是**规整后**的同一份文本。
	if err := b.sendIMBubbles(ctx, out, func(ctx context.Context, msg adapter.UnifiedMessage) error {
		return ad.SendProactive(ctx, msg)
	}, out.Content); err != nil {
		return fmt.Errorf("channelbridge: send proactive via %q: %w", target.platform, err)
	}
	// 把主动消息镜像进主对话上下文（design/16
	// §4.4 + §4.6.1）。否则模型看不到自己主动说了什么，
	// 主人后续回复就会显得不连贯（该消息「丢失」了历史）。
	// 它用的是普通回复那条写入口径（persistMain，相同的列、相同的时间戳语义），
	// 作用域恒为 MainSessionID，落库的同样是**规整后**的那一份（Web 与 IM 看到的必须是
	// 同一份），因此落库行为与本切片之前逐字相同；变的只有通知的第二步（见下）。
	// turn_id 传 ""（零值）：主动推送是**她单方面开口**，没有对应的入站回合，也就没有
	// 「它属于哪一路回合」可归属 —— 空值诚实地表示这一点（见 turnid.go 的口径）。
	//
	// 通知分两步，两条都是**结构化**的通道（绝不是「像不像主动消息」的猜测）：
	//   ① notifyPersisted 照旧 —— 这就是 P0 起的行为，任何客户端都能渲染这条镜像消息，
	//      逐字不变（`message_persisted`）；
	//   ② 再走 notifyProactive —— 追加一条帧顶带 `proactive:true` 的同类帧，客户端据此在
	//      **非前台**补一发浏览器通知（切片 P0b）。两帧说的是同一条消息（同一个落库行
	//      id），前端按 id 去重 ⇒ 仍是一个气泡、仍只提醒一次。
	//
	// ② 只在**确实落了库**时发：`message_persisted` 的载荷要求权威 id，客户端拿不到 id
	// 会整帧丢弃 —— 发一条没有 id 的帧等于假装提醒过。落库失败本身已由 persistMain
	// 记 WARN（可见，不静默）。
	row := history.Message{
		SessionID: history.MainSessionID, Role: history.RoleAssistant, Content: out.Content,
		Platform: target.platform, DeviceID: proactiveDeviceID(target), AdapterID: target.platform, Mode: modeChat,
	}
	if full := b.persistMain(ctx, row, b.notifyPersisted); full.ID != "" {
		b.notifyProactive(full)
	}
	return nil
}

// proactiveDeviceID 是主动推送落进主会话时写入 device_id 的标签：`<平台>:<账号>:<账号>`。
// 它不是某台设备，而是「这条推送走的是哪个主人渠道账号」——与 sendIMBubbles 出口选中的
// 那个账号逐字对应，因此在管理日志里能一眼看出这条主动消息投到了哪里。
func proactiveDeviceID(target proactiveTarget) string {
	return fmt.Sprintf("%s:%s:%s", target.platform, target.accountID, target.accountID)
}

// proactiveTarget 解析主动推送应当到达的唯一主人账号。
// 候选范围是每个有已连接适配器支撑的白名单主人渠道
// （含 WeChat），选择遵循 backthink 的优先级：
//
//  1. 具体 msg.Platform（她显式的 [渠道:...] 选择）既在白名单
//     又已连接时，选中该渠道。
//  2. 优先选她在主会话中最近使用的渠道（白名单、
//     已连接平台上的主人/对话消息）。
//  3. 否则取白名单默认选中的平台，并按游标轮转，
//     使回退不会总是同一个渠道（例如总是 kook）。
//
// 主人为 nil、白名单为空，或白名单中没有任何
// 已连接适配器时，返回 !ok。
func (b *Bridge) proactiveTarget(ctx context.Context, msg adapter.UnifiedMessage) (proactiveTarget, bool) {
	accounts := b.ownerAccounts()
	if len(accounts) == 0 {
		return proactiveTarget{}, false
	}

	platform := strings.ToLower(strings.TrimSpace(msg.Platform))
	if !isGenericProactivePlatform(platform) {
		if acct, ok := matchOwnerAccount(accounts, platform); ok && b.hasConnectedAdapter(acct.Platform) {
			return proactiveTarget{acct.Platform, acct.AccountID}, true
		}
	}

	if acct, ok := b.recentActiveAccount(ctx, accounts); ok {
		return proactiveTarget{acct.Platform, acct.AccountID}, true
	}

	if acct, ok := b.nextConnectedAccount(accounts); ok {
		return proactiveTarget{acct.Platform, acct.AccountID}, true
	}

	return proactiveTarget{}, false
}

// ownerAccounts 按白名单顺序返回主人带账号 id 的白名单账号。
// 主人为 nil 或白名单为空时返回 nil。
func (b *Bridge) ownerAccounts() []user.OwnerAccount {
	if b.owner == nil {
		return nil
	}
	out := make([]user.OwnerAccount, 0, len(b.owner.Accounts))
	for _, a := range b.owner.Accounts {
		if a.AccountID != "" {
			out = append(out, a)
		}
	}
	if len(out) == 0 {
		return nil
	}
	return out
}

// matchOwnerAccount 返回平台与 platform 相等（忽略大小写）的
// 主人账号。
func matchOwnerAccount(accounts []user.OwnerAccount, platform string) (user.OwnerAccount, bool) {
	for _, a := range accounts {
		if strings.EqualFold(a.Platform, platform) {
			return a, true
		}
	}
	return user.OwnerAccount{}, false
}

// recentActiveAccount 返回最近一条主会话消息所对应的主人账号，
// 该消息的平台须是白名单且已连接的主人渠道——即「她在
// 那里活跃」的信号。历史存储为 nil、读取失败或尾部为空时
// 返回 ok=false，使选择退到默认值而不是瞎猜。
func (b *Bridge) recentActiveAccount(ctx context.Context, accounts []user.OwnerAccount) (user.OwnerAccount, bool) {
	if b.history == nil {
		return user.OwnerAccount{}, false
	}
	recent, err := b.history.Recent(ctx, history.MainSessionID, proactiveRecentCount)
	if err != nil || len(recent) == 0 {
		return user.OwnerAccount{}, false
	}
	// Recent 是最旧在前；从最新往回走，使最近的渠道胜出。
	for i := len(recent) - 1; i >= 0; i-- {
		p := strings.ToLower(strings.TrimSpace(recent[i].Platform))
		if p == "" || isGenericProactivePlatform(p) {
			continue
		}
		if acct, ok := matchOwnerAccount(accounts, p); ok && b.hasConnectedAdapter(acct.Platform) {
			return acct, true
		}
	}
	return user.OwnerAccount{}, false
}

// nextConnectedAccount 返回下一个有已连接适配器的白名单主人账号，
// 并推进游标，使无信号回退会轮转，而不是总落在
// 第一个白名单渠道（例如总是 kook）。它持有桥接器
// 锁，以原子地读取适配器 map 并更新游标。
func (b *Bridge) nextConnectedAccount(accounts []user.OwnerAccount) (user.OwnerAccount, bool) {
	b.mu.Lock()
	defer b.mu.Unlock()
	n := len(accounts)
	for i := 0; i < n; i++ {
		idx := (b.proactiveCursor + i) % n
		a := accounts[idx]
		if b.hasConnectedAdapterLocked(a.Platform) {
			b.proactiveCursor = (idx + 1) % n
			return a, true
		}
	}
	return user.OwnerAccount{}, false
}

// hasConnectedAdapter 报告该基础平台是否有已注册且当前已连接的
// 适配器，因此只有真能投递消息的渠道才会
// 被选中。
func (b *Bridge) hasConnectedAdapter(platform string) bool {
	b.mu.RLock()
	defer b.mu.RUnlock()
	return b.hasConnectedAdapterLocked(platform)
}

// hasConnectedAdapterLocked 是不加锁版本的 hasConnectedAdapter；
// 调用方须持有 b.mu（读锁或写锁）。
func (b *Bridge) hasConnectedAdapterLocked(platform string) bool {
	for name, a := range b.adapters {
		if strings.EqualFold(basePlatform(name), platform) && a.IsConnected() {
			return true
		}
	}
	return false
}

// isGenericProactivePlatform 报告 platform 是路由关键词还是
// 具体渠道 id。只有关键词才会退到最近活跃 / 默认
// 选择；具体 id 必须先解析为白名单且已连接的渠道
// 才被采纳，因此显式的渠道选择绝不会被静默替换成
// 另一个渠道。
func isGenericProactivePlatform(platform string) bool {
	switch platform {
	case "", "cyrene-backthink":
		return true
	default:
		return false
	}
}

// adapterForPlatform 返回基础平台与 platform 匹配的已注册适配器
// （多账号名如 "kook:3be..." 会归为其裸基础名），
// 优先取已连接的，使失效账号只在别无选择时才被尝试。
// 没有已注册的匹配项时返回 nil。
func (b *Bridge) adapterForPlatform(platform string) adapter.PlatformAdapter {
	b.mu.RLock()
	defer b.mu.RUnlock()
	var first adapter.PlatformAdapter
	for name, a := range b.adapters {
		if !strings.EqualFold(basePlatform(name), platform) {
			continue
		}
		if a.IsConnected() {
			return a
		}
		if first == nil {
			first = a
		}
	}
	return first
}