AskaEth
b1f9b11d6e
- #1 TestSpawnTimeoutFromToolDefault：墙钟竞态（Spawn→Wait 的调度抖动 vs 120ms 工具时限）⇒ 用「托底子代理占住唯一槽位 + gate 放行」把 running 变成由门控保证；并新增 DeadlineExceeded 断言。 - #2 TestParallelLimit：真实成因**不是**「同进程迭代状态泄漏」（任务书推测错了）—— 是 transitionRunning 与 runner 内 active++ 之间的**抢占窗口**（复现：-count=200 六次红、三进程并行 1/9）⇒ 改等 runner 内部 activeCount()>=2 这一**稳定态**，并新增「整轮排空峰值==2」。 - #3 TestBackthink_List_NewestFirst：真实成因**不是**时间戳并列（SQL 是 ORDER BY id DESC）—— 是**后台心跳 writer 在两次 List 之间插入新行**（决定性证据：心跳周期缩到 1ms ⇒ 21/200 红，且第二次读到的 id 更大）⇒ 不再 Start() 心跳，改同步 think 写 3 条。 - #4 TestClient_OnConnectErrorTriggersRedial：固定 60ms 睡眠 vs 每周期耗时不定的墙钟竞态（改动前基线 5/9 红）⇒ 改为等第 2 次拨号事件；顺带改正与内容不符的旧注释。 - 证据：逐条 -count=20 绿；整树连跑 3 次 59 ok/0 FAIL；**三进程并行整树 9/9 绿（改动前基线 5/9 红）**；GOMAXPROCS=1 ×4 进程加压；-count=200/500/500/300；-race 三包绿。 - 主会话独立复验：隔离 worktree build/vet 绿、四条 -count=20 全绿、两进程并行整树双 59 ok/0 FAIL；**变异独立复现**（TestParallelLimit 的 MaxParallel 2→3 ⇒ max concurrent runs = 3, want 2 红）。 - 顺带发现第 5 条既有 flake：internal/runtime/exec/TestTerminal_Timeout_Killed（改动前基线 2/9 红）；以及 audit/00-总览.md:321/372 对 #2/#3 成因的记载有误（DevDocs 由主会话更正）。
443 lines
14 KiB
Go
443 lines
14 KiB
Go
package backthink
|
||
|
||
import (
|
||
"context"
|
||
"errors"
|
||
"strings"
|
||
"testing"
|
||
"time"
|
||
|
||
"git.yeij.top/AskaEth/Cyrene_v2/internal/bus"
|
||
"git.yeij.top/AskaEth/Cyrene_v2/internal/iot"
|
||
)
|
||
|
||
// newBackthink 用一份新的内存 DB 与给定 deps 构建 Backthink，
|
||
// 并返回填好 DB 句柄的 deps，好让测试能检查 deps。
|
||
func newBackthink(t *testing.T, cfg Config, deps Deps) (*Backthink, Deps) {
|
||
t.Helper()
|
||
deps.DB = newTestDB(t)
|
||
b, err := New(cfg, deps)
|
||
if err != nil {
|
||
t.Fatalf("New: %v", err)
|
||
}
|
||
t.Cleanup(func() { _ = b.Stop(context.Background()) })
|
||
return b, deps
|
||
}
|
||
|
||
func TestNew_Validation(t *testing.T) {
|
||
t.Parallel()
|
||
if _, err := New(Config{}, Deps{DB: newTestDB(t)}); !errors.Is(err, ErrNoThinker) {
|
||
t.Fatalf("missing thinker: err = %v, want ErrNoThinker", err)
|
||
}
|
||
if _, err := New(Config{}, Deps{Thinker: &mockThinker{}}); !errors.Is(err, ErrNilDB) {
|
||
t.Fatalf("missing db: err = %v, want ErrNilDB", err)
|
||
}
|
||
}
|
||
|
||
// TestConfig_Defaults 验证收紧后的主动推送默认值：15 分钟
|
||
// 心跳、每 60 分钟最多 1 次推送、推送后 30 分钟冷却、
|
||
// 最近对话 10 分钟冷却，以及按风格 2 小时的窗口。
|
||
func TestConfig_Defaults(t *testing.T) {
|
||
t.Parallel()
|
||
cfg := normalize(Config{})
|
||
if cfg.IntervalSec != 900 {
|
||
t.Fatalf("IntervalSec = %d, want 900", cfg.IntervalSec)
|
||
}
|
||
if cfg.Interval != 900*time.Second {
|
||
t.Fatalf("Interval = %v, want 900s", cfg.Interval)
|
||
}
|
||
if cfg.PushLimit != 1 {
|
||
t.Fatalf("PushLimit = %d, want 1", cfg.PushLimit)
|
||
}
|
||
if cfg.PushWindow != 60*time.Minute {
|
||
t.Fatalf("PushWindow = %v, want 60m", cfg.PushWindow)
|
||
}
|
||
if cfg.PushCooldown != 30*time.Minute {
|
||
t.Fatalf("PushCooldown = %v, want 30m", cfg.PushCooldown)
|
||
}
|
||
if cfg.RecentChatCooldown != 10*time.Minute {
|
||
t.Fatalf("RecentChatCooldown = %v, want 10m", cfg.RecentChatCooldown)
|
||
}
|
||
if cfg.KindWindow != 2*time.Hour {
|
||
t.Fatalf("KindWindow = %v, want 2h", cfg.KindWindow)
|
||
}
|
||
}
|
||
|
||
func TestBackthink_Heartbeat_TriggersThink(t *testing.T) {
|
||
t.Parallel()
|
||
thinker := &mockThinker{text: "nothing to report"}
|
||
b, _ := newBackthink(t, Config{Interval: 20 * time.Millisecond}, Deps{Thinker: thinker})
|
||
|
||
b.Start(context.Background())
|
||
defer func() { _ = b.Stop(context.Background()) }()
|
||
|
||
// 启动轮（07.4.4）会先落一条 startup 记录，这里等真正的周期心跳记录。
|
||
eventually(t, 2*time.Second, func() bool {
|
||
recs, _ := b.List(context.Background(), 10)
|
||
for _, r := range recs {
|
||
if r.Trigger == "heartbeat" {
|
||
return true
|
||
}
|
||
}
|
||
return false
|
||
})
|
||
if got := thinker.count(); got < 1 {
|
||
t.Fatalf("thinker calls = %d, want at least 1", got)
|
||
}
|
||
found := false
|
||
recs, err := b.List(context.Background(), 10)
|
||
if err != nil {
|
||
t.Fatalf("List: %v", err)
|
||
}
|
||
for _, r := range recs {
|
||
if r.Trigger == "heartbeat" {
|
||
found = true
|
||
}
|
||
}
|
||
if !found {
|
||
t.Fatalf("no heartbeat record in %d records", len(recs))
|
||
}
|
||
}
|
||
|
||
func TestBackthink_EventTrigger_Push(t *testing.T) {
|
||
t.Parallel()
|
||
thinker := &mockThinker{text: "washing finished, clothes ready"}
|
||
pusher := &mockPusher{result: PushResult{Pushed: true, Route: RouteActiveClient}}
|
||
bs := bus.New()
|
||
b, _ := newBackthink(t, Config{}, Deps{
|
||
Bus: bs,
|
||
Thinker: thinker,
|
||
State: &mockState{state: UserState{}},
|
||
Pusher: pusher,
|
||
Now: func() time.Time { return dayAt() },
|
||
})
|
||
|
||
b.Start(context.Background())
|
||
defer func() { _ = b.Stop(context.Background()) }()
|
||
|
||
if err := bs.Publish(context.Background(), SubjectReminderTrigger, "reminder due"); err != nil {
|
||
t.Fatalf("Publish: %v", err)
|
||
}
|
||
|
||
// 启动轮（07.4.4）同样会思考并推送，因此按触发源过滤出事件记录再断言。
|
||
var rec Record
|
||
eventually(t, 2*time.Second, func() bool {
|
||
recs, _ := b.List(context.Background(), 10)
|
||
for _, r := range recs {
|
||
if r.Trigger == SubjectReminderTrigger {
|
||
rec = r
|
||
return true
|
||
}
|
||
}
|
||
return false
|
||
})
|
||
if got := thinker.count(); got < 1 {
|
||
t.Fatalf("thinker calls = %d, want at least 1", got)
|
||
}
|
||
if got := pusher.count(); got < 1 {
|
||
t.Fatalf("pusher calls = %d, want at least 1 (important + daytime)", got)
|
||
}
|
||
if !rec.ShouldPush {
|
||
t.Fatalf("ShouldPush = false, want true")
|
||
}
|
||
if !rec.PushResult.Pushed || rec.PushResult.Route != RouteActiveClient {
|
||
t.Fatalf("PushResult = %+v, want pushed active_client", rec.PushResult)
|
||
}
|
||
}
|
||
|
||
func TestBackthink_EventTrigger_NoPush(t *testing.T) {
|
||
t.Parallel()
|
||
thinker := &mockThinker{text: "washing finished, clothes ready"}
|
||
pusher := &mockPusher{result: PushResult{Pushed: true, Route: RouteActiveClient}}
|
||
bs := bus.New()
|
||
b, _ := newBackthink(t, Config{}, Deps{
|
||
Bus: bs,
|
||
Thinker: thinker,
|
||
State: &mockState{state: UserState{Sleeping: true}},
|
||
Pusher: pusher,
|
||
Now: func() time.Time { return nightAt() },
|
||
})
|
||
|
||
b.Start(context.Background())
|
||
defer func() { _ = b.Stop(context.Background()) }()
|
||
|
||
if err := bs.Publish(context.Background(), iot.DeviceUpdateSubject, iot.DeviceUpdate{
|
||
Device: iot.Device{ID: "washer", Type: "sensor"},
|
||
Action: "finished",
|
||
}); err != nil {
|
||
t.Fatalf("Publish: %v", err)
|
||
}
|
||
|
||
var rec Record
|
||
eventually(t, 2*time.Second, func() bool {
|
||
recs, _ := b.List(context.Background(), 10)
|
||
for _, r := range recs {
|
||
if r.Trigger == "iot.device_update" {
|
||
rec = r
|
||
return true
|
||
}
|
||
}
|
||
return false
|
||
})
|
||
if got := pusher.count(); got != 0 {
|
||
t.Fatalf("pusher calls = %d, want 0 (asleep + night)", got)
|
||
}
|
||
if rec.ShouldPush {
|
||
t.Fatalf("ShouldPush = true, want false")
|
||
}
|
||
if rec.PushResult.Route != RouteNone {
|
||
t.Fatalf("PushResult.Route = %q, want %q", rec.PushResult.Route, RouteNone)
|
||
}
|
||
}
|
||
|
||
func TestBackthink_SystemEventLowValueDowngraded(t *testing.T) {
|
||
t.Parallel()
|
||
// design/12 §6 策略一：公开平台上的低价值系统事件不会
|
||
// 唤醒一整轮思考——它被降级为仅登记一条记录。
|
||
thinker := &mockThinker{text: "should not be called"}
|
||
bs := bus.New()
|
||
b, _ := newBackthink(t, Config{}, Deps{
|
||
Bus: bs,
|
||
Thinker: thinker,
|
||
Now: func() time.Time { return dayAt() },
|
||
})
|
||
|
||
b.Start(context.Background())
|
||
defer func() { _ = b.Stop(context.Background()) }()
|
||
|
||
// 先等启动轮（07.4.4）落库并记下基线，再发布低价值系统事件。
|
||
eventually(t, 2*time.Second, func() bool {
|
||
recs, _ := b.List(context.Background(), 10)
|
||
return len(recs) >= 1 && recs[0].Trigger == "startup"
|
||
})
|
||
before := thinker.count()
|
||
|
||
if err := bs.Publish(context.Background(), SubjectSystemEvent, SystemEvent{Summary: "member joined"}); err != nil {
|
||
t.Fatalf("Publish: %v", err)
|
||
}
|
||
|
||
var rec Record
|
||
eventually(t, 2*time.Second, func() bool {
|
||
recs, _ := b.List(context.Background(), 10)
|
||
for _, r := range recs {
|
||
if r.Trigger == SubjectSystemEvent {
|
||
rec = r
|
||
return true
|
||
}
|
||
}
|
||
return false
|
||
})
|
||
if got := thinker.count(); got != before {
|
||
t.Fatalf("thinker calls = %d, want %d (low-value system event downgraded)", got, before)
|
||
}
|
||
if rec.ShouldPush {
|
||
t.Fatalf("ShouldPush = true for a downgraded system event")
|
||
}
|
||
if rec.PushResult.Route != RouteNone {
|
||
t.Fatalf("PushResult.Route = %q, want %q", rec.PushResult.Route, RouteNone)
|
||
}
|
||
}
|
||
|
||
func TestBackthink_SystemEventHighValueThinks(t *testing.T) {
|
||
t.Parallel()
|
||
// 高重要度（与主人相关）的系统事件仍会唤醒一轮思考——
|
||
// 降噪门只降级低价值噪声（design/12 §6 策略一）。
|
||
thinker := &mockThinker{text: "important notice"}
|
||
bs := bus.New()
|
||
b, _ := newBackthink(t, Config{}, Deps{
|
||
Bus: bs,
|
||
Thinker: thinker,
|
||
Now: func() time.Time { return dayAt() },
|
||
})
|
||
|
||
b.Start(context.Background())
|
||
defer func() { _ = b.Stop(context.Background()) }()
|
||
|
||
eventually(t, 2*time.Second, func() bool {
|
||
recs, _ := b.List(context.Background(), 10)
|
||
return len(recs) >= 1 && recs[0].Trigger == "startup"
|
||
})
|
||
before := thinker.count()
|
||
|
||
if err := bs.Publish(context.Background(), SubjectSystemEvent, SystemEvent{Summary: "owner tagged", Import: ImportanceHigh}); err != nil {
|
||
t.Fatalf("Publish: %v", err)
|
||
}
|
||
|
||
eventually(t, 2*time.Second, func() bool { return thinker.count() >= before+1 })
|
||
}
|
||
|
||
// TestBackthink_List_NewestFirst 断言 List 严格由新到旧，且单行列表的头部与完整
|
||
// 列表的头部是同一条记录。记录全部由测试同步写入，因此两次读取之间不存在并发
|
||
// 写入者（见下方注释）。
|
||
func TestBackthink_List_NewestFirst(t *testing.T) {
|
||
t.Parallel()
|
||
thinker := &mockThinker{text: "nothing to report"}
|
||
b, _ := newBackthink(t, Config{}, Deps{Thinker: thinker})
|
||
|
||
// 三条可区分的记录，写入顺序即 sources 的顺序。
|
||
sources := []string{"first", "second", "third"}
|
||
// 记录由测试同步写入（直接调用 think），刻意不启动心跳：后台 ticker 会在
|
||
// 两次 List 之间插入一条新记录，使「单行列表的头部 == 完整列表的头部」这个
|
||
// 断言变成跟调度赛跑。此前偶红时读到的正是「刚插入的更新的 id」。
|
||
for _, src := range sources {
|
||
b.think(context.Background(), Trigger{Source: src, Summary: src, Import: ImportanceLow, Kind: TriggerHeartbeat})
|
||
}
|
||
|
||
top, err := b.List(context.Background(), 1)
|
||
if err != nil {
|
||
t.Fatalf("List: %v", err)
|
||
}
|
||
if len(top) != 1 {
|
||
t.Fatalf("len = %d, want 1", len(top))
|
||
}
|
||
all, err := b.List(context.Background(), 10)
|
||
if err != nil {
|
||
t.Fatalf("List: %v", err)
|
||
}
|
||
// 单行列表里的那一条必须与完整列表的头部是同一条最新
|
||
// 记录，且完整列表必须是严格的由新到旧。
|
||
if top[0].ID != all[0].ID {
|
||
t.Fatalf("List(1)[0].ID = %d, want %d (== newest)", top[0].ID, all[0].ID)
|
||
}
|
||
if len(all) != len(sources) {
|
||
t.Fatalf("len(all) = %d, want %d (no writer ran between the two reads)", len(all), len(sources))
|
||
}
|
||
for i := 1; i < len(all); i++ {
|
||
if all[i].ID >= all[i-1].ID {
|
||
t.Fatalf("records not newest-first: %d before %d", all[i-1].ID, all[i].ID)
|
||
}
|
||
}
|
||
// 最新在前不仅体现在 id 上，也体现在内容上：列表是写入顺序的逆序。
|
||
for i := range sources {
|
||
want := sources[len(sources)-1-i]
|
||
if all[i].Trigger != want {
|
||
t.Fatalf("all[%d].Trigger = %q, want %q (newest first)", i, all[i].Trigger, want)
|
||
}
|
||
}
|
||
}
|
||
|
||
func TestBackthink_Stop_NoLeak(t *testing.T) {
|
||
t.Parallel()
|
||
thinker := &mockThinker{text: "nothing to report"}
|
||
b, _ := newBackthink(t, Config{Interval: 10 * time.Millisecond}, Deps{Thinker: thinker})
|
||
|
||
b.Start(context.Background())
|
||
eventually(t, 2*time.Second, func() bool { return thinker.count() >= 1 })
|
||
|
||
if err := b.Stop(context.Background()); err != nil {
|
||
t.Fatalf("Stop: %v", err)
|
||
}
|
||
// 幂等。
|
||
if err := b.Stop(context.Background()); err != nil {
|
||
t.Fatalf("second Stop: %v", err)
|
||
}
|
||
|
||
before := thinker.count()
|
||
time.Sleep(40 * time.Millisecond)
|
||
if got := thinker.count(); got != before {
|
||
t.Fatalf("thinker calls grew after stop: %d -> %d", before, got)
|
||
}
|
||
}
|
||
|
||
func TestBackthink_Stop_UnsubscribesBus(t *testing.T) {
|
||
t.Parallel()
|
||
thinker := &mockThinker{text: "nothing to report"}
|
||
bs := bus.New()
|
||
b, _ := newBackthink(t, Config{Interval: 25 * time.Millisecond}, Deps{Bus: bs, Thinker: thinker})
|
||
|
||
b.Start(context.Background())
|
||
// 通过触发一次思考来确认订阅是活的（以它持久化的
|
||
// 记录为准，这样 07.4.4 的启动轮与心跳 tick 各自的
|
||
// 触发字符串不会掩盖或覆盖它）。
|
||
if err := bs.Publish(context.Background(), SubjectReminderTrigger, "first"); err != nil {
|
||
t.Fatalf("Publish: %v", err)
|
||
}
|
||
eventually(t, 2*time.Second, func() bool {
|
||
recs, _ := b.List(context.Background(), 10)
|
||
for _, r := range recs {
|
||
if r.Trigger == SubjectReminderTrigger {
|
||
return true
|
||
}
|
||
}
|
||
return false
|
||
})
|
||
|
||
if err := b.Stop(context.Background()); err != nil {
|
||
t.Fatalf("Stop: %v", err)
|
||
}
|
||
|
||
before := thinker.count()
|
||
if err := bs.Publish(context.Background(), iot.DeviceUpdateSubject, iot.DeviceUpdate{
|
||
Device: iot.Device{ID: "washer", Type: "sensor"}, Action: "finished",
|
||
}); err != nil {
|
||
t.Fatalf("Publish after stop: %v", err)
|
||
}
|
||
time.Sleep(40 * time.Millisecond)
|
||
if got := thinker.count(); got != before {
|
||
t.Fatalf("thinker calls grew after stop (bus not unsubscribed): %d -> %d", before, got)
|
||
}
|
||
}
|
||
|
||
// TestBackthink_EventPush_DedupSameEvent 验证同一事件去重门：
|
||
// 重复的相同事件（同源 + 同摘要）在首次推送后
|
||
// 被抑制（防刷），而真正不同的事件仍会发出。重复的那条
|
||
// 被记为已抑制（route none），而不是投递出去。
|
||
func TestBackthink_EventPush_DedupSameEvent(t *testing.T) {
|
||
t.Parallel()
|
||
thinker := &mockThinker{text: "washing finished, clothes ready"}
|
||
pusher := &mockPusher{result: PushResult{Pushed: true, Route: RouteActiveClient}}
|
||
b, _ := newBackthink(t, Config{}, Deps{
|
||
Thinker: thinker,
|
||
State: &mockState{state: UserState{}},
|
||
Pusher: pusher,
|
||
Now: func() time.Time { return dayAt() },
|
||
})
|
||
|
||
ev := triggerForEvent(bus.Event{Subject: iot.DeviceUpdateSubject, Payload: iot.DeviceUpdate{
|
||
Device: iot.Device{ID: "washer", Type: "sensor"}, Action: "finished",
|
||
}})
|
||
|
||
b.think(context.Background(), ev) // 第一次：推送
|
||
b.think(context.Background(), ev) // 第二次：同一事件 -> 被抑制
|
||
b.think(context.Background(), ev) // 第三次：同一事件 -> 被抑制
|
||
|
||
if got := pusher.count(); got != 1 {
|
||
t.Fatalf("pusher calls = %d, want 1 (repeating same event deduped)", got)
|
||
}
|
||
recs, err := b.List(context.Background(), 10)
|
||
if err != nil {
|
||
t.Fatalf("List: %v", err)
|
||
}
|
||
if len(recs) < 3 {
|
||
t.Fatalf("len(recs) = %d, want at least 3 (one per think)", len(recs))
|
||
}
|
||
// 最新的在前：最近的这两条就是被抑制的重复。
|
||
for i := 0; i < 2; i++ {
|
||
if recs[i].PushResult.Pushed || recs[i].PushResult.Route != RouteNone {
|
||
t.Fatalf("recs[%d] = %+v, want suppressed duplicate", i, recs[i].PushResult)
|
||
}
|
||
if !strings.Contains(recs[i].PushResult.Detail, "duplicate") {
|
||
t.Fatalf("recs[%d].Detail = %q, want it to mention duplicate", i, recs[i].PushResult.Detail)
|
||
}
|
||
}
|
||
// 第三新的那条是首次（已投递的）推送。
|
||
if !recs[2].PushResult.Pushed || recs[2].PushResult.Route != RouteActiveClient {
|
||
t.Fatalf("recs[2] = %+v, want the first delivered push", recs[2].PushResult)
|
||
}
|
||
|
||
// 真正不同的事件会重置该链并发出。
|
||
ev2 := triggerForEvent(bus.Event{Subject: iot.DeviceUpdateSubject, Payload: iot.DeviceUpdate{
|
||
Device: iot.Device{ID: "dryer", Type: "sensor"}, Action: "finished",
|
||
}})
|
||
b.think(context.Background(), ev2)
|
||
if got := pusher.count(); got != 2 {
|
||
t.Fatalf("pusher calls after a different event = %d, want 2", got)
|
||
}
|
||
// 在出现不同事件之后，原先那个相同事件不再是
|
||
// 连续重复，因此它会再次推送——该门只去重连续重复。
|
||
b.think(context.Background(), ev)
|
||
if got := pusher.count(); got != 3 {
|
||
t.Fatalf("pusher calls after an intervening different event = %d, want 3", got)
|
||
}
|
||
}
|