diff --git a/cmd/cyrene/main.go b/cmd/cyrene/main.go index e3f2a7d..eef82c7 100644 --- a/cmd/cyrene/main.go +++ b/cmd/cyrene/main.go @@ -99,6 +99,9 @@ func run() error { // 文档回收(切片 S7):启动期清扫一次 + 此后按 DOCUMENT_GC_INTERVAL 定时清扫。 // 它不等这一次跑完就去拉网关(见 wire_document_gc.go 的取舍 1)。 a.startDocumentGC(ctx) + // 渠道事件留存维护(切片 E3+E4):保留期清扫一次 + 此后定时清扫,内存计数 + // 按 CHANNEL_EVENT_COUNTER_FLUSH 定时落盘(退出前最后一次落盘在循环里)。 + a.startChannelEvents(ctx) // LLM 用量落库与保留期清理(切片 U1):单写者 + 有界队列的异步写入器, // 加上按 LLM_USAGE_RETENTION 的过期样本清理。 a.startUsage(ctx) diff --git a/cmd/cyrene/wire.go b/cmd/cyrene/wire.go index e357eae..d97229a 100644 --- a/cmd/cyrene/wire.go +++ b/cmd/cyrene/wire.go @@ -20,6 +20,7 @@ import ( personareg "git.yeij.top/AskaEth/Cyrene_v2/internal/persona/registry" "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/embedding" "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/exec" + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/events" "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/memory" @@ -95,6 +96,13 @@ type app struct { // docGC 是文档回收的常驻任务(切片 S7):未被引用且超过保留期的文档连同 // sidecar 与元数据行一起回收。GC 被配置关闭时为 nil(见 wire_document_gc.go)。 docGC *documentGC + // channelEvents 是渠道事件留存面(design/26 §4;切片 E3+E4):非消息事件逐条 + // 落行 / T1 计数都经它,桥接层由 wire_channels.go 注入。库未接时为 nil + // (桥接层随即走内存计数降级,见 wire_channel_events.go)。 + channelEvents *events.SQLStore + // channelEventKeeper 是事件留存的常驻维护任务(保留期清扫 + 计数落盘), + // 由 main 的 startChannelEvents 拉起(见 wire_channel_events.go)。 + channelEventKeeper *channelEventKeeper // usage 是 LLM 用量的异步落库出口(切片 U1):注入模型栈后,每次成功模型 // 调用都会把非敏感计数样本塞进它的有界队列。库未接时为 nil(= 只记日志)。 usage *model.UsageWriter @@ -183,6 +191,9 @@ func newApp(ctx context.Context, cfg *config.Config, log *slog.Logger) (*app, er return a, err } a.wirePlugins(ctx) + // 渠道事件留存面(design/26 §4;切片 E3+E4):必须在 wireChannels 之前构造 —— + // 桥接层在构造时就把它注入(Options.EventStore),运行时不再重取。 + a.wireChannelEvents() if err := a.wireChannels(); err != nil { return a, err } diff --git a/cmd/cyrene/wire_channel_events.go b/cmd/cyrene/wire_channel_events.go new file mode 100644 index 0000000..208a21d --- /dev/null +++ b/cmd/cyrene/wire_channel_events.go @@ -0,0 +1,188 @@ +package main + +// 本文件把**渠道事件留存**接进组合根(design/26 §4.3/§4.4;切片 E3+E4):按配置构造 +// 事件存储、交给桥接层注入(wire_channels.go 的 channelbridge.Options.EventStore), +// 拉起常驻的「保留期清扫 + 内存计数落盘」循环,并在退出前最后一次落盘。 +// +// 三条取舍逐字对齐参照实现 `wire_document_gc.go`(design/26 §4.4 明写「不另造」): +// +// 1. **启动期清扫不阻塞启动**:它是常驻 goroutine 的第一件事(紧随子系统启动之后 +// 开始,网关监听之前就绪),而不是在 newApp/run 里同步等它跑完。 +// 2. **失败不退出进程**:清扫 / 落盘失败只记一条 WARN 并等下一轮;一轮失败只是 +// 「这一轮没扫干净」,绝不让进程退出。 +// 3. **关闭 = 显式配 ≤0**:`CHANNEL_EVENT_RETENTION=0`(或负值)或 +// `CHANNEL_EVENT_GC_INTERVAL=0` 都不开启清扫,NewStore 会把关闭原因记进日志 +// (静默的「配了却不生效」不存在)。 +// +// ⚠️ 与参照实现的一处**已知差异**(如实登记,不假装对齐):docGC 的 `stop()` 由 +// `wire_services.go` 的 close 调用,而该文件由别的切片在改(本片不许动)。因此这里的 +// 退出路径做成**循环自己的 ctx.Done 分支**:ctx 取消即用独立超时的 context 落最后一次盘。 +// 语义与「退出前最后一次 Flush」相同,区别只在**不阻塞** close —— 计数本就是尽力而为的 +// 近似值(§4.3),并与关库并发(真撞上只记 WARN,不丢留存面本身)。 +// +// 三个键的读取走 `config.LoadDurationEnv`(进程环境 > `.env` > E1 导出的默认常量): +// `.env` 不写回进程环境,直接 os.Getenv 会让只写在 `.env` 里的声明静默失效。 + +import ( + "context" + "log/slog" + "time" + + "git.yeij.top/AskaEth/Cyrene_v2/internal/config" + "git.yeij.top/AskaEth/Cyrene_v2/internal/logger" + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/events" +) + +// 三个 env 键名:必须与 `.env.example` **逐字一致**(AGENTS §10 的 E6 条目)。 +const ( + channelEventRetentionEnv = "CHANNEL_EVENT_RETENTION" + channelEventGCIntervalEnv = "CHANNEL_EVENT_GC_INTERVAL" + channelEventCounterFlushEnv = "CHANNEL_EVENT_COUNTER_FLUSH" +) + +// channelEventFlushTimeout 是**退出前最后一次 Flush** 的时长上界。它必须配一个独立的 +// context:ctx 在这一刻已被取消,拿它去写库只会立刻拿到 context.Canceled —— 那正是 +// 「退出时最后一次 flush」最容易悄悄失效的地方。 +const channelEventFlushTimeout = 5 * time.Second + +// wireChannelEvents 构造事件留存面并挂到 app 上(随后由 wireChannels 注入桥接层)。 +// 它**不**启动任何 goroutine:常驻循环由 main 的 startChannelEvents 在子系统启动之后 +// 拉起(与文档 GC 的分工逐字相同)。 +func (a *app) wireChannelEvents() { + a.channelEvents = a.newChannelEventStore(envPath) +} + +// newChannelEventStore 按配置构造事件存储。库未接(`a.db == nil`,只可能出现在测试 +// 夹具里)时返回 nil:桥接层随即走「未接线」的 fail-visible 降级路径(内存计数 + +// 首次一条 DEBUG),入站流水线照常,绝不因为没有存储而失败或丢帧。 +// +// 默认值一律取 E1(`internal/runtime/events`)导出的常量 —— **不在这里另写一份字面量**: +// 两处各写一份 720h/1h/30s,改一处就会让「文档承诺的默认值」与「代码里的默认值」漂移。 +// `MaxPendingKeys` 没有 env 键(design/26 §4.4 只给了三个旋钮),同样取 E1 的默认常量。 +func (a *app) newChannelEventStore(envFilePath string) *events.SQLStore { + if a.db == nil { + a.log.Warn("channel event store skipped: no database connection") + return nil + } + store, err := events.NewStore(events.Config{ + DB: a.db, + Log: a.log, + FlushInterval: durationEnv(a.log, envFilePath, channelEventCounterFlushEnv, events.DefaultFlushInterval), + MaxPendingKeys: events.DefaultMaxPendingKeys, + GCInterval: durationEnv(a.log, envFilePath, channelEventGCIntervalEnv, events.DefaultGCInterval), + Retention: durationEnv(a.log, envFilePath, channelEventRetentionEnv, events.DefaultRetention), + }) + if err != nil { + // NewStore 唯一的失败是 DB 为 nil(上面已挡掉):仍如实处理,绝不吞错。 + a.log.Warn("channel event store unavailable", "err", err) + return nil + } + return store +} + +// durationEnv 读一个 duration 型 env 键(进程环境 > `.env` > def)。配了但读不懂时记 +// 一条带原始取值的 WARN 并回落 def —— 错的是那一个键,不该让整条留存面失效 +// (AGENTS §10:非法配置不许静默失效,但也不许一票否决)。 +func durationEnv(log *slog.Logger, envFilePath, key string, def time.Duration) time.Duration { + value, err := config.LoadDurationEnv(envFilePath, key, def) + if err != nil { + log.Warn("channel event env ignored, falling back to default", + "env", key, "error", err, "fallback", def.String()) + } + return value +} + +// channelEventStore 返回注入桥接层的留存面:未接线时返回**真正的 nil 接口**,而不是 +// 「装着 nil 指针的接口」—— 后者在 Go 里不等于 nil,会让桥接层的「未接线 ⇒ 内存计数」 +// 判据失效(症状是每帧都撞一次 nil 指针 panic)。 +func (a *app) channelEventStore() events.Store { + if a.channelEvents == nil { + return nil + } + return a.channelEvents +} + +// startChannelEvents 拉起留存维护循环(未接线时是空操作)。 +func (a *app) startChannelEvents(ctx context.Context) { + if a.channelEvents == nil { + a.log.Info("channel event upkeep disabled: store not wired", + "effect", "events are counted in memory only (bridged to the api layer counters)") + return + } + keeper := &channelEventKeeper{ + store: a.channelEvents, + log: logger.WithSource(a.log, "cmd/cyrene/wire_channel_events"), + done: make(chan struct{}), + } + a.channelEventKeeper = keeper + go keeper.loop(ctx) +} + +// channelEventKeeper 是渠道事件留存的常驻维护任务:保留期清扫 + 内存计数落盘 + +// 退出前最后一次落盘。它持有**存储句柄**(不是 app),因此生命周期与用法一目了然。 +type channelEventKeeper struct { + store *events.SQLStore + log *slog.Logger + // done 在循环退出时关闭。它让「退出前最后一次 Flush」**可被观测**:测试据此确认 + // 内存里的计数真的落了盘,而不是随 goroutine 一起消失。 + done chan struct{} +} + +// loop 是维护循环:启动期先清扫一次、落一次盘,此后按各自间隔定时跑,直到 ctx 被取消 +// (取消时落最后一次盘再退出)。 +func (k *channelEventKeeper) loop(ctx context.Context) { + defer close(k.done) + k.sweep(ctx) + k.flush(ctx) + + gcTicks, stopGC := k.gcTicker() + defer stopGC() + flushTicks := time.NewTicker(k.store.FlushInterval()) + defer flushTicks.Stop() + + for { + select { + case <-ctx.Done(): + k.flushOnShutdown() + return + case <-gcTicks: + k.sweep(ctx) + case <-flushTicks.C: + k.flush(ctx) + } + } +} + +// gcTicker 给出清扫的定时通道;GC 关闭时返回 **nil 通道**(select 上永不触发)。 +// `time.NewTicker(0)` 会 panic,所以「配了 ≤0 = 关闭」必须在**这里**就分岔,而不是 +// 建一个零间隔 ticker 再指望它不跑。 +func (k *channelEventKeeper) gcTicker() (<-chan time.Time, func()) { + if !k.store.Enabled() { + return nil, func() {} + } + ticker := time.NewTicker(k.store.Interval()) + return ticker.C, ticker.Stop +} + +// sweep 跑一轮保留期清扫。每轮的**成功**汇总由 store 自己记(E1 的 Sweep 已记一条, +// 含删除行数与批次),这里只负责把「这一轮整个没跑成」记成 WARN 并让循环继续。 +func (k *channelEventKeeper) sweep(ctx context.Context) { + if _, err := k.store.Sweep(ctx); err != nil { + k.log.Warn("channel event sweep failed", "err", err) + } +} + +// flush 把内存里待落盘的计数写进库;失败只 WARN(计数是尽力而为的近似值,§4.3)。 +func (k *channelEventKeeper) flush(ctx context.Context) { + if err := k.store.Flush(ctx); err != nil { + k.log.Warn("channel event counter flush failed", "err", err) + } +} + +// flushOnShutdown 是退出前的最后一次落盘:用独立超时的 context(此刻调用方的 ctx +// 已取消),失败只 WARN —— 退出路径绝不因为一次写库失败而 panic 或阻塞。 +func (k *channelEventKeeper) flushOnShutdown() { + ctx, cancel := context.WithTimeout(context.Background(), channelEventFlushTimeout) + defer cancel() + k.flush(ctx) +} diff --git a/cmd/cyrene/wire_channel_events_test.go b/cmd/cyrene/wire_channel_events_test.go new file mode 100644 index 0000000..32265a3 --- /dev/null +++ b/cmd/cyrene/wire_channel_events_test.go @@ -0,0 +1,341 @@ +package main + +// 本文件是**渠道事件留存组合根接线**的验收(design/26 §4.3/§4.4;切片 E3+E4)。 +// +// 两道并行的取证: +// +// 1. **接线存在性**(与 TestWireDocumentGC_IsWired 同一手法):组合根本身无法在单元 +// 测试里实例化(开库、迁移、绑端口,而后端还在运行),因此「存储真的被构造」「真的 +// 注入桥接层」「常驻循环真的被拉起」「退出前真的落最后一次盘」只能在源码层钉住 —— +// 少任何一处,留存都会**静默不工作**(事件照丢、日志里一句都没有),而这不 +// 会让任何既有用例变红。最后一段还核对 `.env.example` 的三个键(AGENTS §10)。 +// 2. **行为**:三个 env 键的默认值必须来自 E1 导出的常量(**不另写一份字面量**)、 +// 显式 ≤0 = 关闭清扫、只写在 `.env` 里也真实生效、非法取值回落默认 + WARN; +// 以及「退出前最后一次落盘」真的把内存里的计数写进库。 + +import ( + "bytes" + "context" + "log/slog" + "os" + "path/filepath" + "strings" + "testing" + "time" + + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/events" +) + +// TestWireChannelEvents_IsWired 钉住组合根的四个接缝 + `.env.example` 同步。 +func TestWireChannelEvents_IsWired(t *testing.T) { + t.Parallel() + + seams := map[string][]string{ + "wire_channel_events.go": { + // 存储真的被构造,且默认值取 E1 导出的常量(不另写字面量)。 + "events.NewStore(events.Config{", + "MaxPendingKeys: events.DefaultMaxPendingKeys", + "durationEnv(a.log, envFilePath, channelEventCounterFlushEnv, events.DefaultFlushInterval)", + "durationEnv(a.log, envFilePath, channelEventGCIntervalEnv, events.DefaultGCInterval)", + "durationEnv(a.log, envFilePath, channelEventRetentionEnv, events.DefaultRetention)", + // 未接线时交出**真 nil 接口**(装 nil 指针的接口不等于 nil,会让桥接层的 + // 「未接线 ⇒ 内存计数」降级判据失效)。 + "func (a *app) channelEventStore() events.Store {", + // GC 关闭时不许建零间隔 ticker(time.NewTicker(0) 会 panic)。 + "if !k.store.Enabled() {", + // 退出前最后一次落盘,用独立 ctx(此刻调用方的 ctx 已取消)。 + "k.flushOnShutdown()", + }, + "wire_channels.go": { + "EventStore: a.channelEventStore(),", + }, + "wire.go": { + // 必须在 wireChannels 之前构造:桥接层在构造时就注入它。 + "a.wireChannelEvents()", + }, + "main.go": { + // 常驻循环:没有这一行,留存面只存在于代码里而从不维护(清扫与计数落盘都不跑)。 + "a.startChannelEvents(ctx)", + }, + } + for file, wants := range seams { + src, err := os.ReadFile(file) + if err != nil { + t.Fatalf("read %s: %v", file, err) + } + for _, want := range wants { + if !strings.Contains(string(src), want) { + t.Fatalf("%s 里找不到 %q —— 渠道事件留存没有被接上组合根", file, want) + } + } + } + + // 三个新键必须与代码里的键名**逐字一致**(AGENTS §10 的 E6 条目)。 + src, err := os.ReadFile("wire_channel_events.go") + if err != nil { + t.Fatalf("read wire_channel_events.go: %v", err) + } + raw, err := os.ReadFile(filepath.Join("..", "..", ".env.example")) + if err != nil { + t.Fatalf("read .env.example: %v", err) + } + for _, key := range []string{channelEventRetentionEnv, channelEventGCIntervalEnv, channelEventCounterFlushEnv} { + if !strings.Contains(string(src), `"`+key+`"`) { + t.Fatalf("wire_channel_events.go 里没有键名常量 %q", key) + } + if !strings.Contains(string(raw), key+"=") { + t.Fatalf(".env.example 里找不到 %q(新 env 键必须同步模板,AGENTS §10)", key+"=") + } + } +} + +// channelEventTestApp 构造一个只带留存面所需依赖的 app(库 + 静默 logger)。 +func channelEventTestApp(t *testing.T, log *slog.Logger) *app { + t.Helper() + if log == nil { + log = testLog() + } + return &app{db: openMigratedTestDB(t, filepath.Join(t.TempDir(), "cyrene.db")), log: log} +} + +// unsetChannelEventEnv 把三个键从**进程环境**里拿掉并在用例结束后复原。 +// +// ⚠️ 不能用 `t.Setenv(key, "")` 代替:空值在优先级链里仍然「存在」,会**盖住** `.env` +// 里的同名声明(lookupRaw 先看 os.LookupEnv),而「只写在 .env 里也生效」正是要验的东西。 +func unsetChannelEventEnv(t *testing.T) { + t.Helper() + for _, key := range []string{channelEventRetentionEnv, channelEventGCIntervalEnv, channelEventCounterFlushEnv} { + key := key + prev, had := os.LookupEnv(key) + if had { + t.Cleanup(func() { _ = os.Setenv(key, prev) }) + } else { + t.Cleanup(func() { _ = os.Unsetenv(key) }) + } + if err := os.Unsetenv(key); err != nil { + t.Fatalf("unset %s: %v", key, err) + } + } +} + +// TestNewChannelEventStore_默认值取E1导出常量_正常:未配置时三个旋钮必须**就是** +// E1 导出的默认值(720h / 1h / 30s),并且清扫是开启的 —— 若这里另写了一份字面量, +// 「文档承诺的默认值」与「代码里的默认值」迟早漂移。 +func TestNewChannelEventStore_默认值取E1导出常量_正常(t *testing.T) { + unsetChannelEventEnv(t) + store := channelEventTestApp(t, nil).newChannelEventStore("") + if store == nil { + t.Fatal("有库时必须构造出事件存储") + } + if store.Retention() != events.DefaultRetention { + t.Fatalf("Retention = %v, want %v(E1 导出的默认常量)", store.Retention(), events.DefaultRetention) + } + if store.Interval() != events.DefaultGCInterval { + t.Fatalf("GC 间隔 = %v, want %v", store.Interval(), events.DefaultGCInterval) + } + if store.FlushInterval() != events.DefaultFlushInterval { + t.Fatalf("计数落盘间隔 = %v, want %v", store.FlushInterval(), events.DefaultFlushInterval) + } + if !store.Enabled() { + t.Fatal("未配置时清扫必须开启(漏配 != 关闭)") + } +} + +// TestNewChannelEventStore_显式零关闭清扫_边界:`≤0 = 关闭`(design/26 §4.4)—— +// 0 刻意**不**解释成「立即清扫」,那会让一次笔误抹掉唯一的取证面。 +func TestNewChannelEventStore_显式零关闭清扫_边界(t *testing.T) { + cases := []struct { + name string + key string + value string + }{ + {"保留期为 0", channelEventRetentionEnv, "0"}, + {"保留期为负", channelEventRetentionEnv, "-1h"}, + {"清扫间隔为 0", channelEventGCIntervalEnv, "0"}, + } + for _, tc := range cases { + tc := tc + t.Run(tc.name, func(t *testing.T) { + unsetChannelEventEnv(t) + t.Setenv(tc.key, tc.value) + store := channelEventTestApp(t, nil).newChannelEventStore("") + if store == nil { + t.Fatal("配置非法不该让存储整个消失") + } + if store.Enabled() { + t.Fatalf("%s=%s 时清扫仍开启,want 关闭(显式 ≤0 = 关闭)", tc.key, tc.value) + } + }) + } +} + +// TestNewChannelEventStore_只写在env文件里也生效_正常:`.env` 不写回进程环境,直接 +// os.Getenv 会**静默忽略**它(= 配了却不生效)。本用例把三个键只写进临时 `.env`, +// 取值必须原样生效。 +func TestNewChannelEventStore_只写在env文件里也生效_正常(t *testing.T) { + unsetChannelEventEnv(t) + path := filepath.Join(t.TempDir(), ".env") + body := strings.Join([]string{ + channelEventRetentionEnv + "=24h", + channelEventGCIntervalEnv + "=30m", + channelEventCounterFlushEnv + "=5s", + }, "\n") + "\n" + if err := os.WriteFile(path, []byte(body), 0o600); err != nil { + t.Fatalf("写临时 .env: %v", err) + } + + store := channelEventTestApp(t, nil).newChannelEventStore(path) + if store == nil { + t.Fatal("有库时必须构造出事件存储") + } + if store.Retention() != 24*time.Hour || store.Interval() != 30*time.Minute || store.FlushInterval() != 5*time.Second { + t.Fatalf("取值 = (%v, %v, %v), want (24h, 30m, 5s)", + store.Retention(), store.Interval(), store.FlushInterval()) + } +} + +// TestNewChannelEventStore_非法取值回落默认并告警_异常:配了但读不懂 ⇒ 回落 E1 默认值 +// **且**记一条带键名与原始取值的 WARN(AGENTS §10:非法配置不许静默失效,也不许一票 +// 否决整条留存面)。 +func TestNewChannelEventStore_非法取值回落默认并告警_异常(t *testing.T) { + unsetChannelEventEnv(t) + path := filepath.Join(t.TempDir(), ".env") + if err := os.WriteFile(path, []byte(channelEventRetentionEnv+"=三十天\n"), 0o600); err != nil { + t.Fatalf("写临时 .env: %v", err) + } + var buf bytes.Buffer + log := slog.New(slog.NewTextHandler(&buf, &slog.HandlerOptions{Level: slog.LevelDebug})) + + store := channelEventTestApp(t, log).newChannelEventStore(path) + if store == nil { + t.Fatal("单个键非法不该让存储整个消失") + } + if store.Retention() != events.DefaultRetention { + t.Fatalf("非法取值时的保留期 = %v, want %v", store.Retention(), events.DefaultRetention) + } + logged := buf.String() + if !strings.Contains(logged, "level=WARN") || !strings.Contains(logged, channelEventRetentionEnv) { + t.Fatalf("非法取值没有留下带键名的 WARN:%q", logged) + } +} + +// TestChannelEventStore_未接线时交出真nil接口_边界:未接数据库(或构造失败)时 +// `channelEventStore()` 必须返回**真 nil 接口** —— 返回「装着 nil 指针的接口」会让 +// 桥接层的 nil 判据失效,症状是每帧撞一次 nil 指针 panic,而单测全绿。 +func TestChannelEventStore_未接线时交出真nil接口_边界(t *testing.T) { + blank := &app{log: testLog()} + if store := blank.newChannelEventStore(""); store != nil { + t.Fatalf("库未接时不该构造存储:%+v", store) + } + var asInterface events.Store = blank.channelEventStore() + if asInterface != nil { + t.Fatalf("未接线时必须交出真 nil 接口,拿到的是 %#v", asInterface) + } +} + +// TestChannelEventKeeper_退出前最后一次落盘_正常:ctx 取消时循环必须用**独立**的 +// context 把内存里的计数写进库。这里把已取消的 ctx 直接交给 loop(同步跑完)—— +// 若 `flushOnShutdown` 用调用方那个已取消的 ctx 落盘,它会立刻拿到 context.Canceled, +// 待落盘的计数就会留在内存里(用例据此变红)。 +func TestChannelEventKeeper_退出前最后一次落盘_正常(t *testing.T) { + unsetChannelEventEnv(t) + a := channelEventTestApp(t, nil) + store := a.newChannelEventStore("") + if store == nil { + t.Fatal("有库时必须构造出事件存储") + } + + ctx := context.Background() + // 一条只计数的 T1 事件:它在内存里待落盘(逐条落行的那类会立刻写库,掩盖问题)。 + if err := store.Record(ctx, events.Event{ + Platform: events.PlatformQQ, + Tier: events.TierT1, + Category: events.CategoryHeartbeat, + PlatformType: "heartbeat", + }); err != nil { + t.Fatalf("Record: %v", err) + } + if store.PendingKeys() == 0 { + t.Fatal("计数没有留在内存里,用例前提不成立") + } + + cancelled, cancel := context.WithCancel(ctx) + cancel() + keeper := &channelEventKeeper{store: store, log: testLog(), done: make(chan struct{})} + keeper.loop(cancelled) + + if store.PendingKeys() != 0 { + t.Fatalf("退出时仍有 %d 个键待落盘:最后一次 Flush 没跑", store.PendingKeys()) + } + counters, err := store.Counters(ctx, events.PlatformQQ) + if err != nil { + t.Fatalf("Counters: %v", err) + } + if len(counters) != 1 || counters[0].Count != 1 { + t.Fatalf("退出后库里的计数 = %+v, want 一条 count=1 的心跳", counters) + } + select { + case <-keeper.done: + default: + t.Fatal("loop 退出时没有关闭 done(退出路径不可观测)") + } +} + +// TestChannelEventKeeper_GC关闭时不建零间隔ticker_边界:`time.NewTicker(0)` 会 panic, +// 所以「配了 ≤0 = 关闭清扫」必须在建 ticker 之前分岔(关闭 ⇒ nil 通道,select 上永不触发)。 +func TestChannelEventKeeper_GC关闭时不建零间隔ticker_边界(t *testing.T) { + t.Run("关闭时不建 ticker", func(t *testing.T) { + unsetChannelEventEnv(t) + t.Setenv(channelEventGCIntervalEnv, "0") + store := channelEventTestApp(t, nil).newChannelEventStore("") + if store == nil || store.Enabled() { + t.Fatalf("GC 应关闭:store=%v enabled=%v", store, store != nil && store.Enabled()) + } + ticks, stop := (&channelEventKeeper{store: store, log: testLog()}).gcTicker() + if ticks != nil { + t.Fatal("GC 关闭时不该建 ticker(零间隔 ticker 会 panic)") + } + stop() + }) + t.Run("开启时给出定时通道", func(t *testing.T) { + unsetChannelEventEnv(t) + store := channelEventTestApp(t, nil).newChannelEventStore("") + if store == nil || !store.Enabled() { + t.Fatal("默认配置下 GC 应开启") + } + ticks, stop := (&channelEventKeeper{store: store, log: testLog()}).gcTicker() + if ticks == nil { + t.Fatal("GC 开启时必须给出定时通道") + } + stop() + }) +} + +// TestChannelEventEnv_模板取值是合法duration_正常 是 E6(协同清单)自查的机械化版本: +// 三个键名必须能在 `.env.example` 里逐字找到,且模板值本身就是合法 duration(占位符形态, +// 从不放凭据)—— 否则照模板部署的新实例会得到一条 WARN 并**静默**回落到默认值。 +func TestChannelEventEnv_模板取值是合法duration_正常(t *testing.T) { + t.Parallel() + + raw, err := os.ReadFile(filepath.Join("..", "..", ".env.example")) + if err != nil { + t.Fatalf("read .env.example: %v", err) + } + lines := strings.Split(string(raw), "\n") + for _, key := range []string{channelEventRetentionEnv, channelEventGCIntervalEnv, channelEventCounterFlushEnv} { + found := false + for _, line := range lines { + value, ok := strings.CutPrefix(line, key+"=") + if !ok { + continue + } + found = true + if _, err := time.ParseDuration(value); err != nil { + t.Fatalf("%s 的模板值 %q 不是合法 duration", key, value) + } + } + if !found { + t.Fatalf(".env.example 缺少 %s(AGENTS §10:键名必须与代码逐字一致)", key) + } + } +} diff --git a/cmd/cyrene/wire_channels.go b/cmd/cyrene/wire_channels.go index c67b18d..5b5bdf2 100644 --- a/cmd/cyrene/wire_channels.go +++ b/cmd/cyrene/wire_channels.go @@ -103,6 +103,12 @@ func (a *app) wireChannels() error { 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")), diff --git a/internal/config/duration_env.go b/internal/config/duration_env.go new file mode 100644 index 0000000..cb65698 --- /dev/null +++ b/internal/config/duration_env.go @@ -0,0 +1,48 @@ +package config + +// 本文件承载「由**组合根**直接消费的 duration 型 env 键」的读取口径(切片 E3+E4 的 +// `CHANNEL_EVENT_*` 三个键是当前唯一调用方,design/26 §4.4)。 +// +// 为什么是读取函数,而不是给 Config 加字段: +// +// - Config 的反射解码只给出**值本身**,承载不了「未配置(用默认)」与「显式 ≤0 +// (关闭)」必须分开的三态语义 —— 而渠道事件 GC 的关闭判据正是 ≤0(design/26 §4.4 +// 写死「≤0 = 关闭」)。若用零值代表「未配置」,一次漏配就会把清扫关掉; +// - 直接 `os.Getenv` 会让**只写在 `.env` 里**的声明被静默忽略:`.env` 由 parseDotEnv +// 读进独立 map,并不写回进程环境(与 LoadContextBudgetRatio / LoadOwnerIdentityInjection +// 同一理由,「配了却不生效」是被本项目点名禁止的失效模式)。 +// +// 因此这里复用**同一套优先级**(进程环境 > `.env` > 默认值)与**同一份 `.env` 解析**, +// 两种写法都真实生效。envPath 为空表示不看 `.env`(与 Load 的约定一致)。 + +import ( + "fmt" + "strings" + "time" +) + +// LoadDurationEnv 读取一个 duration 型 env 键(Go duration 文本,如 `30s` / `1h` / +// `720h`,接受负值),按「进程环境 > .env > def」取值。键未配置或值为空白(`KEY=`) +// 时一律返回 def —— 「没写」与「写了空」都当没配,绝不把空串当成一个合法取值。 +// +// 返回的 error **只承载「配了但读不懂」**:此时返回值仍是 def(调用方据此回落), +// 但必须记一条带原始取值的 WARN 并继续 —— 非法配置不许静默失效(AGENTS §10)。 +func LoadDurationEnv(envPath, key string, def time.Duration) (time.Duration, error) { + fromEnv := map[string]string{} + if envPath != "" { + m, err := parseDotEnv(envPath) + if err != nil { + return def, fmt.Errorf("config: read %s: %w", envPath, err) + } + fromEnv = m + } + raw := strings.TrimSpace(lookupRaw(key, "", fromEnv)) + if raw == "" { + return def, nil + } + value, err := time.ParseDuration(raw) + if err != nil { + return def, fmt.Errorf("config: %s=%q 不是合法的 duration(如 30s / 1h / 720h): %w", key, raw, err) + } + return value, nil +} diff --git a/internal/config/duration_env_test.go b/internal/config/duration_env_test.go new file mode 100644 index 0000000..e84c5a5 --- /dev/null +++ b/internal/config/duration_env_test.go @@ -0,0 +1,108 @@ +package config + +// 本文件覆盖 LoadDurationEnv(切片 E3+E4 的 CHANNEL_EVENT_* 读取口径):三条判据 +// 各自的用例 —— **未配置回落默认**、**`.env` 只写在文件里也真实生效**(这正是不能用 +// os.Getenv 的原因)、**配了但读不懂 ⇒ 回落默认 + 报错**(绝不静默失效)。 +// +// 与 owner_identity_test.go 同一约定:读进程环境 / 写 .env 的用例**不并行** +// (unsetEnvForTest 会改动进程环境,与并行读取互相干扰)。 + +import ( + "path/filepath" + "testing" + "time" +) + +// TestLoadDurationEnv_未配置回落默认_正常:键既不在进程环境也不在 .env ⇒ 返回默认值、 +// 不报错;值为空白(`KEY=`)同样按「没配」处理。 +func TestLoadDurationEnv_未配置回落默认_正常(t *testing.T) { + const key = "CYRENE_TEST_DURATION_UNSET" + unsetEnvForTest(t, key) + + value, err := LoadDurationEnv("", key, 720*time.Hour) + if err != nil { + t.Fatalf("未配置时不该报错:%v", err) + } + if value != 720*time.Hour { + t.Fatalf("未配置时取值 = %v, want 720h(默认)", value) + } + + blank := writeDotEnv(t, key+"=") + value, err = LoadDurationEnv(blank, key, time.Hour) + if err != nil { + t.Fatalf("空白取值不该报错:%v", err) + } + if value != time.Hour { + t.Fatalf("空白取值时 = %v, want 1h(按未配置处理)", value) + } +} + +// TestLoadDurationEnv_进程环境优先_正常:优先级与其余配置一致(进程环境 > .env > 默认)。 +func TestLoadDurationEnv_进程环境优先_正常(t *testing.T) { + const key = "CYRENE_TEST_DURATION_PRECEDENCE" + t.Setenv(key, "45m") + + value, err := LoadDurationEnv(writeDotEnv(t, key+"=3h"), key, 30*time.Second) + if err != nil { + t.Fatalf("LoadDurationEnv: %v", err) + } + if value != 45*time.Minute { + t.Fatalf("取值 = %v, want 45m(进程环境优先于 .env)", value) + } +} + +// TestLoadDurationEnv_只写在env文件里也生效_正常:这是本函数存在的理由 —— `.env` 不写回 +// 进程环境,os.Getenv 会**静默忽略**它,而本函数必须读到(否则照模板部署的新实例漏配且 +// 静默失效,AGENTS §10 的 E6 条目)。显式 0 与负值也必须**原样抵达**:渠道事件 GC 正是 +// 靠 ≤0 判「显式关闭」(design/26 §4.4)。 +func TestLoadDurationEnv_只写在env文件里也生效_正常(t *testing.T) { + const key = "CYRENE_TEST_DURATION_DOTENV" + unsetEnvForTest(t, key) + + zero, err := LoadDurationEnv(writeDotEnv(t, "# 注释行", key+"=0"), key, 720*time.Hour) + if err != nil { + t.Fatalf("LoadDurationEnv(0): %v", err) + } + if zero != 0 { + t.Fatalf("取值 = %v, want 0(.env 里的显式 0 必须原样抵达)", zero) + } + + neg, err := LoadDurationEnv(writeDotEnv(t, key+"=-1h"), key, time.Hour) + if err != nil { + t.Fatalf("LoadDurationEnv(负值): %v", err) + } + if neg != -time.Hour { + t.Fatalf("负值 = %v, want -1h(负值必须原样抵达,不得被当成默认)", neg) + } +} + +// TestLoadDurationEnv_非法取值回落默认并报错_异常:配了但读不懂 ⇒ 返回默认值 **且**返回 +// 错误(调用方据此记 WARN 并继续),绝不静默失效。 +func TestLoadDurationEnv_非法取值回落默认并报错_异常(t *testing.T) { + const key = "CYRENE_TEST_DURATION_INVALID" + unsetEnvForTest(t, key) + + value, err := LoadDurationEnv(writeDotEnv(t, key+"=一小时"), key, 30*time.Second) + if err == nil { + t.Fatal("非法 duration 必须报错(AGENTS §10:非法配置不许静默失效)") + } + if value != 30*time.Second { + t.Fatalf("非法取值时的回落值 = %v, want 30s", value) + } +} + +// TestLoadDurationEnv_env文件不存在也不报错_边界:路径为空(不看 .env)与路径指向不存在的 +// 文件都不该让启动失败 —— 三个键都是可选旋钮。⚠️ 路径非空但文件缺失时 parseDotEnv 的行为 +// 由既有约定决定(缺失不算错误),本用例把它钉在「返回默认值」这一侧。 +func TestLoadDurationEnv_env文件不存在也不报错_边界(t *testing.T) { + const key = "CYRENE_TEST_DURATION_MISSING_DOTENV" + unsetEnvForTest(t, key) + + value, err := LoadDurationEnv(filepath.Join(t.TempDir(), "absent.env"), key, time.Hour) + if err != nil { + t.Fatalf("缺失的 .env 不该让读取失败:%v", err) + } + if value != time.Hour { + t.Fatalf("取值 = %v, want 1h(默认)", value) + } +} diff --git a/internal/services/channelbridge/channelbridge.go b/internal/services/channelbridge/channelbridge.go index 5fb3539..7c42b25 100644 --- a/internal/services/channelbridge/channelbridge.go +++ b/internal/services/channelbridge/channelbridge.go @@ -24,6 +24,7 @@ import ( "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/documents" + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/events" "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" @@ -162,6 +163,15 @@ type Options struct { // 空即视为**未接线**:带文件的消息照旧能跑(模型会收到一行「读不到」的诚实 // 提示),只是不落库、不抽取 —— 绝不假装读到了。 DocumentStore documents.StoreConfig + // EventStore 是**渠道事件留存**面(design/26 §4/§5;切片 E3):所有非消息事件 + // (含未知类型与解析失败)经 channelbridge/events.go 的 recordEvent **无条件** + // 写进它 —— T2/T3 与 unknown 逐条落行、T1 只累加计数。 + // + // 它只依赖 `events.Store` 接口,因此为 nil 时桥接器走**降级路径**(内存计数 + + // 首次一条 DEBUG),入站流水线**绝不**因为没有存储而失败或丢帧(fail-visible 而 + // 非 fail-crash,§4.3)。注入必须传**真 nil 接口**而不是装了 nil 指针的接口, + // 否则这条判据会失效(见 cmd/cyrene/wire_channel_events.go 的 channelEventStore)。 + EventStore events.Store } // Bridge 掌管渠道适配器生命周期与入站消息流水线。它 @@ -180,13 +190,19 @@ type Bridge struct { quietStartHour int quietEndHour int now func() time.Time - // noticeLog 是系统通知传入前审查(design/12 §6 策略一补充)的子功能日志器, - // 与 voiceLog 同款:`source=services/channelbridge/notice` 便于跨模块定位。 + // noticeLog 是系统通知传入前审查(design/12 §6 策略一补充)的子功能日志器: + // `source=services/channelbridge/notice` 便于跨模块定位。三个通知类日志 + // (notice / flow / event)都在构造期一次性标好来源,使「这条为什么没进她上下文」 + // 「这一帧有没有留下可查痕迹」在日志里可追溯。 noticeLog *slog.Logger // flowLog 是流控解析(design/12 §6 策略三,含账号级 directMsgPerMin)的子功能 // 日志器:`source=services/channelbridge/flowgate`,使「主人显式关掉某项流控」 // 在日志里可追溯(见 flowgate.go 的 directLimit)。 flowLog *slog.Logger + // eventLog 是渠道事件留存(design/26 §4;切片 E3)的子功能日志器: + // `source=services/channelbridge/events`。留存成功走 DEBUG(含 NoticeClass 的 + // 裁决结果),未接线 / 落库失败走 DEBUG(首帧)+ WARN(失败),全部带来源。 + eventLog *slog.Logger // 参与度引擎状态(design/18):按会话的内存状态机, // 加上约束它的角色提醒词 / 参与度参数,以及概率触发 @@ -219,6 +235,11 @@ type Bridge struct { // documentsCfg 是入站附件落盘配置(design/24 §3;切片 M1)。零值 = 未接线: // 带文件的消息照旧能跑,模型会收到一行「读不到」的诚实提示(绝不静默)。 documentsCfg documents.StoreConfig + // eventStore 是渠道事件留存面(design/26 §4;切片 E3)。**真 nil 接口** = 未接线: + // 非消息事件走内存计数降级,入站照常(见 events.go 的 recordEvent)。 + // eventFallback 是那条降级路径的计数面(有界,fail-visible)。 + eventStore events.Store + eventFallback *eventFallback // voiceSTT / voiceSTTTimeout / voiceCircuit 是入站语音转写面(切片 IM-ASR; // design/06 §1.1):注入点、独立短超时,以及按原因分类的**临时熔断**。 // voiceLog 是带 `services/channelbridge/voice` 来源的子功能日志器。 @@ -286,6 +307,7 @@ func New(opts Options) (*Bridge, error) { now: now, noticeLog: logger.WithSource(log, "services/channelbridge/notice"), flowLog: logger.WithSource(log, flowgateLogSource), + eventLog: logger.WithSource(log, "services/channelbridge/events"), triggerWords: opts.TriggerWords, personaPrompt: opts.PersonaPrompt, engage: opts.Engage, @@ -301,6 +323,8 @@ func New(opts Options) (*Bridge, error) { channelPersona: opts.ChannelPersona, adapterPersona: make(map[string]string), documentsCfg: opts.DocumentStore, + eventStore: opts.EventStore, + eventFallback: newEventFallback(), voiceSTT: opts.VoiceSTT, voiceSTTTimeout: voiceSTTTimeout(opts.VoiceSTTTimeout), voiceLog: logger.WithSource(log, "services/channelbridge/voice"), diff --git a/internal/services/channelbridge/events.go b/internal/services/channelbridge/events.go new file mode 100644 index 0000000..b5abbff --- /dev/null +++ b/internal/services/channelbridge/events.go @@ -0,0 +1,404 @@ +package channelbridge + +// 本文件承载**渠道事件留存**在桥接层的唯一接线点(design/26 §4/§5;切片 E3): +// 把适配器归一化出来的结构化事件**无条件**写进 `Options.EventStore`(T2/T3 与 unknown +// 逐条落行、T1 只累加计数),两轴各归各位 —— 留存档(本文件按归一化类别 + SelfRelated +// 映射,§2.3)与上下文档(notice.go 的 preEntryReview,既有)**互不决定**。 +// +// 三条不变量写在这里: +// +// 1. **留存无条件**:留存与「进不进上下文」是两个独立动作(§4.3)。因此本层在任何 +// 早退分支(仅登记 / fail-safe 静默 / 升级真会话)**之前**记账,且**绝不**决定 loop +// 走不走、也绝不改写消息内容。 +// 2. **绝不静默,也绝不 fail-crash**(§1.1/§4.3):未接线(store 为 nil)或落库失败 +// 都只降级为「内存计数 + 日志」——**绝不**让入站流水线失败、丢帧或 panic。 +// 这正是 §0.2 事故要的:看不见可以,但必须留得下痕、答得出「上游推过没有」。 +// 3. **脱敏唯一出处**(§4.2):摘要一律过 `adapter.SanitizeNoticeText`。`events.Event` +// 里**没有**承载平台正文 / 媒体 URL / OneBot `flag` 的字段,因此凭据不可能随这条 +// 链路流出去(§9.2 第 8 条:连结构都没有那些字段)。 +// +// 类别命名归 `adapter`(`adapter.NoticeCategory*`),本层**逐字搬运**、绝不改名 +// (§9.2 第 7 条:两轴常量分居两包,防两表漂移)。 + +import ( + "context" + "log/slog" + "strings" + "sync" + + "git.yeij.top/AskaEth/Cyrene_v2/internal/channel/adapter" + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/events" +) + +// 摘要的固定文案(design/26 §4.2 的「能落」清单:一句话常量,**不含任何平台原文**)。 +const ( + // heartbeatSummary / lifecycleSummary / selfEchoSummary 覆盖 T1 三类刻意留空 + // SafeText 的类别(§9.2 第 9 条:适配器留空,摘要由**留存侧**按平台原生类型生成)。 + heartbeatSummary = "心跳(链路遥测)" + lifecycleSummary = "生命周期(链路遥测)" + selfEchoSummary = "她自己发出消息的回声" + // decodeErrorSummary 对应 §3.7 的「JSON 坏」:把「解析失败」这件事本身说清楚。 + decodeErrorSummary = "事件载荷解析失败,只留结构痕迹" + // unknownSummaryPrefix 后面拼**平台原生类型码**(notice_type / extra.type / + // Telegram 的 update 键名):那是「上游到底报了什么类型」的答案(§2.4), + // 而且它是平台枚举值,不是用户文本。 + unknownSummaryPrefix = "未识别的事件类型:" + unknownSummary = "未识别的事件类型" + // platformEventSummary 是类别与种类都归不了类的兜底文案(既非未知类型、也无安全文本)。 + platformEventSummary = "平台系统事件" +) + +// recordEvent 是渠道事件留存的**唯一插入点**(design/26 §5.1/§6.2):在入站归一化 +// 之后、任何早退分支之前调用一次,把一条结构化事件交给 `Options.EventStore`。 +// +// 它只影响**留存面**:不改写 msg、不决定 loop 走不走、不返回值 —— 调用方照旧走既有 +// 流水线(仅登记 / 限定注入 / 升级真会话)。普通聊天消息(非 ContentAction)在这里 +// 直接返回:主消息路径**零副作用**。 +func (b *Bridge) recordEvent(ctx context.Context, msg adapter.UnifiedMessage, a adapter.PlatformAdapter, out noticeOutcome) { + if !out.Applicable { + return + } + ev := b.buildEvent(msg, a, out) + if b.eventStore == nil { + // 未接线(§4.3 的 fail-visible 降级):内存计数 + **首次**一条 DEBUG, + // 绝不中断入站、绝不丢帧。 + b.noteUnrecorded(ev, "no event store wired") + return + } + if err := b.eventStore.Record(ctx, ev); err != nil { + // 逐条留存的失败**必须可见**(§4.3):WARN + 内存计数。但既**不**中断入站, + // 也**不**关掉留存面 —— 下一帧照旧尝试。 + b.eventLog.Warn("channel event not persisted", + "platform", ev.Platform, "platform_type", ev.PlatformType, + "tier", string(ev.Tier), "category", string(ev.Category), "err", err) + b.noteUnrecorded(ev, "record failed") + return + } + b.logEventRecorded(ctx, ev, out) +} + +// buildEvent 把归一化结果投影成一条待留存事件(design/26 §4.1 的列,逐列对应)。 +// +// 投影**只取结构化字段**:平台 / 账号 / 会话 id、归一化类别与平台原生类型、当事人 id、 +// 是否命中她自己、关联消息 id,以及一句由安全文本(T2/T3)或固定文案(T1)得来的摘要。 +// CreatedAt 刻意留零值:由存储层填**我们的接收时刻**(适配器给的平台时刻可能是 0 ⇒ +// 1970,一条 1970 的行会被保留期立刻扫掉,反而丢掉刚发生的证据)。 +func (b *Bridge) buildEvent(msg adapter.UnifiedMessage, a adapter.PlatformAdapter, out noticeOutcome) events.Event { + notice := out.Notice + return events.Event{ + Platform: msg.Platform, + AccountID: accountIDFromAdapter(a), + Tier: eventTier(notice), + Category: eventCategory(notice), + PlatformType: notice.PlatformType, + ChannelID: msg.ChannelID, + ChannelType: string(msg.ChannelType), + GroupID: msg.GroupID, + ActorID: notice.Actor.UserID, + ActorRole: b.eventActorRole(msg.Platform, notice.Actor.UserID), + TargetID: notice.Target.UserID, + SelfRelated: notice.SelfRelated, + MessageID: firstNonEmptyStr(notice.RefMessage, msg.MessageID), + Summary: eventSummary(notice), + } +} + +// logEventRecorded 是这一帧成功进入留存面后的**结构级 DEBUG**(不留存则走 noteUnrecorded)。 +// +// 它记的是 §9.2 第 2 条要求的那条日志:**NoticeClass 的裁决结果只进日志、不入事件行** +// (§4.1 的 DDL 本就没有该列),连留存档一起记,方便回答「这条为什么没进她上下文」。 +// 只在真的开了 DEBUG 时才组装字段(与前缀一致,默认 INFO 下零开销), +// 且**绝不**记 SafeText 之外的平台文本,更绝不记 flag / 正文 / URL。 +func (b *Bridge) logEventRecorded(ctx context.Context, ev events.Event, out noticeOutcome) { + if !b.eventLog.Enabled(ctx, slog.LevelDebug) { + return + } + b.eventLog.Debug("channel event recorded", + "platform", ev.Platform, "platform_type", ev.PlatformType, + "category", string(ev.Category), "tier", string(ev.Tier), + "self_related", ev.SelfRelated, + "notice_class", noticeClassName(out.Class), "notice_reason", out.Reason) +} + +// noteUnrecorded 是「这一帧没能进入留存面」的唯一收口(未接线 / 落库失败): +// 内存计数**永远**累加(fail-visible:有数可查),日志只在首帧记一条 DEBUG —— +// 心跳每 ~15s 一帧,逐帧刷日志只会把日志淹没(AGENTS §10 的压测纪律:有界、不刷屏)。 +func (b *Bridge) noteUnrecorded(ev events.Event, reason string) { + stats := b.eventFallback.append(ev.Platform, ev.PlatformType) + if stats.Frames != 1 { + return + } + b.eventLog.Debug("channel event not recorded, counting in memory instead", + "reason", reason, "platform", ev.Platform, "platform_type", ev.PlatformType, + "tier", string(ev.Tier), "category", string(ev.Category)) +} + +// EventFallbackStats 是「没能进入留存面」的计数快照(design/26 §4.3 的降级面)。 +type EventFallbackStats struct { + // Keys 是按 `"/"` 聚合的帧数快照(与计数表的键同形)。 + Keys map[string]int64 + // Frames 是**未进入留存面**的总帧数(含下面溢出键的那部分)。 + Frames int64 + // OverflowKeys 是键上界用尽后未能按类记账的帧数:它们仍算进 Frames, + // 只是不再有各自的分类(键数**有界**,绝不无界增长)。 + OverflowKeys int64 +} + +// EventFallback 返回降级计数的快照。未接线(Options.EventStore 为 nil)或落库失败时 +// 才有内容 —— 它是「有多少帧没留下可查痕迹」的**唯一**答案:面板与排障据此知道留存面 +// 当时是活的还是降级的(design/26 §1.1 的诚实边界)。 +func (b *Bridge) EventFallback() EventFallbackStats { + return b.eventFallback.snapshot() +} + +// eventFallback 是未接线 / 落库失败时的内存计数面(§4.3 的降级)。 +// +// 键数上界复用 `events.DefaultMaxPendingKeys`(与存储层的内存计数缓冲同界:绝不因为 +// 一直失败而无界增长),键口径与计数表的唯一键同形(平台 + 平台原生类型;不含账号 +// —— 降级面连账号归属都未必取得准,键越简单越不会误导)。 +type eventFallback struct { + mu sync.Mutex + keys map[string]int64 + frames int64 + overflow int64 +} + +// newEventFallback 构造空的内存计数面。 +func newEventFallback() *eventFallback { + return &eventFallback{keys: make(map[string]int64)} +} + +// append 累加一帧并返回**当前**快照(日志据此判断「是不是首帧」)。 +func (f *eventFallback) append(platform, platformType string) EventFallbackStats { + key := platform + "/" + platformType + f.mu.Lock() + defer f.mu.Unlock() + f.frames++ + if _, ok := f.keys[key]; ok { + f.keys[key]++ + } else if len(f.keys) < events.DefaultMaxPendingKeys { + f.keys[key] = 1 + } else { + f.overflow++ + } + return f.snapshotLocked() +} + +// snapshot 返回快照(拷贝键值,调用方拿到的是私有副本)。 +func (f *eventFallback) snapshot() EventFallbackStats { + f.mu.Lock() + defer f.mu.Unlock() + return f.snapshotLocked() +} + +// snapshotLocked 是 snapshot 的加锁内核(append 已在锁内,直接复用)。 +func (f *eventFallback) snapshotLocked() EventFallbackStats { + keys := make(map[string]int64, len(f.keys)) + for k, v := range f.keys { + keys[k] = v + } + return EventFallbackStats{Keys: keys, Frames: f.frames, OverflowKeys: f.overflow} +} + +// eventTier 是 design/26 §2 的两轴换算:**归一化类别 → 留存档**(§2.3 的换算表)。 +// +// 判据顺序刻意固定为「先按类别定档、再按 self 命中升档」: +// +// 1. 类别(缺类别时按 Kind)给出基础档; +// 2. `SelfRelated`(结构化 id 命中她自己 / 她的账号)对**平台事件类**做 T1 → T2 升档 +// (§2.2:T1 → T2、T2 → 高价值;T3 不因 self 命中改变档位); +// 3. **帧级链路遥测不升档**(见 isFrameTelemetry):见那里的取舍说明。 +func eventTier(notice adapter.SystemNotice) events.Tier { + tier := baseEventTier(notice) + if tier != events.TierT1 || !notice.SelfRelated { + return tier + } + if isFrameTelemetry(notice.Category) { + return events.TierT1 + } + return events.TierT2 +} + +// baseEventTier 给出归一化类别的基础留存档(design/26 §3 逐平台映射表收敛后的共同口径)。 +// +// 类别为零值时按 Kind 兜底 —— `adapter.SystemNotice.Category` 的注释明写「零值表示 +// 适配器未给类别,留存侧按 Kind 兜底」。 +// +// default 分两支刻意都取**更全的留存**(T2 = 逐条落行):新增类别漏登记时会多落行, +// 绝不静默丢帧(design/26 §1.1)。E5 新增平台类别时**必须**在这里补档位。 +func baseEventTier(notice adapter.SystemNotice) events.Tier { + if notice.Category == "" { + return kindTier(notice.Kind) + } + switch notice.Category { + case adapter.NoticeCategoryHeartbeat, adapter.NoticeCategoryLifecycle, + adapter.NoticeCategoryMessageSent, adapter.NoticeCategoryClientStatus, + adapter.NoticeCategoryUnknown, adapter.NoticeCategoryDecodeError: + return events.TierT1 + case adapter.NoticeCategoryPoke, adapter.NoticeCategoryLuckyKing: + return events.TierT3 + default: + // join / friend_add / leave / recall / admin / ban / ban_all / upload / essence / + // card / offline_file / honor / title / friend_request / group_request:一律 T2。 + return events.TierT2 + } +} + +// kindTier 是类别缺失时按 Kind 兜底(与 §3 的逐条映射同口径):成员 / 权限 / 消息状态的 +// 变动与请求一律 T2,纯遥测 T1,定向交互 T3。default 取 T1 —— 此时类别会兜底成 +// `unknown`,而 `unknown` 在存储层本就是**逐条落行**的(§3.7),留痕一条不少。 +func kindTier(kind adapter.NoticeKind) events.Tier { + switch kind { + case adapter.NoticeInteraction: + return events.TierT3 + case adapter.NoticeJoin, adapter.NoticeLeave, adapter.NoticeRecall, + adapter.NoticeBotAdd, adapter.NoticeRequest: + return events.TierT2 + default: + return events.TierT1 + } +} + +// isFrameTelemetry 报告该类别是否属于**帧级链路遥测**:心跳(~15s 一帧)、生命周期、 +// 她自己发出消息的回声。这三类的 self 命中是**恒定**的(帧帧都命中 self_id),所以 +// 按 §2.2 升档等于「每帧写一行 channel_events」—— 而 §4.3 明令「心跳 / 在线状态逐帧 +// 写库是纯浪费」(计数表存在的理由),§3.1 给这三类的档位也正是 T1。 +// +// ⚠️ **由此登记的债**(design/26 §2.6「她自己账号的在线 / 离线」= T2 高价值): +// 那是**状态迁移**(在线→离线 / 离线→在线),不是每一帧。判定它需要迁移信号 +// (heartbeat 的 `status.online` / lifecycle 的 `sub_type`),而 `adapter.SystemNotice` +// 目前没有投影这两个字段(E2 定稿,本片不许改 plugins/**),`qq_obv11/probe.go` 的 +// 主动探活也还没有留存入点。本片只能保证**每一帧都进了计数**(「上游推过没有」答得上), +// 迁移事件(T2)留待后续切片;已在切片报告「未做 / 存疑」中列明。 +func isFrameTelemetry(cat adapter.NoticeCategory) bool { + switch cat { + case adapter.NoticeCategoryHeartbeat, adapter.NoticeCategoryLifecycle, + adapter.NoticeCategoryMessageSent: + return true + default: + return false + } +} + +// eventCategory 把适配器的归一化类别投影到 events 的类别轴。 +// +// 两轴常量分居两包(§9.2 第 7 条):**类别的命名归 adapter**,events 只承载类型与存储。 +// 这里逐字搬运、绝不改名 —— adapter 的类别比 events 的常量集**更细一档** +// (ban_all / honor / title / lucky_king …),改名只能靠丢信息来对齐,而丢掉的正是 +// 「这条为什么没进她上下文」的答案。适配器未给类别(零值)时兜底成 `unknown`: +// 「读不出具体类别」本身就该如实记为未知(存储层对 `unknown` **逐条落行**,§3.7)。 +func eventCategory(notice adapter.SystemNotice) events.Category { + cat := strings.TrimSpace(string(notice.Category)) + if cat == "" { + return events.CategoryUnknown + } + return events.Category(cat) +} + +// eventSummary 给出一条事件的**脱敏结构级摘要**(§4.1 的 summary 列)。 +// +// 适配器给了安全文本就用它(T2/T3 的固定文案,如「群文件上传」);T1 四类刻意留空 +// (§9.2 第 9 条)⇒ 由本函数按类别 / 平台原生类型**生成固定文案**。 +// +// 净化唯一出处是 `adapter.SanitizeNoticeText`(§4.2)。即便 SafeText 已在 +// preEntryReview 里净化过一遍,这里再过一次:保证「Event.Summary 恒已净化」与调用 +// 顺序无关(该函数幂等)。 +func eventSummary(notice adapter.SystemNotice) string { + text := strings.TrimSpace(notice.SafeText) + if text == "" { + text = fallbackSummary(notice) + } + return adapter.SanitizeNoticeText(text) +} + +// fallbackSummary 按归一化类别生成固定文案(T1 那几类 SafeText 恒为空)。 +// 只描述「这是什么类型的事件」,绝不夹带平台原文 / URL / 凭据;未知类型拼的是 +// **平台原生类型码**(平台枚举值,回答「上游报了哪种类型」)。 +func fallbackSummary(notice adapter.SystemNotice) string { + switch notice.Category { + case adapter.NoticeCategoryHeartbeat: + return heartbeatSummary + case adapter.NoticeCategoryLifecycle: + return lifecycleSummary + case adapter.NoticeCategoryMessageSent: + return selfEchoSummary + case adapter.NoticeCategoryDecodeError: + return decodeErrorSummary + case adapter.NoticeCategoryUnknown: + if notice.PlatformType != "" { + return unknownSummaryPrefix + notice.PlatformType + } + return unknownSummary + default: + return kindSummary(notice.Kind) + } +} + +// kindSummary 是类别与安全文本都缺时的兜底文案(按 Kind 说清是哪一类系统事件)。 +func kindSummary(kind adapter.NoticeKind) string { + switch kind { + case adapter.NoticeJoin: + return "有成员加入了会话" + case adapter.NoticeLeave: + return "有成员离开了会话" + case adapter.NoticeRecall: + return "有人撤回了一条消息" + case adapter.NoticeTyping: + return "有会话正在输入" + case adapter.NoticeBotAdd: + return "她的账号成员身份发生变更" + case adapter.NoticeRequest: + return "有一条好友 / 入群请求" + case adapter.NoticeInteraction: + return "有一个定向互动事件" + case adapter.NoticeUnknown: + return unknownSummary + default: + return platformEventSummary + } +} + +// eventActorRole 只按**白名单**判定事件当事人的角色(owner.yaml 的账号白名单 + +// 每渠道 adminIds 层,与 session.ResolveActor 同源),**绝不看正文、也绝不采信平台 +// 上报的 role 字段**(design/26 §1.2 / AGENTS §9 红线:身份只由结构化元数据 + 白名单 +// + 策略引擎决定)。 +// +// 三种取值:没有当事人的事件(心跳 / 链路遥测)来自平台系统本身 ⇒ system;白名单命中 +// ⇒ owner;其余 ⇒ **unknown**。刻意不填 member:「不是主人」不等于「是某个群的成员」 +// (加好友请求的申请人本就不是任何群的成员),判不出来时如实说判不出来。 +func (b *Bridge) eventActorRole(platform, actorID string) string { + if strings.TrimSpace(actorID) == "" { + return events.RoleSystem + } + if b.owner.IsOwner(platform, actorID) { + return events.RoleOwner + } + return events.RoleUnknown +} + +// accountIDFromAdapter 从适配器注册名里取配置账号 id("telegram:conf0" → "conf0", +// 多账号命名口径见 cmd/cyrene/channels.go)。单账号(名字里没有 ':')时返回空串: +// `events.Event.AccountID` 把空串定义为「单账号 / 未知」(§4.1),绝不编一个 id 出来。 +func accountIDFromAdapter(a adapter.PlatformAdapter) string { + _, accountID, ok := strings.Cut(adapterName(a), ":") + if !ok { + return "" + } + return accountID +} + +// noticeClassName 把 `adapter.NoticeClass`(上下文档的裁决结果)渲染成日志里的可读名。 +// 它**只进日志、不入事件行**(§9.2 第 2 条):留存档记 tier + category 两列, +// 这一轴是另一个正交轴的结论,回答「这条为什么没进她上下文」。 +func noticeClassName(class adapter.NoticeClass) string { + switch class { + case adapter.NoticeRegisterOnly: + return "register_only" + case adapter.NoticeLimitedInject: + return "limited_inject" + case adapter.NoticeEscalate: + return "escalate" + default: + return "unknown" + } +} diff --git a/internal/services/channelbridge/events_mapping_test.go b/internal/services/channelbridge/events_mapping_test.go new file mode 100644 index 0000000..d99dc52 --- /dev/null +++ b/internal/services/channelbridge/events_mapping_test.go @@ -0,0 +1,230 @@ +package channelbridge + +// 本文件是**两轴映射与脱敏红线**的验收(design/26 §2/§4.2;切片 E3): +// 类别 → 留存档的唯映射(含 self 命中升档与帧级遥测不升档)、类别逐字搬运、 +// 摘要净化且绝不夹带平台原文 / URL / OneBot `flag`、当事人角色只认白名单、 +// 账号 id 的解析口径。 +// +// 它与 events_test.go 的分工:那边验「留存这件事真的发生了」(落行 / 计数 / 降级), +// 这边验「记进去的东西对不对」(映射、脱敏、字段口径)。 + +import ( + "context" + "fmt" + "strings" + "testing" + + "git.yeij.top/AskaEth/Cyrene_v2/internal/channel/adapter" + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/events" +) + +// TestEventTier_两轴映射_正常:类别 → 留存档的唯映射(§2.1/§2.3/§3),含 +// `SelfRelated` 升档与「帧级遥测不升档」两侧。 +func TestEventTier_两轴映射_正常(t *testing.T) { + t.Parallel() + + cases := []struct { + name string + notice adapter.SystemNotice + want events.Tier + }{ + {"心跳是 T1", adapter.SystemNotice{Category: adapter.NoticeCategoryHeartbeat}, events.TierT1}, + {"生命周期是 T1", adapter.SystemNotice{Category: adapter.NoticeCategoryLifecycle}, events.TierT1}, + {"自身回声是 T1", adapter.SystemNotice{Category: adapter.NoticeCategoryMessageSent}, events.TierT1}, + {"他人在线状态是 T1", adapter.SystemNotice{Category: adapter.NoticeCategoryClientStatus}, events.TierT1}, + {"未知类型是 T1", adapter.SystemNotice{Category: adapter.NoticeCategoryUnknown}, events.TierT1}, + {"解析失败是 T1", adapter.SystemNotice{Category: adapter.NoticeCategoryDecodeError}, events.TierT1}, + {"群名片变更是 T2", adapter.SystemNotice{Category: adapter.NoticeCategoryCard}, events.TierT2}, + {"消息撤回是 T2", adapter.SystemNotice{Category: adapter.NoticeCategoryRecall}, events.TierT2}, + {"加好友请求是 T2", adapter.SystemNotice{Category: adapter.NoticeCategoryFriendRequest}, events.TierT2}, + {"全员禁言是 T2", adapter.SystemNotice{Category: adapter.NoticeCategoryBanAll}, events.TierT2}, + {"戳一戳是 T3", adapter.SystemNotice{Category: adapter.NoticeCategoryPoke}, events.TierT3}, + {"运气王是 T3", adapter.SystemNotice{Category: adapter.NoticeCategoryLuckyKing}, events.TierT3}, + {"他人存在性命中自己 ⇒ 升档 T2", + adapter.SystemNotice{Category: adapter.NoticeCategoryClientStatus, SelfRelated: true}, events.TierT2}, + {"资源类命中自己 ⇒ 升档 T2", + adapter.SystemNotice{Category: adapter.NoticeCategoryCard, SelfRelated: true}, events.TierT2}, + {"心跳命中自己 ⇒ 不升档(帧级遥测)", + adapter.SystemNotice{Category: adapter.NoticeCategoryHeartbeat, SelfRelated: true}, events.TierT1}, + {"生命周期命中自己 ⇒ 不升档(帧级遥测)", + adapter.SystemNotice{Category: adapter.NoticeCategoryLifecycle, SelfRelated: true}, events.TierT1}, + {"自身回声命中自己 ⇒ 不升档(帧级遥测)", + adapter.SystemNotice{Category: adapter.NoticeCategoryMessageSent, SelfRelated: true}, events.TierT1}, + {"类别缺失按 Kind 兜底:join ⇒ T2", + adapter.SystemNotice{Kind: adapter.NoticeJoin}, events.TierT2}, + {"类别缺失按 Kind 兜底:request ⇒ T2", + adapter.SystemNotice{Kind: adapter.NoticeRequest}, events.TierT2}, + {"类别缺失按 Kind 兜底:interaction ⇒ T3", + adapter.SystemNotice{Kind: adapter.NoticeInteraction}, events.TierT3}, + {"类别缺失按 Kind 兜底:typing ⇒ T1", + adapter.SystemNotice{Kind: adapter.NoticeTyping}, events.TierT1}, + } + for _, tc := range cases { + tc := tc + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + if got := eventTier(tc.notice); got != tc.want { + t.Fatalf("eventTier = %s, want %s", got, tc.want) + } + }) + } +} + +// TestEventCategory_逐字搬运适配器类别_正常:类别的命名归 adapter(§9.2 第 7 条), +// 留存侧只做类型转换、**绝不改名**(改名会让同一件事在适配器与库里各有一个名字)。 +// 类别为零值时兜底成 unknown —— 那正是「读不出具体类别」的诚实表达。 +func TestEventCategory_逐字搬运适配器类别_正常(t *testing.T) { + t.Parallel() + + cases := []struct { + category adapter.NoticeCategory + want events.Category + }{ + {adapter.NoticeCategoryBanAll, events.Category("ban_all")}, + {adapter.NoticeCategoryLuckyKing, events.Category("lucky_king")}, + {adapter.NoticeCategoryHonor, events.Category("honor")}, + {adapter.NoticeCategoryJoin, events.Category("join")}, + {adapter.NoticeCategoryUnknown, events.CategoryUnknown}, + {adapter.NoticeCategoryDecodeError, events.CategoryDecodeError}, + {"", events.CategoryUnknown}, + } + for _, tc := range cases { + tc := tc + t.Run(string(tc.want), func(t *testing.T) { + t.Parallel() + if got := eventCategory(adapter.SystemNotice{Category: tc.category}); got != tc.want { + t.Fatalf("eventCategory(%q) = %q, want %q", tc.category, got, tc.want) + } + }) + } +} + +// TestEventSummary_净化且绝不夹带平台原文_正常:§4.2 的红线逐条钉—— +// 正文 / 平台动作凭据(OneBot `request.flag`)/ 媒体 URL 绝不入事件(连结构都没有那些 +// 字段),摘要必须过 `adapter.SanitizeNoticeText`(唯一净化出处)。 +func TestEventSummary_净化且绝不夹带平台原文_正常(t *testing.T) { + t.Parallel() + ctx := context.Background() + + const ( + secretFlag = "SECRET-FLAG-9f2c" + bodyText = "这是群消息正文片段" + mediaURL = "https://cdn.example.com/f.mp4?sign=abc" + ) + raw := []byte(`{"post_type":"request","request_type":"friend","flag":"` + secretFlag + + `","comment":"` + bodyText + `","file":{"url":"` + mediaURL + `"}}`) + + msg := groupNoticeMessage() + msg.RawData, msg.Content = raw, bodyText + dirty := "群文件\u200b上传\x1b[31m\x1b[0m\n\x07" + strings.Repeat("啊", 300) + ad := eventAdapter("qq-obv11", adapter.SystemNotice{ + Kind: adapter.NoticeRequest, Category: adapter.NoticeCategoryFriendRequest, + PlatformType: "friend", SafeText: dirty, HighValue: true, + }) + b, _, _, _ := eventBridgeFixture(t, ad, nil) + + out := b.preEntryReview(ctx, &msg, ad) + ev := b.buildEvent(msg, ad, out) + + // 逐字段取证:把整条事件打成一行文本,验证三个红线取值一个都不在里面。 + dump := fmt.Sprintf("%+v", ev) + for _, secret := range []string{secretFlag, bodyText, mediaURL} { + if strings.Contains(dump, secret) { + t.Fatalf("事件里出现了绝不该落的内容 %q:%s", secret, dump) + } + } + // 第二次净化是**本层**的责任(§4.2 的「脱敏唯一出处」),不能靠调用顺序兜底: + // 即便 SafeText 未经 preEntryReview(顺序变了、或别的调用方直接问它), + // `Event.Summary` 也必须已经净化。未知类型的兜底摘要(纯文本由本层生成)同样如此。 + for _, notice := range []adapter.SystemNotice{ + {Category: adapter.NoticeCategoryUnknown, PlatformType: "weird\x1b[31m\u200btype"}, + {Kind: adapter.NoticeOther, SafeText: dirty}, + } { + got := eventSummary(notice) + if strings.ContainsAny(got, "\u200b\x1b\x07\n") { + t.Fatalf("eventSummary 没净化(依赖调用顺序会让这条不变量随时失效):%q", got) + } + if got == "" { + t.Fatalf("eventSummary 生成了空摘要:%+v", notice) + } + } + if strings.ContainsAny(ev.Summary, "\u200b\x1b\x07\n") { + t.Fatalf("摘要未净化(含零宽/控制字符):%q", ev.Summary) + } + if len([]rune(ev.Summary)) > adapter.NoticeSafeTextMaxRunes+1 { + t.Fatalf("摘要未按 NoticeSafeTextMaxRunes 截断:%d runes", len([]rune(ev.Summary))) + } + if !strings.Contains(ev.Summary, "群文件上传") { + t.Fatalf("摘要 = %q, want 保留安全文本的可读部分", ev.Summary) + } +} + +// TestEventActorRole_只认白名单_正常:事件当事人的角色**只**由白名单判定(与 +// session.ResolveActor 同源),绝不看正文、也绝不采信平台上报的 role 字段; +// 「不是主人」不等于「是某个群的成员」,故非主人一律如实记 unknown。 +func TestEventActorRole_只认白名单_正常(t *testing.T) { + t.Parallel() + + b, _, _, _ := eventBridgeFixture(t, eventAdapter("qq-obv11", adapter.SystemNotice{}), nil) + cases := []struct { + name string + actorID string + want string + }{ + {"白名单命中 ⇒ owner", "1000", events.RoleOwner}, + {"不在白名单 ⇒ unknown(判不出来就说判不出来)", "2000", events.RoleUnknown}, + {"无当事人 ⇒ system(事件来自平台系统本身)", "", events.RoleSystem}, + } + for _, tc := range cases { + tc := tc + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + if got := b.eventActorRole("qq-obv11", tc.actorID); got != tc.want { + t.Fatalf("eventActorRole(%q) = %q, want %q", tc.actorID, got, tc.want) + } + }) + } +} + +// TestAccountIDFromAdapter_多账号与单账号_正常:账号 id 从适配器注册名解析 +// (`platform:confID`);单账号与 nil 适配器一律空串 —— 空串在 events 里定义为 +// 「单账号 / 未知」,绝不编一个 id 出来。 +func TestAccountIDFromAdapter_多账号与单账号_正常(t *testing.T) { + t.Parallel() + + cases := []struct { + name string + ad adapter.PlatformAdapter + want string + }{ + {"多账号取配置 id", ¬iceAdapter{fakeAdapter: fakeAdapter{name: "qq-obv11:conf0"}}, "conf0"}, + {"单账号留空", ¬iceAdapter{fakeAdapter: fakeAdapter{name: "qq-obv11"}}, ""}, + {"无适配器留空", nil, ""}, + } + for _, tc := range cases { + tc := tc + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + if got := accountIDFromAdapter(tc.ad); got != tc.want { + t.Fatalf("accountIDFromAdapter = %q, want %q", got, tc.want) + } + }) + } +} + +// TestNoticeClassName_可读名_正常:上下文档(NoticeClass)的裁决结果只进日志、不入事件行 +// (§9.2 第 2 条),因此它的可读名是日志可读性的唯一依赖。 +func TestNoticeClassName_可读名_正常(t *testing.T) { + t.Parallel() + + cases := map[adapter.NoticeClass]string{ + adapter.NoticeRegisterOnly: "register_only", + adapter.NoticeLimitedInject: "limited_inject", + adapter.NoticeEscalate: "escalate", + } + for class, want := range cases { + if got := noticeClassName(class); got != want { + t.Fatalf("noticeClassName(%d) = %q, want %q", class, got, want) + } + } +} diff --git a/internal/services/channelbridge/events_test.go b/internal/services/channelbridge/events_test.go new file mode 100644 index 0000000..137510b --- /dev/null +++ b/internal/services/channelbridge/events_test.go @@ -0,0 +1,316 @@ +package channelbridge + +// 本文件是**渠道事件留存接线**的验收(design/26 §4/§5;切片 E3):唯一插入点 +// `recordEvent` 的三条要件各有用例 —— +// +// 1. **T1 只计数、不进上下文**(§4.3/§3.1); +// 2. **unknown 不进上下文但逐条落行**(§3.7「落库 + 计数 + 绝不进上下文」); +// 3. **未接线 / 落库失败时入站照常**(§4.3 的 fail-visible 降级,绝不中断入站)。 +// +// 另加一道红线钉:普通聊天消息在留存面**零副作用**(主消息路径不受影响)。 +// 两轴映射(类别 → 留存档)与「摘要必须净化、绝不夹带平台原文」另在 events_mapping_test.go。 +// +// 行/计数的判据本身归 E1(`internal/runtime/events` 的 persistsAsRow 与计数缓冲), +// 这里刻意用**真的 SQLite 存储**(内存库 + 迁移 0021)做端到端断言,而不是假存储: +// 否则「T1 不落行」「unknown 落行」会退化成「我们调了 Record」这种同义反复。 + +import ( + "bytes" + "context" + "errors" + "log/slog" + "strings" + "testing" + + "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/db" + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/events" + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/loop" +) + +// newTestEventStore 打开一个带迁移 0021 的内存库并构造事件存储(E1 的 SQLStore)。 +// GC 关闭(保留期与间隔都为 0):本文件验的是写入面,清扫归 E1 自己的用例。 +func newTestEventStore(t *testing.T) *events.SQLStore { + t.Helper() + ctx := context.Background() + conn, err := db.Open(ctx, ":memory:", nil) + if err != nil { + t.Fatalf("打开内存库: %v", err) + } + t.Cleanup(func() { _ = conn.Close() }) + if err := db.MigrateUp(ctx, conn, nil); err != nil { + t.Fatalf("内存库跑迁移: %v", err) + } + store, err := events.NewStore(events.Config{DB: conn, GCInterval: 0, Retention: 0}) + if err != nil { + t.Fatalf("events.NewStore: %v", err) + } + return store +} + +// eventBridgeFixture 组装一条「系统事件入站」所需的桥:受限群会话 + 捕获历史 + +// 记录调用的循环 + 给定的留存面(nil = 未接线)+ 捕获到 buffer 的 DEBUG 日志 +// (用它断言日志带 source、且降级只记一条 DEBUG)。 +func eventBridgeFixture(t *testing.T, ad adapter.PlatformAdapter, store events.Store) (*Bridge, *stubChatter, *captureHistory, *bytes.Buffer) { + t.Helper() + chat := &stubChatter{res: &loop.Result{Text: "嗯"}} + sess := &stubSessions{sess: session.Session{SessionID: "sess-g1", IsGroup: true}} + hist := &captureHistory{} + var buf bytes.Buffer + log := slog.New(slog.NewTextHandler(&buf, &slog.HandlerOptions{Level: slog.LevelDebug})) + b, err := New(Options{ + Log: log, Loop: chat, Sessions: sess, History: hist, + Owner: testOwnerProfile("qq-obv11", "1000"), EventStore: store, + }) + if err != nil { + t.Fatalf("New: %v", err) + } + b.Add(ad) + return b, chat, hist, &buf +} + +// eventAdapter 是「按给定归一化结果回答」的最小适配器:它复用 notice_test.go 的 +// noticeAdapter,只多一层固定适配器名(多账号命名口径:`platform:confID`)。 +func eventAdapter(name string, notice adapter.SystemNotice) *noticeAdapter { + return ¬iceAdapter{fakeAdapter: fakeAdapter{name: name}, notice: notice, ok: true} +} + +// failingEventStore 是**一直接着失败**的留存面(库不可用 / 表缺失):用于钉住 +// 「落库失败只降级,绝不中断入站」(§4.3)。计数不加锁:本文件的调用都在测试 +// goroutine 上同步发生。 +type failingEventStore struct { + records int + flushes int +} + +func (f *failingEventStore) Record(context.Context, events.Event) error { + f.records++ + return errors.New("storage unavailable") +} +func (f *failingEventStore) Count(context.Context, events.Filter) (int64, error) { return 0, nil } +func (f *failingEventStore) List(context.Context, events.Filter) ([]events.Event, error) { + return nil, nil +} +func (f *failingEventStore) Counters(context.Context, string) ([]events.Counter, error) { + return nil, nil +} +func (f *failingEventStore) Flush(context.Context) error { + f.flushes++ + return nil +} +func (f *failingEventStore) Sweep(context.Context) (events.SweepStats, error) { + return events.SweepStats{}, nil +} + +// TestRecordEvent_T1只计数不进上下文_正常:心跳是典型的 T1 遥测 —— 它必须**进计数** +// (「上游到底推没推」的唯一答案),但**绝不**逐条落行(每 ~15s 一帧会把库写胖,§4.3), +// 也绝不进她的上下文。用例让心跳带着 `SelfRelated=true`(真实心跳帧帧都带 self_id): +// 它同时把「帧级遥测不因 self 命中升档」这条取舍钉在这里。 +func TestRecordEvent_T1只计数不进上下文_正常(t *testing.T) { + t.Parallel() + ctx := context.Background() + store := newTestEventStore(t) + + ad := eventAdapter("qq-obv11:conf0", adapter.SystemNotice{ + Kind: adapter.NoticeOther, + Category: adapter.NoticeCategoryHeartbeat, + PlatformType: "heartbeat", + Ignorable: true, + SelfRelated: true, + }) + b, chat, hist, _ := eventBridgeFixture(t, ad, store) + + b.processFrom(ctx, groupNoticeMessage(), ad) + + if chat.sessionID != "" { + t.Fatalf("心跳驱动了 loop:session=%q", chat.sessionID) + } + if recs := hist.records(); len(recs) != 0 { + t.Fatalf("心跳被写进了登记面(会话历史):%+v", recs) + } + rows, err := store.Count(ctx, events.Filter{}) + if err != nil { + t.Fatalf("Count: %v", err) + } + if rows != 0 { + t.Fatalf("T1 遥测逐条落行 %d 行,want 0(§4.3:心跳逐帧写库是纯浪费)", rows) + } + + if err := store.Flush(ctx); err != nil { + t.Fatalf("Flush: %v", err) + } + counters, err := store.Counters(ctx, "qq-obv11") + if err != nil { + t.Fatalf("Counters: %v", err) + } + if len(counters) != 1 { + t.Fatalf("计数行数 = %d, want 1(T1 必须进计数)", len(counters)) + } + got := counters[0] + if got.Count != 1 || got.Tier != events.TierT1 || got.Category != events.Category(adapter.NoticeCategoryHeartbeat) { + t.Fatalf("计数 = %+v, want count=1 tier=t1 category=heartbeat", got) + } + if got.AccountID != "conf0" || got.PlatformType != "heartbeat" { + t.Fatalf("计数键 = (account=%q, type=%q), want (conf0, heartbeat)", got.AccountID, got.PlatformType) + } +} + +// TestRecordEvent_unknown不进上下文但落行_正常:未知类型是「兜底不等于放行」的正面用例 +// (§3.7)—— 它照旧逐条落行(可查)并进计数,但**绝不**进她的上下文,且平台原生类型码 +// 必须原样保留(那是「上游到底报了什么类型」的唯一凭据)。 +func TestRecordEvent_unknown不进上下文但落行_正常(t *testing.T) { + t.Parallel() + ctx := context.Background() + store := newTestEventStore(t) + + ad := eventAdapter("qq-obv11:conf0", adapter.SystemNotice{ + Kind: adapter.NoticeUnknown, + Category: adapter.NoticeCategoryUnknown, + PlatformType: "mystery_frame", + Ignorable: true, + }) + b, chat, hist, _ := eventBridgeFixture(t, ad, store) + + b.processFrom(ctx, groupNoticeMessage(), ad) + + if chat.sessionID != "" { + t.Fatalf("unknown 事件进了真会话:session=%q", chat.sessionID) + } + if recs := hist.records(); len(recs) != 0 { + t.Fatalf("unknown 事件被登记进会话历史:%+v", recs) + } + rows, err := store.List(ctx, events.Filter{}) + if err != nil { + t.Fatalf("List: %v", err) + } + if len(rows) != 1 { + t.Fatalf("unknown 事件落行数 = %d, want 1(unknown 必须逐条留痕)", len(rows)) + } + ev := rows[0] + if ev.Tier != events.TierT1 || ev.Category != events.CategoryUnknown { + t.Fatalf("落行档位/类别 = (%s,%s), want (t1,unknown)", ev.Tier, ev.Category) + } + if ev.PlatformType != "mystery_frame" { + t.Fatalf("平台原生类型码 = %q, want mystery_frame(原样保留是排障凭据)", ev.PlatformType) + } + if ev.Platform != "qq-obv11" || ev.ChannelID != "g1" || ev.ChannelType != string(adapter.ChannelGroup) { + t.Fatalf("行来源字段 = (platform=%q, channel=%q, type=%q)", ev.Platform, ev.ChannelID, ev.ChannelType) + } + if ev.AccountID != "conf0" { + t.Fatalf("account_id = %q, want conf0(多账号覆盖;空串留给单账号/未知)", ev.AccountID) + } + if !strings.Contains(ev.Summary, "mystery_frame") { + t.Fatalf("摘要 = %q, want 含未识别的平台原生类型码", ev.Summary) + } +} + +// TestRecordEvent_未接线时入站照常_异常:`Options.EventStore == nil` 时降级为 +// 「内存计数 + 首次一条 DEBUG」,**绝不**中断入站 —— 高价值事件的升级真会话行为必须 +// 逐字不变(fail-visible 而非 fail-crash,§4.3)。 +func TestRecordEvent_未接线时入站照常_异常(t *testing.T) { + t.Parallel() + ctx := context.Background() + + ad := eventAdapter("qq-obv11", adapter.SystemNotice{ + Kind: adapter.NoticeLeave, + Category: adapter.NoticeCategoryLeave, + PlatformType: "group_decrease", + SafeText: "系统通知:她(机器人账号)被移出了群聊", + HighValue: true, + SelfRelated: true, + }) + b, chat, _, buf := eventBridgeFixture(t, ad, nil) + + b.processFrom(ctx, groupNoticeMessage(), ad) + + if chat.sessionID != "sess-g1" { + t.Fatalf("未接线改变了入站行为:高价值事件没进真会话(session=%q)", chat.sessionID) + } + stats := b.EventFallback() + if stats.Frames != 1 { + t.Fatalf("降级计数 = %d, want 1(未接线的帧也必须留下可查的计数)", stats.Frames) + } + if stats.Keys["qq-obv11/group_decrease"] != 1 { + t.Fatalf("降级计数键 = %v, want qq-obv11/group_decrease=1", stats.Keys) + } + logged := buf.String() + if !strings.Contains(logged, "source=services/channelbridge/events") { + t.Fatalf("降级日志没有来源标记(AGENTS §10 禁无来源裸日志):%q", logged) + } + if !strings.Contains(logged, "channel event not recorded") { + t.Fatalf("未接线时没有留下可见痕迹:%q", logged) + } + + // 「DEBUG 一次」:第二帧不再刷日志,但计数继续累加(有界、不刷屏)。 + buf.Reset() + b.processFrom(ctx, groupNoticeMessage(), ad) + if again := b.EventFallback().Frames; again != 2 { + t.Fatalf("降级计数 = %d, want 2", again) + } + if got := buf.String(); strings.Contains(got, "channel event not recorded") { + t.Fatalf("降级日志被逐帧刷写(应只在首帧记一条 DEBUG):%q", got) + } +} + +// TestRecordEvent_落库失败仍不中断入站_异常:存储报错 ⇒ WARN + 内存计数 + 继续 —— +// **不关留存面、不中断回合、不让入站失败**(§4.3)。 +func TestRecordEvent_落库失败仍不中断入站_异常(t *testing.T) { + t.Parallel() + ctx := context.Background() + store := &failingEventStore{} + + ad := eventAdapter("qq-obv11", adapter.SystemNotice{ + Kind: adapter.NoticeLeave, + Category: adapter.NoticeCategoryLeave, + PlatformType: "group_decrease", + SafeText: "系统通知:她(机器人账号)被移出了群聊", + HighValue: true, + }) + b, chat, _, buf := eventBridgeFixture(t, ad, store) + + b.processFrom(ctx, groupNoticeMessage(), ad) + + if store.records != 1 { + t.Fatalf("留存面被调用 %d 次, want 1", store.records) + } + if chat.sessionID != "sess-g1" { + t.Fatalf("落库失败中断了入站(高价值事件没进真会话):session=%q", chat.sessionID) + } + if stats := b.EventFallback(); stats.Frames != 1 { + t.Fatalf("落库失败的帧没有进降级计数:%+v", stats) + } + logged := buf.String() + if !strings.Contains(logged, "level=WARN") || + !strings.Contains(logged, "channel event not persisted") || + !strings.Contains(logged, "source=services/channelbridge/events") { + t.Fatalf("落库失败的可观测面不足(要 WARN + source):%q", logged) + } +} + +// TestRecordEvent_聊天消息零副作用_正常:普通聊天消息不是渠道事件 —— 留存面**一次都 +// 不该被调用**(主消息路径不受影响,§5.1)。 +func TestRecordEvent_聊天消息零副作用_正常(t *testing.T) { + t.Parallel() + ctx := context.Background() + store := newTestEventStore(t) + + ad := eventAdapter("qq-obv11", adapter.SystemNotice{Kind: adapter.NoticeJoin}) + b, _, _, _ := eventBridgeFixture(t, ad, store) + + msg := groupNoticeMessage() + msg.ContentType, msg.Content = adapter.ContentText, "大家好" + msg.Sender.Role, msg.SenderID = adapter.RoleMember, "42" + b.processFrom(ctx, msg, ad) + + rows, err := store.Count(ctx, events.Filter{}) + if err != nil { + t.Fatalf("Count: %v", err) + } + if rows != 0 || store.PendingKeys() != 0 { + t.Fatalf("普通聊天消息进了留存面:rows=%d pending=%d", rows, store.PendingKeys()) + } + if stats := b.EventFallback(); stats.Frames != 0 { + t.Fatalf("普通聊天消息进了降级计数:%+v", stats) + } +} diff --git a/internal/services/channelbridge/inbound.go b/internal/services/channelbridge/inbound.go index e70d588..222c700 100644 --- a/internal/services/channelbridge/inbound.go +++ b/internal/services/channelbridge/inbound.go @@ -144,6 +144,14 @@ func (b *Bridge) processFrom(ctx context.Context, msg adapter.UnifiedMessage, a // 事件只登记、绝不驱动 loop;被放行的事件已就地换成**剥离后的安全文本**, // 继续走常规流水线(限定注入 / 升级真会话)。 notice := b.preEntryReview(ctx, &msg, a) + // design/26 §4/§5(切片 E3):**渠道事件留存**的唯一插入点。 + // + // 位置说明(与 §5.1 示意图的一行之差):preEntryReview **就是**那一步归一化 + // (它调用适配器的 SystemNoticeNormalizer 并净化 SafeText),把 recordEvent 放在 + // 它之后一行可以**只归一化一次**,并且 §9.2 第 2 条要求把 NoticeClass 的裁决结果 + // 记进日志 —— 那只有这一行之后才有结论。关键性质不变:留存**无条件**,在任何 + // 早退分支(仅登记 / 静默 / 升级)之前,且**绝不**决定 loop 走不走。 + b.recordEvent(ctx, msg, a, notice) if notice.Applicable && !notice.Proceed { b.registerNoticeOnly(ctx, msg, a, notice) return