From 7ae4298871de4dc5e49b62466b55c24e8cbd6482 Mon Sep 17 00:00:00 2001 From: AskaEth Date: Mon, 5 Oct 2026 13:26:15 +0800 Subject: [PATCH] =?UTF-8?q?feat(channels):=20=E9=99=8C=E7=94=9F=E4=BA=BA?= =?UTF-8?q?=E7=A7=81=E8=81=8A=E6=AD=A3=E5=B8=B8=E8=81=8A=E5=A4=A9=EF=BC=88?= =?UTF-8?q?=E8=B7=B3=E6=A6=82=E7=8E=87=E9=97=A8=EF=BC=89+=20=E5=BC=BA?= =?UTF-8?q?=E5=88=B6=E6=B5=81=E6=8E=A7=206=E6=9D=A1/=E5=88=86=E4=B8=8E?= =?UTF-8?q?=E5=9B=9E=E5=90=88=E6=A7=BD=E4=BD=8D=20+=20=E5=AE=89=E9=9D=99?= =?UTF-8?q?=E6=97=B6=E6=AE=B5=E4=B8=8D=E5=86=8D=E7=A1=AC=E9=9D=99=E9=BB=98?= =?UTF-8?q?=20+=20=E5=8F=97=E9=99=90=E8=BD=A6=E9=81=93=E8=BE=B9=E7=95=8C?= =?UTF-8?q?=E8=A1=8C=E5=89=8D=E7=BD=AE?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 陌生人私聊：不再走「值不值得回」的概率门，唯一闸门 = 每发送者 6 条/分滑窗 + 同会话 1 个在跑回合 - 被限流只登记不回，且留带 source 的可见告警（不静默） - 安静时段在回复侧只作信号（规则层不再静默），群聊取消夜间静默 - 锁定流控常量：FlowDirectDefault 加字面量钉桩测试（放大它 = 静默关流控，且会让额度循环跑百万轮） --- internal/channel/adapter/flowgate.go | 121 +++++++++-- .../channel/adapter/flowgate_direct_test.go | 104 +++++++++ internal/services/channelbridge/flowgate.go | 19 +- .../services/channelbridge/flowgate_test.go | 9 +- internal/services/channelbridge/inbound.go | 31 ++- internal/services/channelbridge/light.go | 98 ++++++--- internal/services/channelbridge/light_test.go | 108 +++++++--- .../services/channelbridge/stranger_test.go | 203 ++++++++++++++++++ internal/services/channelbridge/transcript.go | 35 ++- .../services/channelbridge/transcript_test.go | 44 ++-- 10 files changed, 654 insertions(+), 118 deletions(-) create mode 100644 internal/channel/adapter/flowgate_direct_test.go create mode 100644 internal/services/channelbridge/stranger_test.go diff --git a/internal/channel/adapter/flowgate.go b/internal/channel/adapter/flowgate.go index 915da96..d1e8abd 100644 --- a/internal/channel/adapter/flowgate.go +++ b/internal/channel/adapter/flowgate.go @@ -44,6 +44,13 @@ type FlowGateOptions struct { // GroupMsgPerMin caps group-chat inbound messages per minute **per // conversation** (design/12 §6 策略三 / 裁决 06.4.3). Zero disables the limit. GroupMsgPerMin int + // DirectMsgPerMin caps **direct (private)** inbound messages per minute + // **per conversation** — and for a private message the conversation key is + // (platform, channel, sender), so the limit throttles one sender rather than + // the whole account. It is the mandatory flow control for stranger private + // chats (主人 2026-10-05 裁决③): strangers chat normally, so a sender must be + // unable to flood her. Zero disables the limit. + DirectMsgPerMin int // MsgPerMin caps total inbound non-owner messages per minute across the whole // account — the anti-flood total guard. Zero disables it. MsgPerMin int @@ -76,6 +83,12 @@ const ( // FlowLimitDefault is the protective default of 6 group messages per minute // (design/12 §6 策略三). It is applied when an account pins no group rate limit. FlowLimitDefault = 6 + // FlowDirectDefault is the protective default of 6 direct (private) messages + // per minute **per sender** — the mandatory flow control for stranger private + // chats (主人 2026-10-05 裁决③). 6/分钟与群聊保护性默认同一口径：一个人正常 + // 一来一回打字到不了 6 条/分钟，而刷屏/机器人会在第 7 条被挡下。它不随 + // groupMsgPerMin 变化（陌生人流控不允许被「没配那项」静默关掉）。 + FlowDirectDefault = 6 // FlowCapDefault is the protective default total-cap of 50 pending messages // (design/12 §6 策略四). FlowCapDefault = 50 @@ -113,16 +126,22 @@ func ConversationKey(msg UnifiedMessage) string { // 每个会话保存一份，因此两个群并发互不挤占（裁决 06.4.3）。 type flowSession struct { groupHits []time.Time + directHits []time.Time buffer []queuedMsg lastReview time.Time + // inFlight 是该会话**正在跑**的回合数（AcquireTurn/ReleaseTurn 维护）。受限车道 + // 用它把「同一发送者同时最多 1 个在跑的回合」落到实处。 + inFlight int } // FlowGate is a concurrency-safe, per-account ingress gate implementing the two // measures from design/12 §6 策略三/四: // // - 策略三: a sliding-window rate limit on group chat (and, optionally, all -// non-owner messages). Over-limit messages are dropped (downgraded) rather -// than blocking a flood of messages from saturating the loop. +// non-owner messages), plus the mandatory **per-sender** rate limit on direct +// (private) messages that backs the stranger private-chat flow control +// (AllowDirect). Over-limit messages are dropped (downgraded) rather than +// blocking a flood of messages from saturating the loop. // - 策略四: a bounded pending-message set. When it nears its cap the Reviewer is // invoked to prune low-value messages; with no Reviewer the oldest messages are // evicted FIFO. Either way every dropped message is audited. @@ -131,15 +150,18 @@ type flowSession struct { // // Owner-private messages are not routed through the gate (the caller bypasses it), // so the owner is never throttled or pruned. Identity is never examined here. +// The gate also carries the per-conversation running-turn slot (AcquireTurn / +// ReleaseTurn) used by the stranger private-chat flow control. type FlowGate struct { - groupPerMin int - msgPerMin int - maxMessages int - reviewAt int - window time.Duration - review Reviewer - onPruned func(UnifiedMessage, string) - now func() time.Time + groupPerMin int + directPerMin int + msgPerMin int + maxMessages int + reviewAt int + window time.Duration + review Reviewer + onPruned func(UnifiedMessage, string) + now func() time.Time mu sync.Mutex totalHits []time.Time @@ -160,14 +182,15 @@ type queuedMsg struct { // except Window which defaults to one minute when left zero. func NewFlowGate(opts FlowGateOptions) *FlowGate { g := &FlowGate{ - groupPerMin: opts.GroupMsgPerMin, - msgPerMin: opts.MsgPerMin, - maxMessages: opts.MaxMessages, - window: opts.Window, - review: opts.Review, - onPruned: opts.OnPruned, - now: opts.Now, - sessions: make(map[string]*flowSession), + groupPerMin: opts.GroupMsgPerMin, + directPerMin: opts.DirectMsgPerMin, + msgPerMin: opts.MsgPerMin, + maxMessages: opts.MaxMessages, + window: opts.Window, + review: opts.Review, + onPruned: opts.OnPruned, + now: opts.Now, + sessions: make(map[string]*flowSession), } if g.window <= 0 { g.window = defaultRateWindow @@ -258,6 +281,61 @@ func (g *FlowGate) Admit(ctx context.Context, msg UnifiedMessage) GateVerdict { return GateVerdict{Proceed: true, Pruned: queueMessages(dropped)} } +// AllowDirect enforces the per-sender sliding-window rate limit for a **direct +// (private)** message (主人 2026-10-05 裁决③：陌生人私聊=正常聊天，因此必须有流控). +// For a private message the conversation key is (platform, channel, sender), so +// the window throttles one sender instead of the whole account and two strangers +// never crowd each other out. It returns false when that sender is over +// DirectMsgPerMin; the refused message is recorded to the OnPruned audit hook so +// nothing is dropped silently. A zero limit or a group message always allows. +func (g *FlowGate) AllowDirect(msg UnifiedMessage) bool { + if g.directPerMin <= 0 || msg.ChannelType != ChannelDirect { + return true + } + now := g.now() + key := ConversationKey(msg) + g.mu.Lock() + defer g.mu.Unlock() + st := g.sessionLocked(key) + st.directHits = g.trimHits(st.directHits, now) + if len(st.directHits) >= g.directPerMin { + g.auditMsg(msg, "rate_limited") + return false + } + st.directHits = append(st.directHits, now) + return true +} + +// AcquireTurn reserves the single running-turn slot of a conversation, returning +// false when a turn is already in flight for it. Callers must pair every true +// result with ReleaseTurn. It is the second half of the mandatory stranger flow +// control (主人 2026-10-05 裁决③：「同一发送者同时最多 1 个在跑的回合」): a stranger +// can send several messages inside the per-minute allowance, yet only one of them +// may occupy an expensive agent turn at a time. It never decides identity — the +// key is the same structural ConversationKey the rest of the gate uses. +func (g *FlowGate) AcquireTurn(msg UnifiedMessage) bool { + key := ConversationKey(msg) + g.mu.Lock() + defer g.mu.Unlock() + st := g.sessionLocked(key) + if st.inFlight >= 1 { + return false + } + st.inFlight++ + return true +} + +// ReleaseTurn releases a slot reserved by AcquireTurn. +func (g *FlowGate) ReleaseTurn(msg UnifiedMessage) { + key := ConversationKey(msg) + g.mu.Lock() + defer g.mu.Unlock() + st := g.sessionLocked(key) + if st.inFlight > 0 { + st.inFlight-- + } +} + // drop builds a refused verdict from a single message that never entered the // buffer (rate-limited at the door), so the audit hook still sees it. func (g *FlowGate) drop(reason string, msg UnifiedMessage) GateVerdict { @@ -281,7 +359,8 @@ func (g *FlowGate) sessionLocked(key string) *flowSession { } // maybeGCLocked 每 sessionGCInterval 次放行清理一次空闲会话，避免会话状态表随历史会话 -// 无界增长：滑窗时间戳全部过期且待处理队列为空的会话直接删除。调用方须持有 g.mu。 +// 无界增长：滑窗时间戳全部过期、待处理队列为空、且没有回合在跑的会话直接删除。 +// 调用方须持有 g.mu。 func (g *FlowGate) maybeGCLocked(now time.Time) { g.admitted++ if g.admitted%sessionGCInterval != 0 { @@ -289,7 +368,9 @@ func (g *FlowGate) maybeGCLocked(now time.Time) { } for key, st := range g.sessions { st.groupHits = g.trimHits(st.groupHits, now) - if len(st.groupHits) == 0 && len(st.buffer) == 0 { + st.directHits = g.trimHits(st.directHits, now) + if len(st.groupHits) == 0 && len(st.directHits) == 0 && + len(st.buffer) == 0 && st.inFlight == 0 { delete(g.sessions, key) } } diff --git a/internal/channel/adapter/flowgate_direct_test.go b/internal/channel/adapter/flowgate_direct_test.go new file mode 100644 index 0000000..a1deb39 --- /dev/null +++ b/internal/channel/adapter/flowgate_direct_test.go @@ -0,0 +1,104 @@ +package adapter + +// 本文件覆盖主人 2026-10-05 裁决③的**陌生人私聊流控**两件套： +// - AllowDirect：私聊按发送者（会话键 platform+channel+sender）的每分钟滑窗； +// - AcquireTurn/ReleaseTurn：同一发送者同时最多 1 个在跑的回合。 +// 两者都复用同一张 flowSession 表与同一套 ConversationKey 口径，不另造第二套限流。 + +import ( + "testing" + "time" +) + +// TestFlowGate_AllowDirectIsPerSender 钉住私聊滑窗按发送者独立计数： +// A 用尽额度不挤占 B；窗口滚过后 A 恢复；群聊消息不经这条通道。 +func TestFlowGate_AllowDirectIsPerSender(t *testing.T) { + t.Parallel() + + now, clock := fixedClock(time.Unix(2000, 0)) + var audited []string + g := NewFlowGate(FlowGateOptions{ + DirectMsgPerMin: 2, + Window: time.Minute, + Now: now, + OnPruned: func(_ UnifiedMessage, reason string) { audited = append(audited, reason) }, + }) + + for i := 0; i < 2; i++ { + if !g.AllowDirect(directMsgTo("peer-1", i)) { + t.Fatalf("peer-1 第 %d 条被拒，want allowed", i+1) + } + } + if g.AllowDirect(directMsgTo("peer-1", 2)) { + t.Fatal("peer-1 超出自身额度仍被放行") + } + if len(audited) != 1 || audited[0] != "rate_limited" { + t.Fatalf("拒绝未入审计：%v", audited) + } + if !g.AllowDirect(directMsgTo("peer-2", 0)) { + t.Fatal("peer-2 被 peer-1 的额度挤占（私聊限流未按发送者独立）") + } + // 群聊走 Admit 的群聊滑窗，AllowDirect 对它恒放行。 + if !g.AllowDirect(groupMsgIn("-100", 0)) { + t.Fatal("群聊消息不该被私聊限流拦下") + } + + *clock = clock.Add(61 * time.Second) + if !g.AllowDirect(directMsgTo("peer-1", 3)) { + t.Fatal("窗口滚过后 peer-1 应恢复放行") + } +} + +// TestFlowGate_AllowDirectDisabledByZero 钉住零值 = 关闭：没有配置私聊限流时 +// （DirectMsgPerMin 为 0）AllowDirect 恒放行，既有调用行为不变。 +func TestFlowGate_AllowDirectDisabledByZero(t *testing.T) { + t.Parallel() + + g := NewFlowGate(FlowGateOptions{}) + for i := 0; i < 50; i++ { + if !g.AllowDirect(directMsgTo("peer-1", i)) { + t.Fatalf("零值额度下第 %d 条被拒，want allowed", i+1) + } + } +} + +// TestFlowDirectDefaultIsPinned 用**字面量 6** 钉住保护性默认值（与 FlowLimitDefault +// 的既有钉桩同口径）。不这么钉的后果：`stranger_test.go` 的额度循环以该常量为上界， +// 一旦有人把它改成近似无限，测试不会断言失败，而是**跑一百万轮把整包测试拖到超时** +// ——看起来像「偶发卡死」，实际是流控被静默关掉。主人 2026-10-05 裁决③要求陌生人 +// 私聊**必须有**流控，因此这条默认值属于契约，不允许随手放大。 +func TestFlowDirectDefaultIsPinned(t *testing.T) { + t.Parallel() + + if FlowDirectDefault != 6 { + t.Fatalf("FlowDirectDefault = %d, want 6（每发送者每分钟 6 条的保护性默认；"+ + "放大它等于静默关闭陌生人私聊流控）", FlowDirectDefault) + } +} + +// TestFlowGate_AcquireTurnSinglePerConversation 钉住「同一会话同时最多 1 个在跑的回合」： +// 第一个占位成功、第二个被拒；换一个发送者互不影响；释放后又能占。 +func TestFlowGate_AcquireTurnSinglePerConversation(t *testing.T) { + t.Parallel() + + g := NewFlowGate(FlowGateOptions{}) + a := directMsgTo("peer-1", 0) + b := directMsgTo("peer-2", 0) + + if !g.AcquireTurn(a) { + t.Fatal("首个回合应能占到槽位") + } + if g.AcquireTurn(a) { + t.Fatal("同一发送者的第二个回合不该同时占到槽位") + } + if !g.AcquireTurn(b) { + t.Fatal("另一个发送者的槽位被挤占") + } + + g.ReleaseTurn(a) + if !g.AcquireTurn(a) { + t.Fatal("释放后应能重新占到槽位") + } + g.ReleaseTurn(a) + g.ReleaseTurn(b) +} diff --git a/internal/services/channelbridge/flowgate.go b/internal/services/channelbridge/flowgate.go index c74f97a..c9fa285 100644 --- a/internal/services/channelbridge/flowgate.go +++ b/internal/services/channelbridge/flowgate.go @@ -19,6 +19,11 @@ import ( // （adapter.ConversationKey：platform+channelID+groupID/peerID），因此两个群绝不会 // 相互挤占（design/12 §6 策略三/四，裁决 06.4.3）。 func (b *Bridge) gateFor(a adapter.PlatformAdapter) *adapter.FlowGate { + if a == nil { + // 便捷路径（process 用平台名找适配器）在平台上未注册任何账号时给出 nil： + // 没有账号就没有闸门，调用方按 nil 跳过（生产路径的 a 恒非 nil）。 + return nil + } name := a.Name() b.gateMu.Lock() defer b.gateMu.Unlock() @@ -31,15 +36,21 @@ func (b *Bridge) gateFor(a adapter.PlatformAdapter) *adapter.FlowGate { } // gateOptions 把适配器 Config 快照中的按账号流控限制解析进 -// FlowGate。键缺失（单账号环境变量适配器）时限制关闭， +// FlowGate。群聊/总量/上限键缺失（单账号环境变量适配器）时限制关闭， // 既有渠道行为不变；键存在且值为 0 时采用保护性默认值 // （每群每分钟 6 条、上限 50），负值则关闭限制—— // 这是对 CHANNELS_JSON 账号的向后兼容约定（design/12 §6 实现要点）。 +// +// **私聊（陌生人）限流例外：它不依赖任何配置键**（主人 2026-10-05 裁决③）， +// 恒定取保护性默认 FlowDirectDefault —— 原本没有任何配置的账号（如 qq-obv11） +// 对陌生人私聊是完全敞口，而裁决要求「陌生人必须有流控」，所以这里不能 +// 因为「没配」而静默关掉。主人私聊在 onMessage 就被绕过本闸门，不受影响。 func (b *Bridge) gateOptions(a adapter.PlatformAdapter) adapter.FlowGateOptions { opts := adapter.FlowGateOptions{ - Review: b.review, - OnPruned: b.onRejected, - Now: b.now, + Review: b.review, + OnPruned: b.onRejected, + Now: b.now, + DirectMsgPerMin: adapter.FlowDirectDefault, } cfg := a.Config() if grp, ok := cfgInt(cfg, "groupMsgPerMin"); ok { diff --git a/internal/services/channelbridge/flowgate_test.go b/internal/services/channelbridge/flowgate_test.go index b6641c6..93145d0 100644 --- a/internal/services/channelbridge/flowgate_test.go +++ b/internal/services/channelbridge/flowgate_test.go @@ -104,7 +104,8 @@ func TestFlowGate_OwnerPrivateBypassesGate(t *testing.T) { // TestFlowGate_GateOptionsDefaults 钉住「配置→限制」的解析规则， // 它保证旧的 CHANNELS_JSON 账号向后兼容：键缺失 → 关闭，0 → 保护性 -// 默认值，负值 → 关闭，正值 → 固定为该值。 +// 默认值，负值 → 关闭，正值 → 固定为该值。**私聊（陌生人）限流是例外**： +// 它不读配置键，恒取保护性默认（裁决③：陌生人必须有流控）。 func TestFlowGate_GateOptionsDefaults(t *testing.T) { t.Parallel() @@ -135,6 +136,12 @@ func TestFlowGate_GateOptionsDefaults(t *testing.T) { if opts.MaxMessages != c.max { t.Fatalf("MaxMessages = %d, want %d", opts.MaxMessages, c.max) } + // 陌生人私聊的强制流控不随账号配置开关：无论群聊/上限怎么配， + // 私聊限流恒为保护性默认（主人 2026-10-05 裁决③）。 + if opts.DirectMsgPerMin != adapter.FlowDirectDefault { + t.Fatalf("DirectMsgPerMin = %d, want the protective default %d", + opts.DirectMsgPerMin, adapter.FlowDirectDefault) + } }) } } diff --git a/internal/services/channelbridge/inbound.go b/internal/services/channelbridge/inbound.go index 5cc1dd3..e70d588 100644 --- a/internal/services/channelbridge/inbound.go +++ b/internal/services/channelbridge/inbound.go @@ -181,17 +181,18 @@ func (b *Bridge) processFrom(ctx context.Context, msg adapter.UnifiedMessage, a "session", sess.SessionID, "actor_role", sess.ActorRole, "env", sess.Env(), "mem_writable", sess.MemWritable) - // design/12 §6 策略一/二/五：面向非主人消息的最小打扰闸门。 - // 被廉价轻量判定认为不值得回复的非主人群聊/陌生人 - // 消息，只登记到其隔离会话日志并静默——绝不 - // 驱动完整 agent 循环。主人消息绝不在此被拦，因此 - // 主人私聊与主人在群里的 @/高价值回复不受影响。 + // design/12 §6 策略一/二：面向非主人消息的最小打扰闸门。 + // 被廉价轻量判定认为不值得回复的非主人**群聊**消息，只登记到 + // 其隔离会话日志并静默——绝不驱动完整 agent 循环；**陌生人私聊**则 + // 一律走完整回合（裁决①，见 shouldRespond），只受流控约束。 + // 主人消息绝不在此被拦，因此主人私聊与主人在群里的 + // @/高价值回复不受影响。 // // 由「传入前审查」放行的系统通知（notice.Applicable）不在此列：它们已由策略一 // 的归一化/分类**终局判定**（高价值升级真会话、其余限定注入），若再走这道面向 // 聊天消息的闸门，actor=system 会在 isLowValueEvent 处被第二次静默掉。 if !actor.IsOwner() && !notice.Applicable { - if ok, reason := b.shouldRespond(ctx, msg, actor, sess); !ok { + if ok, reason := b.shouldRespond(ctx, msg, actor, sess, a); !ok { b.registerSilenced(ctx, msg, actor, sess, a) b.log.Debug("non-owner message silenced (minimal disturbance)", "platform", msg.Platform, "channel", msg.ChannelID, @@ -200,6 +201,21 @@ func (b *Bridge) processFrom(ctx context.Context, msg adapter.UnifiedMessage, a } } + // 裁决①的第二个流控（主人 2026-10-05）：陌生人私聊虽然一律回答，但 + // **同一发送者同时最多 1 个在跑的回合**。槽位复用同一个 flowgate 的 + // 会话计数（AcquireTurn/ReleaseTurn，键口径与限流一致）；已有回合在跑时 + // 这一条只登记、不回，并留可见告警 —— 一个陌生人无法用并发消息占满循环。 + if isStrangerPrivate(sess) { + if gate := b.gateFor(a); gate != nil { + if !gate.AcquireTurn(msg) { + b.registerThrottled(ctx, msg, actor, sess, a, + "stranger private: a turn is already running for this sender") + return + } + defer gate.ReleaseTurn(msg) + } + } + b.runInboundTurn(ctx, msg, a, text, actor, sess) } @@ -397,7 +413,8 @@ func deviceIDFor(msg adapter.UnifiedMessage) string { // RoleUser 消息**上 —— 用户内容只进用户角色消息是项目红线。 // // 历史转录**不在本函数里**：它由 seedTurnTranscript（transcript.go）在本函数返回之后 -// 前置到列表最前面，与网关编排器的播种同序（转录在前、本次消息在后）。因此本函数只负责 +// 插入到列表中（主人私聊前置在最前；受限车道插在**最前的系统框架行之后**， +// 于是来源行 + 第三方边界仍在最前、转录居中、当次消息在最后）。因此本函数只负责 // **本回合自己的**消息（来源行 + 当次用户消息），受限车道的形状在有无转录时保持一致。 func (b *Bridge) buildLoopMessages(_ context.Context, text string, env loop.SessionEnv, actor adapter.Actor, senderName string, src adapter.UnifiedMessage) []model.Message { // design/12 §6 策略一补充：经传入前审查放行的系统通知走「限定注入」——只把剥离后的 diff --git a/internal/services/channelbridge/light.go b/internal/services/channelbridge/light.go index a3c698e..3b329fc 100644 --- a/internal/services/channelbridge/light.go +++ b/internal/services/channelbridge/light.go @@ -11,15 +11,17 @@ import ( ) // 本文件承载渠道桥接器面向公开平台的入站「最小打扰」分级 -// （DevDocs/design/12 §6 策略一/二/五）。非主人的群聊/陌生人 -// 消息在进入完整 agent 循环前先经廉价判断，使繁忙的 -// 公开群不会把每条消息都扔进昂贵的 LLM 轮次。这些逻辑 -// 放在这里是为了让 channelbridge.go 保持在模块体积上限内；它绝不 -// 读取消息内容来判定身份——身份早已由 +// （DevDocs/design/12 §6 策略一/二）。非主人的**群聊**消息在进入完整 +// agent 循环前先经廉价判断，使繁忙的公开群不会把每条消息都扔进 +// 昂贵的 LLM 轮次；**陌生人私聊**则一律正常聊天（主人 2026-10-05 裁决①）， +// 只受强制流控约束（裁决③）。这些逻辑放在这里是为了让 channelbridge.go +// 保持在模块体积上限内；它绝不读取消息内容来判定身份——身份早已由 // 会话/actor 层结构化解析完毕（项目红线）。 // 安静时段默认值，与 backthink 服务保持一致，使桥接器与 -// 后台思考器共用同一个免打扰窗口（design/12 §6 策略五）。 +// 后台思考器共用同一个免打扰窗口（design/12 §6 策略五）。它在本车道里 +// **只作为信号**随 inQuiet 传给参与度引擎/轻量判定器，不再让规则层静默 +// （主人 2026-10-05 裁决②）。 const ( defaultQuietStartHour = 23 defaultQuietEndHour = 7 @@ -32,11 +34,10 @@ type lightVerdict int const ( // verdictRespond 表示规则信号显示该消息值得回复。 verdictRespond lightVerdict = iota - // verdictSilent 表示规则信号显示该消息应登记但不应 - // 回复（例如安静时段内没有强信号）。 - verdictSilent - // verdictInconclusive 表示规则无法判断；交给 LightJudge（未装配 - // 时则失败关闭为静默）。 + // verdictInconclusive 表示规则无法判断；交给参与度引擎（design/18） + // 或注入式 LightJudge（未装配时则失败关闭为静默）。安静时段**不再** + // 单独构成静默（主人 2026-10-05 裁决②）——它只作为信号随 inQuiet + // 传给下游判定器，让轻思考/参与度自己权衡要不要回。 verdictInconclusive ) @@ -130,12 +131,14 @@ func hasHighValueKeyword(text string) bool { } // ruleLightJudge 只依据廉价规则信号判断是否值得回复 -// （设计/12 §6 策略二/五）。优先级：直接点名、提及主人的显示 -// 名、高价值关键词，其次是安静时段偏向静默；否则 -// 结论为无结论（交给注入的 LightJudge，没有则静默）。它 -// 只读取已解析的 actor 元数据与消息内容——绝不推断 -// 身份。 -func (b *Bridge) ruleLightJudge(msg adapter.UnifiedMessage, actor adapter.Actor, inQuiet bool) lightVerdict { +// （设计/12 §6 策略二）。优先级：直接点名、提及主人的显示 +// 名、高价值关键词；除此之外一律无结论（交给参与度引擎 / +// 注入的 LightJudge，没有则失败关闭为静默）。**安静时段不是 +// 这里的静默理由**（主人 2026-10-05 裁决②：安静时段由轻思考判断 +// 是否回复，不是硬性全静默）——该信号由调用方随 inQuiet 传给 +// 下游判定器。它只读取已解析的消息信号（提及/正文）——绝不 +// 推断身份（身份由 actor 元数据在会话层解析，红线）。 +func (b *Bridge) ruleLightJudge(msg adapter.UnifiedMessage) lightVerdict { if len(msg.Mentions) > 0 { return verdictRespond } @@ -146,20 +149,26 @@ func (b *Bridge) ruleLightJudge(msg adapter.UnifiedMessage, actor adapter.Actor, if hasHighValueKeyword(msg.Content) { return verdictRespond } - if inQuiet { - return verdictSilent - } return verdictInconclusive } +// isStrangerPrivate 报告该会话是否为**陌生人私聊**（非群 + 非主人）。 +// 它只读会话层结构化解析出的标志（项目红线：身份不靠正文/提示词）。 +// 主人私聊在 onMessage/shouldRespond 之前就已被 actor 门拦下，因此 +// 这里成真的只会是陌生人私聊。 +func isStrangerPrivate(sess session.Session) bool { + return !sess.IsGroup && !sess.Owner +} + // shouldRespond 实现非主人入站消息的最小打扰分级 -// （设计/12 §6 策略一/二/五 + design/18 §2/§4/§5）。它返回该消息 +// （设计/12 §6 策略一/二 + design/18 §2/§4/§5）。它返回该消息 // 应驱动完整 agent 循环并回复，还是只登记并静默。 // 主人消息绝不会到这里（processFrom 以 !actor.IsOwner 为门）， // 因此主人私聊与主人在群里的 @/高价值回复不受影响。sess 是 // 该消息解析出的隔离群聊/陌生人会话绑定，使参与度 -// 引擎探测时能读到群的近期上下文。 -func (b *Bridge) shouldRespond(ctx context.Context, msg adapter.UnifiedMessage, actor adapter.Actor, sess session.Session) (bool, string) { +// 引擎探测时能读到群的近期上下文；a 是收到该消息的账号适配器， +// 用于取该账号的流控闸门（陌生人私聊的强制限流）。 +func (b *Bridge) shouldRespond(ctx context.Context, msg adapter.UnifiedMessage, actor adapter.Actor, sess session.Session, a adapter.PlatformAdapter) (bool, string) { // 策略一：平台/系统噪声 → 只登记，绝不进完整循环。 if isLowValueEvent(actor) { b.log.Debug("non-owner low-value system event, registered only", @@ -167,18 +176,29 @@ func (b *Bridge) shouldRespond(ctx context.Context, msg adapter.UnifiedMessage, return false, "low-value system event; registered only" } - // 策略二/五：基于规则的轻量判定，感知安静时段。直接 @、 - // 提及主人显示名或高价值关键词一律回复；安静时段 - // 偏向静默。只有无结论的剩余部分才咨询 - // 参与度引擎。 + // 裁决①（主人 2026-10-05）：**陌生人私聊 = 正常聊天** —— 不设安静期、 + // 不走「值不值得回」的概率门，直接进入完整回合。唯一的闸门是强制 + // 流控：同一发送者每分钟 ≤ FlowDirectDefault 条（AllowDirect 记账） + // 与同时最多 1 个在跑的回合（processFrom 用 AcquireTurn 施加）。 + // 被限流时**不回**：调用方会把它登记进隔离会话，这里留一条带 + // source 的可见告警（绝不静默）。 + if isStrangerPrivate(sess) { + if g := b.gateFor(a); g != nil && !g.AllowDirect(msg) { + b.log.Warn("stranger private chat rate limited, registered only", + "platform", msg.Platform, "channel", msg.ChannelID, "sender", msg.SenderID, + "limit_per_min", adapter.FlowDirectDefault) + return false, "stranger private: over the per-sender flow limit" + } + return true, "stranger private chat: normal reply" + } + + // 策略二/五：基于规则的轻量判定。直接 @、提及主人显示名或 + // 高价值关键词一律直接回；规则得不出结论的部分交给参与度 + // 引擎 / 轻量判定器。**安静时段只作为信号**（inQuiet）继续往下传， + // 规则层不再因为它静默（裁决②）。 inQuiet := b.inQuietHours(b.now()) - switch b.ruleLightJudge(msg, actor, inQuiet) { - case verdictRespond: + if b.ruleLightJudge(msg) == verdictRespond { return true, "rule signal: worth responding" - case verdictSilent: - b.log.Debug("non-owner message silenced (quiet/no signal), registered only", - "platform", msg.Platform, "channel", msg.ChannelID, "quiet", inQuiet, "actor_role", actor.Role) - return false, "rule signal: not worth responding" } // 廉价规则得不出结论。参与度引擎（design/18）已配置时 @@ -245,6 +265,18 @@ func (b *Bridge) registerSilenced(ctx context.Context, msg adapter.UnifiedMessag "platform", msg.Platform, "channel", msg.ChannelID, "session", sess.SessionID) } +// registerThrottled 是一次被强制流控（同一发送者已有回合在跑）拦下的 +// 非主人消息的落点：先走最小打扰的登记路径（registerSilenced）把它写进 +// 其隔离会话，再留一条**带 source 的可见告警** —— 被限流绝不静默 +// （主人 2026-10-05 裁决③）。限流路径与普通静默因此区分得开：前者 +// 一定有可见日志，后者只有 debug。 +func (b *Bridge) registerThrottled(ctx context.Context, msg adapter.UnifiedMessage, actor adapter.Actor, sess session.Session, a adapter.PlatformAdapter, reason string) { + b.registerSilenced(ctx, msg, actor, sess, a) + b.log.Warn("inbound message throttled by flow gate, registered only", + "platform", msg.Platform, "channel", msg.ChannelID, + "sender", msg.SenderID, "reason", reason) +} + // inQuietHours 报告 at 是否落在配置的免打扰窗口内 // （与 backthink 语义一致：两个非零且相等的时刻关闭该窗口， // 跨午夜的窗口也能正确处理）。它是 Bridge 上的纯方法，测试 diff --git a/internal/services/channelbridge/light_test.go b/internal/services/channelbridge/light_test.go index 0fc46e5..657f722 100644 --- a/internal/services/channelbridge/light_test.go +++ b/internal/services/channelbridge/light_test.go @@ -14,15 +14,18 @@ import ( "git.yeij.top/AskaEth/Cyrene_v2/internal/user" ) -// stubLightJudge 是可配置的 LightJudge，记录自己被咨询了多少次。 +// stubLightJudge 是可配置的 LightJudge，记录自己被咨询了多少次， +// 并保留最后一次收到的输入（用于断言 inQuiet 真值确实传到了下游判定器）。 type stubLightJudge struct { resp LightDecision err error calls int + last LightInput } -func (s *stubLightJudge) Judge(context.Context, LightInput) (LightDecision, error) { +func (s *stubLightJudge) Judge(_ context.Context, in LightInput) (LightDecision, error) { s.calls++ + s.last = in return s.resp, s.err } @@ -127,40 +130,19 @@ func TestNonOwnerHighValueKeywordResponds(t *testing.T) { } } -func TestNonOwnerQuietHoursSilenced(t *testing.T) { - t.Parallel() - - chat := &stubChatter{res: &loop.Result{Text: "不该回"}} - sess := &stubSessions{sess: nonOwnerSession()} - hist := &captureHistory{} - b, err := New(Options{ - Loop: chat, Sessions: sess, Owner: &user.OwnerProfile{}, History: hist, - QuietStartHour: 23, QuietEndHour: 7, - Now: func() time.Time { return time.Date(2026, 1, 1, 2, 0, 0, 0, time.UTC) }, - }) - if err != nil { - t.Fatalf("New: %v", err) - } - - // 策略五：安静时段内无信号的非主人消息被静默—— - // 只登记、不回复。 - b.process(context.Background(), nonOwnerGroupMsg("hello all")) - - if chat.sessionID != "" || len(chat.messages) != 0 { - t.Fatalf("loop ran during quiet hours for a no-signal message: session=%q messages=%+v", chat.sessionID, chat.messages) - } - if len(hist.records()) != 1 { - t.Fatalf("records = %d, want 1 (registered, no reply)", len(hist.records())) - } -} - -func TestNonOwnerQuietHoursMentionResponds(t *testing.T) { +// TestNonOwnerQuietHoursRulesDoNotSilenceGroup 钉住裁决②（主人 2026-10-05）： +// **安静时段不再是规则层的硬静默**——群里没被 @、无关键词的普通消息在 02:00 +// 仍会走到轻思考（LightJudge），由它决定回不回，而不是连判定器都不咨询就被 +// 静默。变异验证：把 `if inQuiet { return verdictSilent }` 加回 ruleLightJudge， +// 本用例立刻变红（judge.calls 会是 0、循环不会跑）。 +func TestNonOwnerQuietHoursRulesDoNotSilenceGroup(t *testing.T) { t.Parallel() + judge := &stubLightJudge{resp: LightDecision{Respond: true, Reason: "夜里这场值得聊"}} chat := &stubChatter{res: &loop.Result{Text: "在的"}} sess := &stubSessions{sess: nonOwnerSession()} b, err := New(Options{ - Loop: chat, Sessions: sess, Owner: &user.OwnerProfile{}, + Loop: chat, Sessions: sess, Owner: &user.OwnerProfile{}, Light: judge, QuietStartHour: 23, QuietEndHour: 7, Now: func() time.Time { return time.Date(2026, 1, 1, 2, 0, 0, 0, time.UTC) }, }) @@ -168,12 +150,72 @@ func TestNonOwnerQuietHoursMentionResponds(t *testing.T) { t.Fatalf("New: %v", err) } - // 策略五不是硬性静默：被直接点名的消息在安静时段 - // 仍会得到回复。 + // 无 @、无关键词、无主人显示名的群消息，落在安静时段内。 + b.process(context.Background(), nonOwnerGroupMsg("hello all")) + + if judge.calls != 1 { + t.Fatalf("light judge calls = %d, want 1（安静时段不得在规则层硬静默，应交给轻思考）", judge.calls) + } + // 安静时段作为**信号**必须是真的（而非硬编码 false）传到下游判定器。 + if !judge.last.InQuietHours { + t.Fatalf("light judge InQuietHours = false, want true at 02:00（安静时段信号失真）") + } + if chat.sessionID != "sess-g" { + t.Fatalf("loop session = %q, want sess-g（轻思考说回就该回）", chat.sessionID) + } +} + +// TestNonOwnerQuietHoursMentionResponds 保留「安静时段不是硬性全静默」的正面口径： +// 被直接点名的群消息在 02:00 仍直接回，且**根本不咨询轻思考**（规则已能给结论）。 +func TestNonOwnerQuietHoursMentionResponds(t *testing.T) { + t.Parallel() + + judge := &stubLightJudge{resp: LightDecision{Respond: false, Reason: "不该被问到"}} + chat := &stubChatter{res: &loop.Result{Text: "在的"}} + sess := &stubSessions{sess: nonOwnerSession()} + b, err := New(Options{ + Loop: chat, Sessions: sess, Owner: &user.OwnerProfile{}, Light: judge, + QuietStartHour: 23, QuietEndHour: 7, + Now: func() time.Time { return time.Date(2026, 1, 1, 2, 0, 0, 0, time.UTC) }, + }) + if err != nil { + t.Fatalf("New: %v", err) + } + + // 策略二：被直接点名的消息在安静时段仍会得到回复。 b.process(context.Background(), nonOwnerGroupMsg("在吗", adapter.Mention{UserID: "bot"})) if chat.sessionID != "sess-g" { t.Fatalf("loop session = %q, want sess-g (mention should break quiet-hours silence)", chat.sessionID) } + if judge.calls != 0 { + t.Fatalf("light judge calls = %d, want 0（@ 是规则可直接判定的强信号）", judge.calls) + } +} + +// TestNonOwnerQuietHoursKeywordResponds 补齐另一条强信号：命中高价值关键词的群消息 +// 在 02:00 也直接回，同样不咨询轻思考。 +func TestNonOwnerQuietHoursKeywordResponds(t *testing.T) { + t.Parallel() + + judge := &stubLightJudge{resp: LightDecision{Respond: false, Reason: "不该被问到"}} + chat := &stubChatter{res: &loop.Result{Text: "我看看"}} + sess := &stubSessions{sess: nonOwnerSession()} + b, err := New(Options{ + Loop: chat, Sessions: sess, Owner: &user.OwnerProfile{}, Light: judge, + QuietStartHour: 23, QuietEndHour: 7, + Now: func() time.Time { return time.Date(2026, 1, 1, 2, 0, 0, 0, time.UTC) }, + }) + if err != nil { + t.Fatalf("New: %v", err) + } + + b.process(context.Background(), nonOwnerGroupMsg("有人在吗，紧急求助")) + if chat.sessionID != "sess-g" { + t.Fatalf("loop session = %q, want sess-g (keyword should break quiet-hours silence)", chat.sessionID) + } + if judge.calls != 0 { + t.Fatalf("light judge calls = %d, want 0（高价值关键词是规则可直接判定的强信号）", judge.calls) + } } func TestOwnerUnaffectedInQuietHours(t *testing.T) { diff --git a/internal/services/channelbridge/stranger_test.go b/internal/services/channelbridge/stranger_test.go new file mode 100644 index 0000000..a98c88d --- /dev/null +++ b/internal/services/channelbridge/stranger_test.go @@ -0,0 +1,203 @@ +package channelbridge + +import ( + "bytes" + "context" + "log/slog" + "strings" + "testing" + "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/runtime/loop" + "git.yeij.top/AskaEth/Cyrene_v2/internal/user" +) + +// 本文件覆盖主人 2026-10-05 裁决①②③的三条口径： +// - 陌生人私聊 = 正常聊天：非群 + 非主人一律直接进入完整回合，任何时刻（含安静时段 +// 23:30）都不受安静期或概率门影响； +// - 安静时段的硬静默已从规则层摘除（断言在 light_test.go）； +// - 陌生人**必须有流控**：同一发送者每分钟 ≤ FlowDirectDefault 条、同时最多 1 个在跑的 +// 回合；触发时只登记 + 一条带 source 的可见告警，绝不静默。 + +// strangerPrivateSession 是陌生人私聊的隔离会话绑定：非群、非主人、记忆只读。 +func strangerPrivateSession() session.Session { + return session.Session{SessionID: "sess-s", IsGroup: false, Owner: false, MemWritable: false} +} + +// strangerDirectMsg 构造一条来自陌生人（非主人，由 owner 白名单结构化判定）的私聊消息。 +// 私聊的会话键是 (platform, channel, sender)，因此换一个 sender 就换一个限流桶。 +func strangerDirectMsg(sender, text string) adapter.UnifiedMessage { + return adapter.UnifiedMessage{ + Platform: "telegram", + ChannelID: "100", + ChannelType: adapter.ChannelDirect, + SenderID: sender, + SenderName: "陌生人" + sender, + Sender: adapter.Actor{Role: adapter.RoleMember}, + Content: text, + ContentType: adapter.ContentText, + } +} + +// newStrangerBridge 构建一条接好 telegram 适配器的陌生人私聊测试桥接器。 +func newStrangerBridge(t *testing.T, chat *stubChatter, hist *captureHistory, at time.Time, log *slog.Logger) (*Bridge, *stubSessions, *fakeAdapter) { + t.Helper() + sess := &stubSessions{sess: strangerPrivateSession()} + ad := &fakeAdapter{name: "telegram"} + b, err := New(Options{ + Loop: chat, Sessions: sess, Owner: &user.OwnerProfile{}, History: hist, Log: log, + QuietStartHour: 23, QuietEndHour: 7, + Now: func() time.Time { return at }, + }) + if err != nil { + t.Fatalf("New: %v", err) + } + b.Add(ad) + return b, sess, ad +} + +// TestStrangerPrivateChatRespondsIncludingQuietHours 钉住裁决①：陌生人私聊（非群 + 非主人） +// 在任何时刻都是正常聊天 —— 白天如此，安静时段 23:30 也如此（安静时段不设在她与陌生人之间）。 +// 变异验证：把「陌生人跳过概率门」那段摘掉，本用例立刻变红（shouldRespond 会落到规则/参与度 +// 链上，无信号 ⇒ 静默、循环不跑）。 +func TestStrangerPrivateChatRespondsIncludingQuietHours(t *testing.T) { + t.Parallel() + + cases := []struct { + name string + at time.Time + }{ + {"daytime", time.Date(2026, 1, 1, 12, 0, 0, 0, time.UTC)}, + {"quiet hours 23:30", time.Date(2026, 1, 1, 23, 30, 0, 0, time.UTC)}, + } + for _, tc := range cases { + tc := tc + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + + chat := &stubChatter{res: &loop.Result{Text: "你好呀"}} + b, sess, ad := newStrangerBridge(t, chat, &captureHistory{}, tc.at, nil) + + msg := strangerDirectMsg("777", "在吗，普通的一句话") + actor := session.ResolveActor(msg, b.owner) + if ok, reason := b.shouldRespond(context.Background(), msg, actor, sess.sess, ad); !ok { + t.Fatalf("shouldRespond = false (%s), want true（陌生人私聊一律正常聊天）", reason) + } + + b.process(context.Background(), msg) + if chat.sessionID != "sess-s" { + t.Fatalf("loop session = %q, want the isolated sess-s", chat.sessionID) + } + if len(ad.sends) != 1 || ad.sends[0].Content != "你好呀" { + t.Fatalf("stranger replies = %+v, want one text reply", ad.sends) + } + }) + } +} + +// TestStrangerPrivateFlowControlPerSender 钉住裁决③：陌生人私聊有强制流控。 +// 同一发送者在额度内连发都会被回答；超出 FlowDirectDefault 的第 N+1 条**不回**、 +// 只登记，并留一条**带 source 的可见告警**；另一条发送者的额度完全独立、不受挤占。 +func TestStrangerPrivateFlowControlPerSender(t *testing.T) { + t.Parallel() + + chat := &stubChatter{res: &loop.Result{Text: "嗯"}} + hist := &captureHistory{} + var logBuf bytes.Buffer + log := slog.New(slog.NewTextHandler(&logBuf, &slog.HandlerOptions{Level: slog.LevelDebug})) + at := time.Date(2026, 1, 1, 23, 30, 0, 0, time.UTC) + b, sess, ad := newStrangerBridge(t, chat, hist, at, log) + + for i := 0; i < adapter.FlowDirectDefault; i++ { + chat.sessionID = "" + b.process(context.Background(), strangerDirectMsg("777", "第 "+itoa(i)+" 句")) + if chat.sessionID != "sess-s" { + t.Fatalf("额度内第 %d 条被拒（session=%q）", i+1, chat.sessionID) + } + } + + // 第 N+1 条超限：不回，只登记，且有可见日志。 + before := len(hist.records()) + chat.sessionID = "" + b.process(context.Background(), strangerDirectMsg("777", "刷屏第 N+1 句")) + if chat.sessionID != "" { + t.Fatalf("超限消息驱动了循环：session=%q", chat.sessionID) + } + if got := len(hist.records()) - before; got != 1 { + t.Fatalf("超限消息新增登记 %d 条，want 1（只登记不回）", got) + } + logged := logBuf.String() + if !strings.Contains(logged, "stranger private chat rate limited") || + !strings.Contains(logged, "source=services/channelbridge") { + t.Fatalf("被限流没有留下带 source 的可见日志：%q", logged) + } + + // 另一条发送者：额度独立，第一条照常回答。 + other := strangerDirectMsg("888", "我是另一个人") + if ok, reason := b.shouldRespond(context.Background(), other, session.ResolveActor(other, b.owner), sess.sess, ad); !ok { + t.Fatalf("另一个发送者被同一额度挤占：reason=%s", reason) + } +} + +// TestStrangerPrivateSecondTurnRegisteredWhileOneRuns 钉住裁决③的后半句： +// 「同一发送者同时最多 1 个在跑的回合」。先用 AcquireTurn 占住该发送者的槽位（模拟 +// 一个回合正在跑），再让下一条消息过门 —— 它必须只登记、不回，并留可见告警。 +func TestStrangerPrivateSecondTurnRegisteredWhileOneRuns(t *testing.T) { + t.Parallel() + + chat := &stubChatter{res: &loop.Result{Text: "嗯"}} + hist := &captureHistory{} + var logBuf bytes.Buffer + log := slog.New(slog.NewTextHandler(&logBuf, &slog.HandlerOptions{Level: slog.LevelDebug})) + b, _, ad := newStrangerBridge(t, chat, hist, time.Date(2026, 1, 1, 12, 0, 0, 0, time.UTC), log) + + msg := strangerDirectMsg("777", "在吗") + gate := b.gateFor(ad) + if gate == nil || !gate.AcquireTurn(msg) { + t.Fatal("占位失败：测试前置条件不成立") + } + t.Cleanup(func() { gate.ReleaseTurn(msg) }) + + b.process(context.Background(), msg) + if chat.sessionID != "" { + t.Fatalf("已有回合在跑时仍开了第二个回合：session=%q", chat.sessionID) + } + if got := len(hist.records()); got != 1 { + t.Fatalf("被并发拦下的消息登记 %d 条，want 1", got) + } + logged := logBuf.String() + if !strings.Contains(logged, "throttled by flow gate") || + !strings.Contains(logged, "source=services/channelbridge") { + t.Fatalf("并发拦下没有留下带 source 的可见日志：%q", logged) + } +} + +// TestNonOwnerLowValueSystemEventRegisteredOnly 钉住既有行为不变：系统/合成发送者的 +// 低价值事件只登记、绝不驱动完整循环（design/12 §6 策略一）。 +func TestNonOwnerLowValueSystemEventRegisteredOnly(t *testing.T) { + t.Parallel() + + chat := &stubChatter{res: &loop.Result{Text: "不该回"}} + sess := &stubSessions{sess: nonOwnerSession()} + hist := &captureHistory{} + b, err := New(Options{ + Loop: chat, Sessions: sess, Owner: &user.OwnerProfile{}, History: hist, + Now: func() time.Time { return time.Date(2026, 1, 1, 12, 0, 0, 0, time.UTC) }, + }) + if err != nil { + t.Fatalf("New: %v", err) + } + + msg := nonOwnerGroupMsg("群成员变更通知") + msg.Sender.Role = adapter.RoleSystem + b.process(context.Background(), msg) + + if chat.sessionID != "" || len(chat.messages) != 0 { + t.Fatalf("低价值系统事件驱动了循环：session=%q messages=%+v", chat.sessionID, chat.messages) + } + if got := len(hist.records()); got != 1 { + t.Fatalf("records = %d, want 1（只登记，不回复）", got) + } +} diff --git a/internal/services/channelbridge/transcript.go b/internal/services/channelbridge/transcript.go index fd7f778..1979a73 100644 --- a/internal/services/channelbridge/transcript.go +++ b/internal/services/channelbridge/transcript.go @@ -22,7 +22,9 @@ import ( // 看到的历史逐条一致 —— 她不会像第一次见到主人（主人 2026-10-05 要求两条车道环境一致）。 // - **群聊 / 陌生人（受限车道）**：播种**该隔离会话自己**（`sess-*`）的最近转录，因此群里 // 前几句、以及被最小打扰静默掉而只登记不回的那几条（light.go 的 registerSilenced 已把它们 -// 落进同一会话）都在她的上下文里，她据此才能理解一段连贯对话的意图。 +// 落进同一会话）都在她的上下文里，她据此才能理解一段连贯对话的意图。**顺序为 +// frame 先行**：来源行 + 第三方身份边界在最前、转录居中、当次消息在最后 +// （主人 2026-10-05 裁定，见 seedTurnTranscript）。 // // 两条车道读的会话不同，重建逻辑与条数上限却是同一份：internal/runtime/transcript // （ContextCount / Read / Message）—— 与网关同一份实现、同一个常量，绝不另写第二套 @@ -51,6 +53,13 @@ func turnEnv(sess session.Session, target inboundTarget, personaID string) loop. // - 主人私聊 → 主会话（`web`）的转录，与 Web 主会话逐条一致； // - 群聊 / 陌生人 → 该隔离会话（`sess-*`）的转录，含被静默登记的群成员发言。 // +// **受限车道的顺序是「frame 先行」**（主人 2026-10-05 裁定）：来源系统行 + +// 第三方身份边界必须排在**最前**，转录居中，当次用户消息在最后——即 +// `[边界/来源系统行, …转录…, 当次用户消息]`。理由是提示词注入面：若把转录 +// 前置在边界之前，群友的历史发言会先于「这是群、这些人是第三方」进入她的 +// 上下文。主人私聊车道**顺序逐字不变**（转录最前、来源行、当次消息），因为 +// 那条车道的上下文隔离本就不同。 +// // 时刻与网关一致：调用点在本回合用户消息落库**之前**（runInboundTurn 里装配先于 // persistToMain / persistIsolated），因此当次消息绝不会在上下文里出现两次。 // @@ -69,7 +78,27 @@ func (b *Bridge) seedTurnTranscript(ctx context.Context, target inboundTarget, m if len(seed) == 0 { return msgs } - out := make([]model.Message, 0, len(seed)+len(msgs)) + // 主人私聊：转录前置在最前（既有顺序，逐字不变）。 + if target.private { + out := make([]model.Message, 0, len(seed)+len(msgs)) + out = append(out, seed...) + return append(out, msgs...) + } + return insertSeedAfterFrame(msgs, seed) +} + +// insertSeedAfterFrame 把受限车道的转录插在**最前面的系统框架行之后** +// （来源行 + 第三方身份边界仍在最前，frame 先行），当次用户消息保持在最后： +// `[系统框架行, …转录…, 当次用户消息]`。受限车道的本回合消息恰好是 +// 「框架系统行 + 当次用户消息」，因此这一插入位置等价于「边界最前、历史居中、 +// 当次最后」，且**当次消息绝不重复**。 +func insertSeedAfterFrame(msgs, seed []model.Message) []model.Message { + frame := 0 + for frame < len(msgs) && msgs[frame].Role == model.RoleSystem { + frame++ + } + out := make([]model.Message, 0, len(msgs)+len(seed)) + out = append(out, msgs[:frame]...) out = append(out, seed...) - return append(out, msgs...) + return append(out, msgs[frame:]...) } diff --git a/internal/services/channelbridge/transcript_test.go b/internal/services/channelbridge/transcript_test.go index 1d569d2..80fa5a3 100644 --- a/internal/services/channelbridge/transcript_test.go +++ b/internal/services/channelbridge/transcript_test.go @@ -249,27 +249,29 @@ func TestGroupTurnSeedsItsOwnSessionTranscript(t *testing.T) { b.process(context.Background(), groupMessage("9", adapter.Actor{Role: adapter.RoleMember})) - // 形状：该群最近 3 条转录（最旧优先）+ 第三方边界/来源系统行 + 当次用户消息。 + // 形状（frame 先行）：第三方边界/来源系统行**在最前**，该群最近 3 条转录居中 + // （最旧优先），当次用户消息在最后。 if len(chat.msgs) != 5 { - t.Fatalf("loop messages = %d, want 5 (transcript 3 + boundary + user): %+v", len(chat.msgs), chat.msgs) + t.Fatalf("loop messages = %d, want 5 (boundary + transcript 3 + user): %+v", len(chat.msgs), chat.msgs) + } + // 边界行必须最先：历史里的陌生人发言绝不能在她还不知道「这是群、这些人是外人」 + // 之前进入上下文（提示词注入面，主人 2026-10-05 裁定）。 + if got := chat.msgs[0]; got.Role != model.RoleSystem || !strings.Contains(turnMessageText(got), thirdPartyBoundaryPrompt) || + !strings.Contains(turnMessageText(got), "群「Gang」") { + t.Fatalf("messages[0] = %+v, want the origin + third-party boundary system row first", got) } // 说话人归属逐条正确：两个不同说话人各归各，且走结构化字段（不是织进正文的标签）。 - if got := chat.msgs[0]; got.Role != model.RoleUser || turnMessageText(got) != "晚上要不要开黑" || got.SenderName != "张三" { - t.Fatalf("messages[0] = %+v, want 张三's earlier line attributed to 张三", got) + if got := chat.msgs[1]; got.Role != model.RoleUser || turnMessageText(got) != "晚上要不要开黑" || got.SenderName != "张三" { + t.Fatalf("messages[1] = %+v, want 张三's earlier line attributed to 张三", got) } - if got := chat.msgs[1]; got.Role != model.RoleUser || turnMessageText(got) != "我还在加班" || got.SenderName != "李四" { - t.Fatalf("messages[1] = %+v, want 李四's silenced line attributed to 李四", got) + if got := chat.msgs[2]; got.Role != model.RoleUser || turnMessageText(got) != "我还在加班" || got.SenderName != "李四" { + t.Fatalf("messages[2] = %+v, want 李四's silenced line attributed to 李四", got) } - if got := chat.msgs[2]; got.Role != model.RoleAssistant || turnMessageText(got) != "那你先忙" || got.SenderName != "cyrene" { - t.Fatalf("messages[2] = %+v, want her earlier reply attributed to cyrene", got) - } - // 第三方身份边界与来源行照旧（内容一个字没变，仍在每轮注入）。 - if got := chat.msgs[3]; got.Role != model.RoleSystem || !strings.Contains(turnMessageText(got), thirdPartyBoundaryPrompt) || - !strings.Contains(turnMessageText(got), "群「Gang」") { - t.Fatalf("messages[3] = %+v, want the origin + third-party boundary system row", got) + if got := chat.msgs[3]; got.Role != model.RoleAssistant || turnMessageText(got) != "那你先忙" || got.SenderName != "cyrene" { + t.Fatalf("messages[3] = %+v, want her earlier reply attributed to cyrene", got) } if got := chat.msgs[4]; got.Role != model.RoleUser || turnMessageText(got) != "hello all" || got.SenderName != "张三" { - t.Fatalf("messages[4] = %+v, want this turn's user message", got) + t.Fatalf("messages[4] = %+v, want this turn's user message last", got) } // 当次消息只出现一次：转录是在本回合消息落库**之前**读的。活夹具把已经落库的行也返回， // 因此播种时刻一旦挪到落库之后，这里立刻数出两次。 @@ -430,10 +432,18 @@ func TestOwnerGroupTurnSeedsOnlyItsOwnIsolatedTranscript(t *testing.T) { t.Fatalf("loop session id = %q, want the isolated sess-g", chat.sessionID) } if len(chat.msgs) != 3 { - t.Fatalf("loop messages = %d, want 3 (isolated transcript + origin/boundary + user): %+v", len(chat.msgs), chat.msgs) + t.Fatalf("loop messages = %d, want 3 (boundary + isolated transcript + user): %+v", len(chat.msgs), chat.msgs) } - if got := chat.msgs[0]; turnMessageText(got) != "群里的上一句" || got.SenderName != "张三" { - t.Fatalf("messages[0] = %+v, want the group's own earlier line", got) + // frame 先行：主人本人在群里同样是受限（群聊）车道 —— 边界/来源行仍在最前， + // 转录居中，当次消息在最后。 + if got := chat.msgs[0]; got.Role != model.RoleSystem || !strings.Contains(turnMessageText(got), thirdPartyBoundaryPrompt) { + t.Fatalf("messages[0] = %+v, want the boundary/origin system row first", got) + } + if got := chat.msgs[1]; turnMessageText(got) != "群里的上一句" || got.SenderName != "张三" { + t.Fatalf("messages[1] = %+v, want the group's own earlier line", got) + } + if got := chat.msgs[2]; got.Role != model.RoleUser { + t.Fatalf("messages[2] = %+v, want this turn's user message last", got) } for i, m := range chat.msgs { if strings.Contains(turnMessageText(m), "主人的私事") {