feat(backthink): 自排 C1-2 本体 + TriggerUnknown 修 fail-open（V1）
- selfwake.go：WakeNow + 待跑槽容量 1（先探槽合并不过闸、槽空才过闸、同锁）+ 闸 4 轮/小时
（$DATA_DIR/config/backthink.json，零新 env 键）+ 审计四条 what + 启动日志两键 + 组合根接线
- G-A（主人已批）：TriggerUnknown 作零值、既有两值编号后移（语义不动），未知触发走**最紧分支**
- 5 个变异自检（拆闸/≤0 未回落/过闸挪到探槽前=G1 失效形态/槽无界/零值挪走）
⚠️ 服务侧已就绪但**她仍调不到**（自排工具面属 C1-3）⇒ 不得写「C 已完成」
This commit is contained in:
AskaEth committed
1 parent
301f8b4276
commit
ed5005307a
12 files changed
+1193
-8
No files matched your search
@@ -159,6 +159,11 @@ func (a *app) wireFrontend() error {
|
||||
if a.channelBridge != nil {
|
||||
a.backthink.SetIMSink(a.channelBridge)
|
||||
}
|
||||
// design/12 §N C.5：自排的四个动作（queued/coalesced/capped/denied）走与渠道
|
||||
// 黑名单拒绝同一条审计链（谁/什么/何时/结果），因此它们**面板可查**，不需要
|
||||
// 新端点（GET /api/v1/admin/observability/audit 已存在）。同样是晚绑定：
|
||||
// backthink.New 时网关还没组装。
|
||||
a.backthink.SetAuditSink(gw)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
@@ -10,6 +10,7 @@ import (
|
||||
"git.yeij.top/AskaEth/Cyrene_v2/internal/capability/registry"
|
||||
"git.yeij.top/AskaEth/Cyrene_v2/internal/channel/session"
|
||||
"git.yeij.top/AskaEth/Cyrene_v2/internal/config"
|
||||
"git.yeij.top/AskaEth/Cyrene_v2/internal/config/runtimeconfig"
|
||||
"git.yeij.top/AskaEth/Cyrene_v2/internal/iot"
|
||||
"git.yeij.top/AskaEth/Cyrene_v2/internal/logger"
|
||||
"git.yeij.top/AskaEth/Cyrene_v2/internal/platform"
|
||||
@@ -66,7 +67,7 @@ func (a *app) wireAuxiliary() error {
|
||||
}
|
||||
a.automation = automation
|
||||
|
||||
bt, err := backthink.New(backthink.ConfigFromBackground(a.cfg.Background), backthink.Deps{
|
||||
bt, err := backthink.New(a.backthinkTuning(), backthink.Deps{
|
||||
Bus: a.bus,
|
||||
Thinker: backthink.NewLoopThinker(a.loop, a.history),
|
||||
Recent: a.history,
|
||||
@@ -128,6 +129,29 @@ func (a *app) buildSkillFace() *loop.SkillFace {
|
||||
})
|
||||
}
|
||||
|
||||
// backthinkTuning 组装后台思考的调优配置：env 引导面（心跳周期 / 安静时段 / 推送频次）
|
||||
// 由 ConfigFromBackground 映射，**自排成本闸**（design/12 §N C.5）从
|
||||
// `$DATA_DIR/config/backthink.json` 读 —— 零新 env 键（主人 2026-10-06 `.env` 冻结红线）。
|
||||
//
|
||||
// 文件缺失/全空白 ⇒ 保持零值，由 backthink 的 normalize 回落代码默认（4 轮/小时；
|
||||
// 「≤0 回落默认」在服务侧统一执行，这里不再造第二套默认值）。存在却解不开 ⇒ 记一条
|
||||
// 带 source 的 WARN 并回落默认：坏配置**不许**静默变成「无上限」。
|
||||
//
|
||||
// 与 QUIET_* 同口径：启动期读一次，**改文件需重启**。
|
||||
func (a *app) backthinkTuning() backthink.Config {
|
||||
cfg := backthink.ConfigFromBackground(a.cfg.Background)
|
||||
tuningPath := runtimeconfig.DefaultBackthinkPath(resolveDataDir(a.cfg))
|
||||
tuning, err := runtimeconfig.LoadBackgroundTuning(tuningPath)
|
||||
if err != nil {
|
||||
logger.WithSource(a.log, "services/backthink").Warn(
|
||||
"backthink tuning unreadable; falling back to code defaults", "path", tuningPath, "error", err)
|
||||
return cfg
|
||||
}
|
||||
cfg.SelfWakeLimit = tuning.SelfWakePerWindow
|
||||
cfg.SelfWakeWindow = time.Duration(tuning.SelfWakeWindowSec) * time.Second
|
||||
return cfg
|
||||
}
|
||||
|
||||
// pushOutletConfig 把环境变量里的推送出口配置块映射为 backthink 的导出器配置。
|
||||
// 个推（Getui）凭据只被 config 读取、并不消费：备用 SDK 尚未接入，因此出口在
|
||||
// 待接入该档位时记日志，而不是伪造一次推送。token 只透传，绝不写日志。
|
||||
|
||||
@@ -0,0 +1,83 @@
|
||||
// backthink.go 是后台思考（backthink）的**可变源**：`$DATA_DIR/config/backthink.json`
|
||||
// （design/12 §N C.5 + design/13 §2.1「运行期生成的配置进 `$DATA_DIR/config/`，
|
||||
// 新增配置家族从一开始就只写这里，不落老位置」）。
|
||||
//
|
||||
// 边界划清（与 websearch.go 同源）：
|
||||
//
|
||||
// 1. 本文件只管**文件形状 + 读取**：它不认识 `backthink.Config`，也不认识
|
||||
// 「≤0 回落默认」这条裁决 —— 那是服务侧 normalize 的事。这里只如实把盘上的
|
||||
// 数字交出去（零值 = 没配，由调用方用代码默认值兜）。
|
||||
// 2. **只读**：写路径（面板保存 + admin 端点）不在 C1 内，属 C2。因此本文件
|
||||
// 刻意没有 `Save*`：多一个没人调用的写口，就多一条没人验证的落盘语义。
|
||||
// 3. **零新 env 键**（主人 2026-10-06 `.env` 冻结红线）：自排的成本旋钮只从这里来，
|
||||
// 不进 `.env` / `.env.example`，也不进面板的事实卡（那张表的键名被模板钉住）。
|
||||
//
|
||||
// 读取口径：文件缺失或全空白 = 还没配过（零值 + nil，**不是错误**）；
|
||||
// 存在但解不开 = 返回错误，调用方记 WARN 并回落**代码默认值** —— 绝不静默变成
|
||||
// 「无上限」（design/12 §N C.5 的 G3 同一条纪律：坏配置不许把成本保护关掉）。
|
||||
package runtimeconfig
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"strings"
|
||||
)
|
||||
|
||||
// backthinkFileName 是文件名（家族命名 = 服务/模块名 + .json）。
|
||||
const backthinkFileName = "backthink.json"
|
||||
|
||||
// backthinkConfigSubdir 是落点的第一段目录（`$DATA_DIR/config/backthink.json`）。
|
||||
// 与 web-search 插件各自声明一份而不是共用一个常量：两处的目录段虽然今天相同，
|
||||
// 但一旦某个家族换位置，共用常量会让另一处**静默**跟着搬家。
|
||||
const backthinkConfigSubdir = "config"
|
||||
|
||||
// ErrInvalidBackthink 是 backthink.json 结构不合法时的分类错误（errors.Is 可判类）。
|
||||
var ErrInvalidBackthink = errors.New("runtimeconfig: invalid backthink config")
|
||||
|
||||
// BackgroundTuning 是 `backthink.json` 的形状：她自己排起一轮思考的成本旋钮。
|
||||
//
|
||||
// 字段名与 design/12 §N C.5 的规格逐字一致（camelCase 的 JSON 键，文件是给人看、
|
||||
// 给人改的）。**没有版本字段**：这个家族只有这两个数字，等它真的长形状时再加版本，
|
||||
// 现在加一个永远为 0 的字段只是噪音。
|
||||
type BackgroundTuning struct {
|
||||
// SelfWakePerWindow 是每窗口允许的自排轮数。零/负 = 未声明或笔误 ⇒
|
||||
// 由服务侧 normalize 回落代码默认（**没有「关闭上限」语义**）。
|
||||
SelfWakePerWindow int `json:"selfWakePerWindow"`
|
||||
// SelfWakeWindowSec 是执行上述上限的滑动窗口秒数。零/负 ⇒ 同上回落默认。
|
||||
SelfWakeWindowSec int `json:"selfWakeWindowSec"`
|
||||
}
|
||||
|
||||
// DefaultBackthinkPath 返回后台思考调优的落点：`$DATA_DIR/config/backthink.json`。
|
||||
func DefaultBackthinkPath(dataDir string) string {
|
||||
return filepath.Join(dataDir, backthinkConfigSubdir, backthinkFileName)
|
||||
}
|
||||
|
||||
// LoadBackgroundTuning 读取 backthink.json（**只读**）。
|
||||
//
|
||||
// 文件缺失或全空白 ⇒ 零值 + nil：调用方用代码默认值（`normalize` 会把 ≤0 回落默认），
|
||||
// 因此「没配过」与「配成 0」得到同一个结果，正是这条裁决要的。
|
||||
//
|
||||
// 存在但解不开 ⇒ 返回错误。调用方应记一条带 source 的 WARN 并回落默认；
|
||||
// **不得覆盖**那个文件（损坏的文件可以人工救回，程序盖掉它就丢了人工写的意图）。
|
||||
//
|
||||
// 它不建目录、不读老位置：新家族没有老位置（design/13 §2.1 第 5 条）。
|
||||
func LoadBackgroundTuning(path string) (BackgroundTuning, error) {
|
||||
raw, err := os.ReadFile(path)
|
||||
if err != nil {
|
||||
if os.IsNotExist(err) {
|
||||
return BackgroundTuning{}, nil
|
||||
}
|
||||
return BackgroundTuning{}, fmt.Errorf("runtimeconfig: read %s: %w", path, err)
|
||||
}
|
||||
if strings.TrimSpace(string(raw)) == "" {
|
||||
return BackgroundTuning{}, nil
|
||||
}
|
||||
var t BackgroundTuning
|
||||
if err := json.Unmarshal(raw, &t); err != nil {
|
||||
return BackgroundTuning{}, fmt.Errorf("%w: decode %s: %w", ErrInvalidBackthink, path, err)
|
||||
}
|
||||
return t, nil
|
||||
}
|
||||
@@ -109,6 +109,10 @@ type Deps struct {
|
||||
// Now 是决策/记录时间戳的墙钟来源。nil 则用
|
||||
// time.Now，从而让依赖时间的行为在测试中可注入。
|
||||
Now func() time.Time
|
||||
// Audit 是自排审计出口（design/12 §N C.5）。它实现 Auditor 这个三字段窄接口，
|
||||
// 生产接线注入网关（后端组装晚于 backthink，故也提供 SetAuditSink 晚绑定）。
|
||||
// nil 则审计静默跳过：审计是留痕，不是门。
|
||||
Audit Auditor
|
||||
// Log 是基础日志器，来源标记为 "services/backthink"。
|
||||
Log *slog.Logger
|
||||
}
|
||||
@@ -135,6 +139,10 @@ type Backthink struct {
|
||||
// 生命周期锁分开，因此去重决策绝不会阻塞槽位换入。
|
||||
eventMu sync.Mutex
|
||||
lastEventKey string
|
||||
|
||||
// selfWake 是她自排一轮思考的成本闸与待跑槽（selfwake.go，design/12 §N C）。
|
||||
// 它在 New 里按**已 normalize** 的配置构建（G3），Start 时由一个 worker 消费。
|
||||
selfWake *selfWakeGate
|
||||
}
|
||||
|
||||
// New 由 cfg 与 deps 构建 Backthink，套用默认值并校验
|
||||
@@ -185,6 +193,9 @@ func New(cfg Config, deps Deps) (*Backthink, error) {
|
||||
gate: gate,
|
||||
now: now,
|
||||
sessionID: sessionID,
|
||||
// 闸只认**已 normalize** 的旋钮：未回落默认的 ≤0 会被滑窗读成「无规则」
|
||||
// （恒放行且不记录），那就把「≤0 回落默认」静默变成「无上限」（G3）。
|
||||
selfWake: newSelfWakeGate(cfg.SelfWakeLimit, cfg.SelfWakeWindow),
|
||||
}, nil
|
||||
}
|
||||
|
||||
@@ -206,6 +217,10 @@ func (b *Backthink) Start(ctx context.Context) {
|
||||
b.cancel = cancel
|
||||
b.wg.Add(1)
|
||||
go b.tickLoop(cctx)
|
||||
// 自排 worker（design/12 §N C.1）：与心跳并列的第二个常驻 goroutine。
|
||||
// Stop 的 cancel()+wg.Wait() 自动收走它，不引入新的生命周期状态。
|
||||
b.wg.Add(1)
|
||||
go b.selfWakeLoop(cctx)
|
||||
b.mu.Unlock()
|
||||
|
||||
if b.deps.Bus != nil {
|
||||
@@ -218,6 +233,10 @@ func (b *Backthink) Start(ctx context.Context) {
|
||||
"quiet_interval_enabled", b.quietIntervalEnabled(),
|
||||
"quiet_hours", quietHoursLabel(b.cfg.QuietStartHour, b.cfg.QuietEndHour),
|
||||
"event_triggers", b.deps.Bus != nil,
|
||||
// 自排的成本闸口径（design/12 §N C.5 的观测要求）：与 quiet_interval 同族，
|
||||
// 缺了它主人无法从日志确认「她一小时最多能叫醒自己几次」。
|
||||
"self_wake_limit", b.cfg.SelfWakeLimit,
|
||||
"self_wake_window", b.cfg.SelfWakeWindow.String(),
|
||||
"session_id", b.sessionID)
|
||||
}
|
||||
|
||||
|
||||
@@ -37,6 +37,11 @@ const (
|
||||
// defaultThinkTimeout 约束单次无人值守思考，因此它绝不会
|
||||
// 卡住 worker goroutine。
|
||||
defaultThinkTimeout = 30 * time.Second
|
||||
// defaultSelfWakeLimit / defaultSelfWakeWindow 是她**自排**一轮思考的上限：
|
||||
// 默认每窗口 4 轮、窗口 1 小时（design/12 §N C.5）。它是**成本闸**，
|
||||
// 与推送频次是两件事 —— PushGate 管「能不能打扰主人」，管不了 LLM 花费。
|
||||
defaultSelfWakeLimit = 4
|
||||
defaultSelfWakeWindow = time.Hour
|
||||
// defaultSessionID 在调用方未提供时限定一次思考的作用域。它是
|
||||
// loop.Run 的持久化键，刻意与 history.MainSessionID 不同，
|
||||
// 这样自主思考绝不会与活跃主会话的检查点
|
||||
@@ -89,12 +94,24 @@ type Config struct {
|
||||
KindWindow time.Duration
|
||||
// ThinkTimeout 约束单轮思考。为零用默认值。
|
||||
ThinkTimeout time.Duration
|
||||
// SelfWakeLimit 限制 SelfWakeWindow 内她自己排起的思考轮数
|
||||
// （design/12 §N C.5）。**≤0 ⇒ 回落默认**：没有「关闭上限」语义——
|
||||
// 一次笔误不许把成本保护关掉（与 channelbridge 的 directLimit 同一裁决口径）。
|
||||
// 它的取值来自 `$DATA_DIR/config/backthink.json`（零新 env 键，`.env` 冻结），
|
||||
// 缺失/为零时由 normalize 填默认。
|
||||
SelfWakeLimit int
|
||||
// SelfWakeWindow 是执行 SelfWakeLimit 的滑动窗口。**≤0 ⇒ 回落默认**。
|
||||
SelfWakeWindow time.Duration
|
||||
}
|
||||
|
||||
// ConfigFromBackground 把 config.BackgroundConfig（心跳周期、
|
||||
// 共享安静时段窗口与主动推送频次旋钮的来源）
|
||||
// 映射成 backthink.Config，env 未暴露的项保持其
|
||||
// 默认值，交给 normalize 填。
|
||||
//
|
||||
// 它**刻意不带**自排上限两个旋钮（design/12 §N C.5，2026-10-06 `.env` 冻结红线）：
|
||||
// 自排的成本旋钮走 `$DATA_DIR/config/backthink.json`，进 `.env` 就是给冻结面开新键。
|
||||
// 组合根读到配置文件后直接赋在返回的 Config 上。
|
||||
func ConfigFromBackground(bg config.BackgroundConfig) Config {
|
||||
return Config{
|
||||
IntervalSec: bg.IntervalSec,
|
||||
@@ -145,6 +162,17 @@ func normalize(cfg Config) Config {
|
||||
if cfg.ThinkTimeout <= 0 {
|
||||
cfg.ThinkTimeout = defaultThinkTimeout
|
||||
}
|
||||
// 自排闸的旋钮与 PushLimit 同形：**≤0 一律回落默认**。这不是可有可无的填充——
|
||||
// session.RateLimiter 在 limit<=0 || window<=0 时 Allow() 恒 true 且一条事件都不记录
|
||||
// （它自称是「无规则」的安全默认），若把未 normalize 的 ≤0 直达滑窗，
|
||||
// 「回落默认」就**静默**变成「无上限」（无报错、无 WARN）。闸必须在构造前就
|
||||
// 拿到已回落的值（design/12 §N C.5 的 G3；探针 = C.8 判据 14）。
|
||||
if cfg.SelfWakeLimit <= 0 {
|
||||
cfg.SelfWakeLimit = defaultSelfWakeLimit
|
||||
}
|
||||
if cfg.SelfWakeWindow <= 0 {
|
||||
cfg.SelfWakeWindow = defaultSelfWakeWindow
|
||||
}
|
||||
return cfg
|
||||
}
|
||||
|
||||
|
||||
@@ -101,9 +101,10 @@ type DecisionInput struct {
|
||||
// 需要区分「周期性心跳（含启动轮，享长期静默豁免）」与「她自己排起的一轮
|
||||
// （要求她说得出理由）」，其余类别按事件处理。
|
||||
//
|
||||
// ⚠️ TriggerKind 的零值是 TriggerHeartbeat：生产调用方**必须**显式传 tr.Kind
|
||||
// （backthink.go 的 think 是唯一调用点，已显式传）。省略 Kind 等于按心跳豁免，
|
||||
// 那是最宽的一档，别让「漏填」静默变成「免检」。
|
||||
// ⚠️ TriggerKind 的**零值是 TriggerUnknown**（G-A 修法）：它是**最紧**的一档，
|
||||
// 不享任何豁免（activityHeld 让它落在扣下上）。生产调用方仍**必须**显式传
|
||||
// tr.Kind（backthink.go 的 think 是唯一调用点，已显式传）—— 零值安全只是最后一道
|
||||
// 兜底，不是「可以不填」的许可。
|
||||
Kind TriggerKind
|
||||
}
|
||||
|
||||
@@ -178,6 +179,9 @@ func (b *Backthink) baseDecide(in DecisionInput) Decision {
|
||||
// ——要求她说得出理由（Import ≥ High，即声明 ≥ warning），否则扣下；
|
||||
// - 其间与活动状态未知 → 不软化，照硬门结论通过。
|
||||
//
|
||||
// 未知触发（TriggerUnknown，即漏填 Kind 的零值）同样**不享**豁免：它掉到最后那条
|
||||
// 扣下上（G-A 修法要的就是这个方向——未知 ≠ 最宽）。
|
||||
//
|
||||
// 活动是软信号：硬门（baseDecide）独立生效，频率门（PushGate）在推送器内部生效。
|
||||
func (b *Backthink) activityHeld(in DecisionInput) (bool, string) {
|
||||
if !in.HasActivity {
|
||||
@@ -200,7 +204,7 @@ func (b *Backthink) activityHeld(in DecisionInput) (bool, string) {
|
||||
if in.Kind == TriggerSelfWake {
|
||||
return true, "self-initiated wake held: owner away and no declared urgency"
|
||||
}
|
||||
return true, "owner away a long while"
|
||||
return true, "owner away a long while" // 其余（含 TriggerUnknown）一并扣下
|
||||
}
|
||||
return false, ""
|
||||
}
|
||||
|
||||
@@ -286,3 +286,59 @@ func TestDecide_ActivityGate_SelfWakeNeedsDeclaredUrgency(t *testing.T) {
|
||||
t.Fatalf("long idle heartbeat should still push (07.4.3), got %+v", got)
|
||||
}
|
||||
}
|
||||
|
||||
// TestTriggerKind_ZeroValueIsUnknownAndTight 是 G-A 修法（design/12 §C.4 的零值语义，
|
||||
// 主人 2026-10-06 已批）的探针：`TriggerKind` 的零值必须是 **TriggerUnknown**，且它
|
||||
// 在判定侧走**最紧**分支 —— 长期静默段里不豁免（`{Unknown, Low/Normal}` 扣下）、
|
||||
// 推送侧不被当成事件（照旧过 PushGate）。
|
||||
//
|
||||
// 它守的是 fail-open 的方向：改动前零值 = TriggerHeartbeat，「漏填 Kind」静默变成
|
||||
// 「拿最宽的豁免」。变异：把 TriggerUnknown 挪出零值（如让 TriggerHeartbeat 回到 0），
|
||||
// 或把零值分支混进心跳豁免 ⇒ 本用例变红。
|
||||
func TestTriggerKind_ZeroValueIsUnknownAndTight(t *testing.T) {
|
||||
t.Parallel()
|
||||
var zero TriggerKind
|
||||
if zero != TriggerUnknown {
|
||||
t.Fatalf("TriggerKind 的零值 = %v（%d），want TriggerUnknown", zero, int(zero))
|
||||
}
|
||||
if TriggerUnknown != 0 {
|
||||
t.Fatalf("TriggerUnknown = %d, want 0（零值必须是最紧的那一档）", int(TriggerUnknown))
|
||||
}
|
||||
if got := TriggerUnknown.String(); got != "unknown" {
|
||||
t.Fatalf("TriggerUnknown.String() = %q, want %q", got, "unknown")
|
||||
}
|
||||
// 既有三个值的标签一个字都不许变（落库/面板/审计都以它们为真源）。
|
||||
for kind, want := range map[TriggerKind]string{
|
||||
TriggerHeartbeat: "heartbeat",
|
||||
TriggerEvent: "event",
|
||||
TriggerSelfWake: SelfWakeSource,
|
||||
} {
|
||||
if got := kind.String(); got != want {
|
||||
t.Fatalf("TriggerKind(%d).String() = %q, want %q", int(kind), got, want)
|
||||
}
|
||||
}
|
||||
|
||||
b := testBackthink()
|
||||
longIdle := DecisionInput{
|
||||
Trigger: "unknown.source",
|
||||
At: dayAt(),
|
||||
UserState: UserState{Boredom: 0.9},
|
||||
LastActivity: dayAt().Add(-3 * time.Hour),
|
||||
HasActivity: true,
|
||||
ConclusionPush: true,
|
||||
Kind: TriggerUnknown, // 零值：漏填 Kind 的等价形态
|
||||
}
|
||||
for _, imp := range []Importance{ImportanceLow, ImportanceNormal} {
|
||||
in := longIdle
|
||||
in.Import = imp
|
||||
if got := b.Decide(in); got.ShouldPush {
|
||||
t.Fatalf("未知触发（Import=%v）在长期静默段必须扣下，got %+v", imp, got)
|
||||
}
|
||||
}
|
||||
// 高重要度仍按「值得说」的既有规则放行：G-A 收的是**豁免**，不是把未知一律封死。
|
||||
high := longIdle
|
||||
high.Import = ImportanceHigh
|
||||
if got := b.Decide(high); !got.ShouldPush {
|
||||
t.Fatalf("未知触发但重要度高（>= High）应放行，got %+v", got)
|
||||
}
|
||||
}
|
||||
@@ -13,6 +13,9 @@ package backthink
|
||||
// 自排轮与心跳同分支（清链、不去重）：它既不是外部事件、就不该参与事件链，
|
||||
// 因此真实事件的重复链不会被自排轮错误抑制；自排轮的推送仍由 PushGate 兜。
|
||||
//
|
||||
// 未知触发（TriggerUnknown）也走这一支：它**不是**事件，故不去重；它同样不绕
|
||||
// PushGate（push.go 只对 TriggerEvent 放行），因此重复投递的上限仍由频次门兜住。
|
||||
//
|
||||
// 调用方把它跑在把推送交给 Pusher 之前；真正的重复
|
||||
// 会被记为已抑制（RouteNone），而不是投递出去。
|
||||
func (b *Backthink) eventRedundant(tr Trigger) bool {
|
||||
|
||||
@@ -0,0 +1,306 @@
|
||||
package backthink
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"fmt"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"git.yeij.top/AskaEth/Cyrene_v2/internal/capability/restrict"
|
||||
"git.yeij.top/AskaEth/Cyrene_v2/internal/channel/session"
|
||||
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/loop"
|
||||
)
|
||||
|
||||
// 本文件是**自我唤醒（自排一次思考）的本体**（design/12 §N C，切片 C1-2）。
|
||||
//
|
||||
// 形状（C.0）：她自己发起一次唤醒请求 → 先探**自排专用、容量 1** 的待跑槽
|
||||
// （满 ⇒ 合并，**不过闸、不扣额度**）→ **槽空才**过窗口闸 → 单 worker 跑一轮
|
||||
// TriggerSelfWake 思考。
|
||||
//
|
||||
// 三条不许破坏的约束（各有探针，见 C.8）：
|
||||
// 1. **顺序**：探槽 → 过闸 → 入槽，三步在**同一把锁**内（C.8 判据 13）。颠倒成
|
||||
// 「先过闸、再探槽」，被合并的请求也会白扣一次额度，与「合并不消耗预算」
|
||||
// 自相矛盾（session.RateLimiter.Allow() **成功即消耗**）。
|
||||
// 2. **闸前 normalize**（C.8 判据 14）：进闸的永远是已回落默认的值，因为
|
||||
// RateLimiter 在 limit<=0 时 Allow() 恒 true 且不记录 ⇒ 未回落的值会把
|
||||
// 「≤0 回落默认」静默变成「无上限」。
|
||||
// 3. **容量 1 只限自排轮之间**：至多 1 轮自排在跑 + 至多 1 轮自排在排队；
|
||||
// 心跳轮与事件轮照旧可与自排轮并发（C1 **不**引入全局回合互斥，G2）。
|
||||
//
|
||||
// 自排只保证「她能被叫醒思考」；要不要发出去仍由既有的 decide 级联 + PushGate 决定
|
||||
// （C.4/C.5/C.7），本文件不碰推送策略。
|
||||
|
||||
// 自排的哨兵错误。用 errors.Is 判定；它们被直接返回或用 %w 包装。
|
||||
var (
|
||||
// ErrSelfWakeUnavailable 在服务未 Start 或已 Stop 时返回 —— **绝不假装排上**
|
||||
// （红线：不伪造成功）。
|
||||
ErrSelfWakeUnavailable = errors.New("backthink: self wake unavailable")
|
||||
// ErrSelfWakeNotOwner 在非主人环境下请求自排时返回（工具面是第一道闸，
|
||||
// 这里是第二道）。design/12 §C.3 / §C.5 的 denied。
|
||||
ErrSelfWakeNotOwner = errors.New("backthink: self wake requires owner environment")
|
||||
)
|
||||
|
||||
// SelfWakeReceipt 是一次自排请求的结果。字段与工具结果一一对应（design/12 §C.1/C.3），
|
||||
// 因此「被上限拒绝」是她**看得见**的合法业务结果，而不是一个 error。
|
||||
type SelfWakeReceipt struct {
|
||||
// Accepted 报告是否真的排出一轮（合并 = true；超限 = false）。
|
||||
Accepted bool
|
||||
// Coalesced 报告这次请求与在队的那一轮合并了（不额外产生一轮、不花钱）。
|
||||
Coalesced bool
|
||||
// Reason 是拒绝/说明文案，可直接显示；成功时为空。
|
||||
Reason string
|
||||
// Used 是**本窗口已用**额度，在消耗之后取值（合并/拒绝路径取当前已用值，
|
||||
// 不因这次请求 +1）。
|
||||
Used int
|
||||
// Limit 与 WindowSec 是本窗口的口径（已 normalize 的值），供面板与工具面显示。
|
||||
Limit int
|
||||
WindowSec int
|
||||
}
|
||||
|
||||
// Auditor 是自排审计的窄接口：三字段与网关的 AuditEvent 同形（gateway/audit.go），
|
||||
// 因此 *gateway.Server 直接满足它。审计出口是**晚绑定**的（见 sinks.go 的
|
||||
// SetAuditSink）：网关比 backthink 晚组装。
|
||||
type Auditor interface {
|
||||
AuditEvent(who, what, result string)
|
||||
}
|
||||
|
||||
// 自排审计行的 who 与四个动作。what 用点号小写的稳定标签（与 tool.reject.not_allowed
|
||||
// 同族）；result 是可直接显示给主人的中文。
|
||||
const (
|
||||
selfWakeAuditWho = "backthink"
|
||||
whatSelfWakeQueued = "backthink.self_wake.queued"
|
||||
whatSelfWakeCoalesced = "backthink.self_wake.coalesced"
|
||||
whatSelfWakeCapped = "backthink.self_wake.capped"
|
||||
whatSelfWakeDenied = "backthink.self_wake.denied"
|
||||
)
|
||||
|
||||
// WakeNow 是她自己排起一轮思考的入口（design/12 §N C.1）。它只负责**排**：
|
||||
// 探槽、过闸、入槽，或如实返回合并/拒绝的收据；真正的思考由单 worker 跑。
|
||||
//
|
||||
// 步骤（顺序不许颠倒，C.1 步骤 1~4）：
|
||||
// 1. 未 Start / 已 Stop ⇒ ErrSelfWakeUnavailable（不假装排上）；
|
||||
// 2. 非主人环境 ⇒ ErrSelfWakeNotOwner + denied 审计（**不做事**）；
|
||||
// 3. 先探容量 1 的待跑槽：满 ⇒ 合并（Accepted=true、Coalesced=true），
|
||||
// **不过窗口闸、不扣额度**，在队的那一轮仍用先到的理由；
|
||||
// 4. 槽空 ⇒ 过闸：拒 ⇒ Accepted=false + capped 审计 + 带 source 的 WARN
|
||||
// （Allow() 返回 false 时本就未消耗 ⇒ 拒绝不占额度、不产生轮次）；
|
||||
// 过 ⇒ 入槽（额度此刻已消耗）+ queued 审计。
|
||||
//
|
||||
// ctx 只用于身份环境与日志；待跑的那一轮不持有它（不引入延时/欠账语义，
|
||||
// 也不让一次请求把调用方的 ctx 生命周期绑进 worker）。
|
||||
func (b *Backthink) WakeNow(ctx context.Context, req SelfWakeRequest) (SelfWakeReceipt, error) {
|
||||
gate, running := b.wakeGate()
|
||||
if gate == nil || !running {
|
||||
return SelfWakeReceipt{}, ErrSelfWakeUnavailable
|
||||
}
|
||||
if ctx == nil {
|
||||
ctx = context.Background()
|
||||
}
|
||||
// 第二道身份闸（第一道在工具面，C.3）：非主人环境一律拒绝，并留一条 denied
|
||||
// 审计 —— 拒绝要看得见，不许静默。loop.SessionEnvFrom 对未挂载 env 的 ctx 返回
|
||||
// 主人默认值（loop 的既有向后兼容口径），因此这条判定与工具面读的是同一个环境。
|
||||
if env := loop.SessionEnvFrom(ctx); env.ActorEnv != restrict.ActorOwner {
|
||||
b.auditEvent(whatSelfWakeDenied, fmt.Sprintf("非主人环境，拒绝自排（actor=%s）", env.ActorEnv))
|
||||
return SelfWakeReceipt{}, ErrSelfWakeNotOwner
|
||||
}
|
||||
|
||||
receipt, queued := gate.offer(req)
|
||||
b.logSelfWake(req, receipt, queued)
|
||||
return receipt, nil
|
||||
}
|
||||
|
||||
// logSelfWake 把一次自排请求落到审计与日志上（四条 what 里的三条；denied 在
|
||||
// WakeNow 的身份闸处记）。审计失败不改变结果：审计是留痕，不是门。
|
||||
func (b *Backthink) logSelfWake(req SelfWakeRequest, receipt SelfWakeReceipt, queued bool) {
|
||||
switch {
|
||||
case receipt.Coalesced:
|
||||
b.auditEvent(whatSelfWakeCoalesced,
|
||||
fmt.Sprintf("已合并到排队中的那一轮（不额外占额度），理由：%s", req.Reason))
|
||||
b.log.Info("backthink self wake coalesced",
|
||||
"reason", req.Reason, "declared", req.Declared.String(), "used", receipt.Used, "limit", receipt.Limit)
|
||||
case receipt.Accepted:
|
||||
b.auditEvent(whatSelfWakeQueued, fmt.Sprintf("已排入一轮自主思考，理由：%s", req.Reason))
|
||||
b.log.Info("backthink self wake queued",
|
||||
"reason", req.Reason, "declared", req.Declared.String(), "used", receipt.Used, "limit", receipt.Limit)
|
||||
case queued:
|
||||
// 不可达：queued 为真时 Accepted 必为真。留着是为了让「排上了却没记审计」
|
||||
// 在下一次改动里立刻显形，而不是静默少一条。
|
||||
b.log.Warn("backthink self wake queued without audit", "reason", req.Reason)
|
||||
default:
|
||||
b.auditEvent(whatSelfWakeCapped, receipt.Reason)
|
||||
b.log.Warn("backthink self wake capped",
|
||||
"reason", req.Reason, "used", receipt.Used, "limit", receipt.Limit,
|
||||
"window", b.cfg.SelfWakeWindow.String())
|
||||
}
|
||||
}
|
||||
|
||||
// auditEvent 把一条审计写进晚绑定的出口（未接线时静默跳过：审计缺失不阻塞业务）。
|
||||
func (b *Backthink) auditEvent(what, result string) {
|
||||
if a := b.currentAuditor(); a != nil {
|
||||
a.AuditEvent(selfWakeAuditWho, what, result)
|
||||
}
|
||||
}
|
||||
|
||||
// wakeGate 在生命周期锁下取出自排闸并报告服务是否在跑。未 Start / 已 Stop 时
|
||||
// running 为 false ⇒ 调用方返回 ErrSelfWakeUnavailable（不假装排上）。
|
||||
func (b *Backthink) wakeGate() (*selfWakeGate, bool) {
|
||||
b.mu.Lock()
|
||||
defer b.mu.Unlock()
|
||||
return b.selfWake, b.running
|
||||
}
|
||||
|
||||
// selfWakeGate 是自排的**成本闸 + 待跑槽**（design/12 §N C.1/C.5）。它把三件事
|
||||
// 锁在一起：容量 1 的待跑槽、每窗口上限的滑窗，以及「探槽 → 过闸 → 入槽」的强制顺序。
|
||||
//
|
||||
// 锁序：本闸的互斥量只包住自己的状态；RateLimiter 自带内部锁且绝不回调本闸，
|
||||
// 因此不存在反向加锁路径（C.1 步骤 4 提到的「无锁序问题」）。
|
||||
type selfWakeGate struct {
|
||||
mu sync.Mutex
|
||||
// limiter 是每窗口上限的滑窗（复用既有件 ⇒ 零新依赖、零新机制）。
|
||||
// 它拿到手的必须是 normalize 之后的值：limit<=0 时它恒放行且不记录。
|
||||
limiter *session.RateLimiter
|
||||
limit int
|
||||
window time.Duration
|
||||
// granted 是「本窗口已用」的镜像，**只为上报**（收据的 Used 与审计文案）。
|
||||
// 判定始终由 limiter 做；两者用同一个时间源与同一个窗口，因此不会各说各话。
|
||||
granted []time.Time
|
||||
// pending 是容量 1 的待跑槽。非 nil ⇒ 满了，后来的请求合并进它。
|
||||
pending *SelfWakeRequest
|
||||
// ready 只用来唤醒 worker（容量 1，非阻塞发送）。它不承载状态：
|
||||
// 状态在 pending 上，因此不会出现「收到通知但槽是空的」以外的失配。
|
||||
ready chan struct{}
|
||||
}
|
||||
|
||||
// newSelfWakeGate 构建闸。limit/window 必须是**已 normalize** 的值
|
||||
// （≤0 会得到恒放行的滑窗，见 cfg.go 的 normalize 注释）。
|
||||
func newSelfWakeGate(limit int, window time.Duration) *selfWakeGate {
|
||||
return &selfWakeGate{
|
||||
limiter: session.NewRateLimiter(limit, window),
|
||||
limit: limit,
|
||||
window: window,
|
||||
ready: make(chan struct{}, 1),
|
||||
}
|
||||
}
|
||||
|
||||
// offer 落实 C.1 步骤 2~4：**先探槽**（满 ⇒ 合并，不过闸、不扣额度）→ **槽空才过闸**
|
||||
// → 入槽；三步在同一把锁内完成，因此两个并发自排不会同时看到「槽空」而各排一轮。
|
||||
//
|
||||
// 返回值 queued 报告是否真排出了一轮（合并 = false）。
|
||||
func (g *selfWakeGate) offer(req SelfWakeRequest) (SelfWakeReceipt, bool) {
|
||||
g.mu.Lock()
|
||||
defer g.mu.Unlock()
|
||||
|
||||
if g.pending != nil {
|
||||
// 合并路径**一次都不调用** Allow() —— 它成功即消耗，碰一下就是白扣额度。
|
||||
return g.receiptLocked(true, true, ""), false
|
||||
}
|
||||
if !g.limiter.Allow() {
|
||||
used := g.usedLocked()
|
||||
receipt := SelfWakeReceipt{
|
||||
Used: used,
|
||||
Limit: g.limit,
|
||||
WindowSec: int(g.window / time.Second),
|
||||
Reason: fmt.Sprintf("已达自我唤醒上限（本窗口 %d/%d，窗口 %s），本轮未排队",
|
||||
used, g.limit, g.window),
|
||||
}
|
||||
return receipt, false
|
||||
}
|
||||
|
||||
g.granted = append(g.granted, time.Now())
|
||||
queued := req
|
||||
g.pending = &queued
|
||||
select {
|
||||
case g.ready <- struct{}{}:
|
||||
default: // 已有一条通知在等：worker 取完槽后会再进内层循环，不会漏。
|
||||
}
|
||||
return g.receiptLocked(true, false, ""), true
|
||||
}
|
||||
|
||||
// take 取走待跑槽（worker 用）。槽空 ⇒ ok=false，调用方回到等待。
|
||||
func (g *selfWakeGate) take() (SelfWakeRequest, bool) {
|
||||
g.mu.Lock()
|
||||
defer g.mu.Unlock()
|
||||
if g.pending == nil {
|
||||
return SelfWakeRequest{}, false
|
||||
}
|
||||
req := *g.pending
|
||||
g.pending = nil
|
||||
return req, true
|
||||
}
|
||||
|
||||
// pendingDepth 报告待跑槽是否占用（探针用：判据 13 要确认合并、14 要确认槽已空）。
|
||||
func (g *selfWakeGate) pendingDepth() int {
|
||||
g.mu.Lock()
|
||||
defer g.mu.Unlock()
|
||||
if g.pending == nil {
|
||||
return 0
|
||||
}
|
||||
return 1
|
||||
}
|
||||
|
||||
// receiptLocked 组装一张收据（调用方必须持有 g.mu）。Used 在**消耗之后**取值。
|
||||
func (g *selfWakeGate) receiptLocked(accepted, coalesced bool, reason string) SelfWakeReceipt {
|
||||
return SelfWakeReceipt{
|
||||
Accepted: accepted,
|
||||
Coalesced: coalesced,
|
||||
Reason: reason,
|
||||
Used: g.usedLocked(),
|
||||
Limit: g.limit,
|
||||
WindowSec: int(g.window / time.Second),
|
||||
}
|
||||
}
|
||||
|
||||
// usedLocked 报告本窗口已用额度并顺手丢弃过期的记录（调用方必须持有 g.mu）。
|
||||
//
|
||||
// 它是**只读口径的镜像**：与 limiter 用同一个时间源（time.Now）与同一个窗口，
|
||||
// 因此报了「已用几档」，而不参与任何判定。
|
||||
func (g *selfWakeGate) usedLocked() int {
|
||||
cutoff := time.Now().Add(-g.window)
|
||||
keep := 0
|
||||
for _, t := range g.granted {
|
||||
if t.After(cutoff) {
|
||||
g.granted[keep] = t
|
||||
keep++
|
||||
}
|
||||
}
|
||||
g.granted = g.granted[:keep]
|
||||
return keep
|
||||
}
|
||||
|
||||
// selfWakeLoop 是自排 worker（design/12 §N C.1）：**单 goroutine** ⇒ 至多一轮自排在跑；
|
||||
// 待跑槽容量 1 ⇒ 至多一轮自排在排队。两者合起来就是「排一次只跑一轮 + 最多额外欠 1 轮」。
|
||||
//
|
||||
// 它不做延时：槽里的请求立刻跑（重启不产生补跑风暴）。ctx 取消（Stop）时退出；
|
||||
// 槽里若还留着一轮未跑，就地丢弃——没有持久化，也就没有欠账。
|
||||
func (b *Backthink) selfWakeLoop(ctx context.Context) {
|
||||
defer b.wg.Done()
|
||||
gate := b.selfWake
|
||||
if gate == nil {
|
||||
return
|
||||
}
|
||||
for {
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return
|
||||
case <-gate.ready:
|
||||
}
|
||||
// 内层循环把当时可取的槽位全部取走：合并语义保证它至多取到一轮，
|
||||
// 但循环写法让「取到空槽」成为正常出口，而不是靠时序假设。
|
||||
for {
|
||||
req, ok := gate.take()
|
||||
if !ok {
|
||||
break
|
||||
}
|
||||
b.runSelfWake(ctx, req)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// runSelfWake 跑一轮自排思考：折算走与事件同一条链（selfWakeTrigger → foldDeclared →
|
||||
// escalateImport），随后完全复用既有的 think（记录、判定、推送、落库）。
|
||||
func (b *Backthink) runSelfWake(ctx context.Context, req SelfWakeRequest) {
|
||||
b.log.Info("backthink self wake round",
|
||||
"reason", req.Reason, "declared", req.Declared.String())
|
||||
b.think(ctx, selfWakeTrigger(req))
|
||||
}
|
||||
@@ -0,0 +1,624 @@
|
||||
package backthink
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"strconv"
|
||||
"strings"
|
||||
"sync"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"git.yeij.top/AskaEth/Cyrene_v2/internal/capability/restrict"
|
||||
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/loop"
|
||||
)
|
||||
|
||||
// 本文件守 design/12 §N C 的**自排本体**（切片 C1-2）：WakeNow 的语义（探槽 →
|
||||
// 过闸 → 入槽，同锁）、容量 1 的待跑槽、每窗口上限、worker、审计四条 what。
|
||||
// 对应 C.8 判据 2 / 3 / 9 / 10 / 11 / 13 / 14。
|
||||
//
|
||||
// ⚠️ 本片交付的是**服务侧**能力：她**还调不到**（工具面 `request_think` 属 C1-3）
|
||||
// ⇒ 不得表述成「C 已完成」。
|
||||
|
||||
// wakeThinker 是自排用例用的假 Thinker：它在**进入**每一轮时记下触发串并发出信号，
|
||||
// 可选地「卡住」自排轮（holdSelfWake 未关闭时阻塞），让测试能构造「worker 正卡在
|
||||
// 一轮里」这个对撞场景。它尊重 ctx 取消，好让 Stop 能排空挂住的轮次。
|
||||
//
|
||||
// 为什么信号点在**进入**而不是完成：worker 在调用 Think 之前就已把待跑槽取空
|
||||
// （C.1 的 take → run），因此「某轮已进入」就是「槽已空」的同步点 —— 用它才能确定地
|
||||
// 测到窗口闸，而不是撞上「槽还满着」。卡住的轮次计入 triggers，但永不返回。
|
||||
type wakeThinker struct {
|
||||
mu sync.Mutex
|
||||
triggers []string
|
||||
// holdSelfWake 非 nil 时，自排轮阻塞到它被关闭（启动轮/心跳轮不受影响：
|
||||
// 它们跑在 tickLoop 上，不占自排 worker）。
|
||||
holdSelfWake chan struct{}
|
||||
entered chan struct{}
|
||||
}
|
||||
|
||||
func newWakeThinker() *wakeThinker {
|
||||
return &wakeThinker{entered: make(chan struct{}, 64)}
|
||||
}
|
||||
|
||||
func (w *wakeThinker) Think(ctx context.Context, sessionID, trigger string) (string, error) {
|
||||
if err := ctx.Err(); err != nil {
|
||||
return "", err
|
||||
}
|
||||
w.mu.Lock()
|
||||
w.triggers = append(w.triggers, trigger)
|
||||
w.mu.Unlock()
|
||||
select {
|
||||
case w.entered <- struct{}{}:
|
||||
default:
|
||||
}
|
||||
if w.holdSelfWake != nil && strings.HasPrefix(trigger, SelfWakeSummaryPrefix) {
|
||||
select {
|
||||
case <-w.holdSelfWake:
|
||||
case <-ctx.Done():
|
||||
return "", ctx.Err()
|
||||
}
|
||||
}
|
||||
return "no push decided", nil
|
||||
}
|
||||
|
||||
// selfWakeRounds 报告已进入的自排轮数（按 SelfWakeSummaryPrefix 认轮，启动轮与
|
||||
// 心跳轮不计）。
|
||||
func (w *wakeThinker) selfWakeRounds() int {
|
||||
w.mu.Lock()
|
||||
defer w.mu.Unlock()
|
||||
n := 0
|
||||
for _, tr := range w.triggers {
|
||||
if strings.HasPrefix(tr, SelfWakeSummaryPrefix) {
|
||||
n++
|
||||
}
|
||||
}
|
||||
return n
|
||||
}
|
||||
|
||||
// total 报告已进入的一轮总数（含启动轮与被卡住的那一轮）。
|
||||
func (w *wakeThinker) total() int {
|
||||
w.mu.Lock()
|
||||
defer w.mu.Unlock()
|
||||
return len(w.triggers)
|
||||
}
|
||||
|
||||
// selfWakeSummaries 返回已进入的自排轮触发摘要。它让用例能断言「合并之后这一轮用的是
|
||||
// **先到**的理由」—— 理由只出现在触发摘要里（落库行只存 Trigger = Source，不存摘要）。
|
||||
func (w *wakeThinker) selfWakeSummaries() []string {
|
||||
w.mu.Lock()
|
||||
defer w.mu.Unlock()
|
||||
out := make([]string, 0, len(w.triggers))
|
||||
for _, tr := range w.triggers {
|
||||
if strings.HasPrefix(tr, SelfWakeSummaryPrefix) {
|
||||
out = append(out, tr)
|
||||
}
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
// waitFor 等到至少 want 轮**已进入**（或超时失败）。它是「槽已空、该轮已在跑」的
|
||||
// 同步点：保证测的是**闸**，而不是「槽还满着 / worker 还没取槽」。
|
||||
func (w *wakeThinker) waitFor(t *testing.T, want int, timeout time.Duration) {
|
||||
t.Helper()
|
||||
deadline := time.Now().Add(timeout)
|
||||
for time.Now().Before(deadline) {
|
||||
if w.total() >= want {
|
||||
return
|
||||
}
|
||||
select {
|
||||
case <-w.entered:
|
||||
case <-time.After(5 * time.Millisecond):
|
||||
}
|
||||
}
|
||||
t.Fatalf("等待第 %d 轮进入超时（已进入 %d 轮）", want, w.total())
|
||||
}
|
||||
|
||||
// fakeAuditor 记录审计三元组。
|
||||
type fakeAuditor struct {
|
||||
mu sync.Mutex
|
||||
rows []auditRow
|
||||
}
|
||||
|
||||
type auditRow struct{ who, what, result string }
|
||||
|
||||
func (a *fakeAuditor) AuditEvent(who, what, result string) {
|
||||
a.mu.Lock()
|
||||
defer a.mu.Unlock()
|
||||
a.rows = append(a.rows, auditRow{who: who, what: what, result: result})
|
||||
}
|
||||
|
||||
// whats 返回按顺序记录的 what 列表。
|
||||
func (a *fakeAuditor) whats() []string {
|
||||
a.mu.Lock()
|
||||
defer a.mu.Unlock()
|
||||
out := make([]string, 0, len(a.rows))
|
||||
for _, r := range a.rows {
|
||||
out = append(out, r.what)
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
// resultOf 返回第一条匹配 what 的 result（没记到则失败）。
|
||||
func (a *fakeAuditor) resultOf(t *testing.T, what string) string {
|
||||
t.Helper()
|
||||
a.mu.Lock()
|
||||
defer a.mu.Unlock()
|
||||
for _, r := range a.rows {
|
||||
if r.what == what {
|
||||
return r.result
|
||||
}
|
||||
}
|
||||
t.Fatalf("审计里没有 %q：%+v", what, a.rows)
|
||||
return ""
|
||||
}
|
||||
|
||||
// countOf 返回某个 what 被记了几条。
|
||||
func (a *fakeAuditor) countOf(what string) int {
|
||||
a.mu.Lock()
|
||||
defer a.mu.Unlock()
|
||||
n := 0
|
||||
for _, r := range a.rows {
|
||||
if r.what == what {
|
||||
n++
|
||||
}
|
||||
}
|
||||
return n
|
||||
}
|
||||
|
||||
func (a *fakeAuditor) len() int {
|
||||
a.mu.Lock()
|
||||
defer a.mu.Unlock()
|
||||
return len(a.rows)
|
||||
}
|
||||
|
||||
// ownerCtx 是主人环境（工具面与 WakeNow 的第二道身份闸都读它）。
|
||||
func ownerCtx() context.Context {
|
||||
return loop.WithSessionEnv(context.Background(), loop.OwnerEnv())
|
||||
}
|
||||
|
||||
// startSelfWake 构建并启动一个后台思考服务，返回它、假 Thinker 与假审计出口。
|
||||
// 它**不**等服务启动轮跑完：启动轮恒占 Think 调用 1 次（计数基线），而 w.hold 非 nil 的
|
||||
// 对撞用例正是要让它卡在那里 —— 等待它反而会死等。自排计数按 SelfWakeSummaryPrefix
|
||||
// 认轮，因此启动轮混在总数里不影响断言。
|
||||
func startSelfWake(t *testing.T, cfg Config, w *wakeThinker) (*Backthink, *fakeAuditor) {
|
||||
t.Helper()
|
||||
audit := &fakeAuditor{}
|
||||
b, _ := newBackthink(t, cfg, Deps{
|
||||
Thinker: w,
|
||||
State: &mockState{state: UserState{}},
|
||||
Audit: audit,
|
||||
Log: discardLog(),
|
||||
})
|
||||
b.Start(context.Background())
|
||||
t.Cleanup(func() { _ = b.Stop(context.Background()) })
|
||||
return b, audit
|
||||
}
|
||||
|
||||
// TestWakeNow_UnavailableBeforeStart 是 C.8 判据 10（诚实）：未 Start 时 WakeNow
|
||||
// 必须返回 ErrSelfWakeUnavailable，**绝不假装排上**（返回 nil + 「已排队」）。
|
||||
//
|
||||
// 变异：未 Start 时返回 `SelfWakeReceipt{Accepted: true}, nil` ⇒ 本用例变红。
|
||||
func TestWakeNow_UnavailableBeforeStart(t *testing.T) {
|
||||
t.Parallel()
|
||||
b, _ := newBackthink(t, Config{}, Deps{Thinker: newWakeThinker(), State: &mockState{}})
|
||||
|
||||
receipt, err := b.WakeNow(ownerCtx(), SelfWakeRequest{Reason: "还没启动"})
|
||||
if !errors.Is(err, ErrSelfWakeUnavailable) {
|
||||
t.Fatalf("err = %v, want ErrSelfWakeUnavailable", err)
|
||||
}
|
||||
if receipt.Accepted {
|
||||
t.Fatalf("receipt = %+v, want 未被接受（不得假装排上）", receipt)
|
||||
}
|
||||
}
|
||||
|
||||
// TestWakeNow_UnavailableAfterStop 钉住 Stop 之后同样不可用（进程在收尾，不许再排）。
|
||||
func TestWakeNow_UnavailableAfterStop(t *testing.T) {
|
||||
t.Parallel()
|
||||
w := newWakeThinker()
|
||||
b, _ := startSelfWake(t, Config{}, w)
|
||||
if err := b.Stop(context.Background()); err != nil {
|
||||
t.Fatalf("Stop: %v", err)
|
||||
}
|
||||
if _, err := b.WakeNow(ownerCtx(), SelfWakeRequest{Reason: "太晚了"}); !errors.Is(err, ErrSelfWakeUnavailable) {
|
||||
t.Fatalf("err = %v, want ErrSelfWakeUnavailable", err)
|
||||
}
|
||||
}
|
||||
|
||||
// TestWakeNow_NotOwnerRefused 是 C.8 判据 7 的服务侧一半（工具面的那一半属 C1-3）：
|
||||
// 非主人环境一律拒绝，且**不做任何事**（不消耗额度、不入槽）。
|
||||
func TestWakeNow_NotOwnerRefused(t *testing.T) {
|
||||
t.Parallel()
|
||||
w := newWakeThinker()
|
||||
b, audit := startSelfWake(t, Config{}, w)
|
||||
|
||||
ctx := loop.WithSessionEnv(context.Background(), loop.SessionEnv{ActorEnv: restrict.ActorGroupMember})
|
||||
receipt, err := b.WakeNow(ctx, SelfWakeRequest{Reason: "群里想插一句"})
|
||||
if !errors.Is(err, ErrSelfWakeNotOwner) {
|
||||
t.Fatalf("err = %v, want ErrSelfWakeNotOwner", err)
|
||||
}
|
||||
if receipt.Accepted {
|
||||
t.Fatalf("receipt = %+v, want 未被接受", receipt)
|
||||
}
|
||||
if got := b.selfWake.pendingDepth(); got != 0 {
|
||||
t.Fatalf("待跑槽占用 = %d, want 0（拒绝时不得占槽）", got)
|
||||
}
|
||||
time.Sleep(50 * time.Millisecond)
|
||||
if got := w.selfWakeRounds(); got != 0 {
|
||||
t.Fatalf("自排轮数 = %d, want 0", got)
|
||||
}
|
||||
if got := audit.countOf(whatSelfWakeDenied); got != 1 {
|
||||
t.Fatalf("denied 审计 = %d 条, want 1（whats=%v）", got, audit.whats())
|
||||
}
|
||||
if !strings.Contains(audit.resultOf(t, whatSelfWakeDenied), string(restrict.ActorGroupMember)) {
|
||||
t.Fatalf("denied result = %q, want 它带上 actor", audit.resultOf(t, whatSelfWakeDenied))
|
||||
}
|
||||
}
|
||||
|
||||
// TestWakeNow_WindowLimitIsTheGate 是 C.8 判据 2：**每次等该轮真跑完再发下一次**
|
||||
// （保证槽是空的 ⇒ 测的是窗口闸而不是槽），limit=2 时第 3 次被拒、只跑 2 轮。
|
||||
//
|
||||
// 变异：把闸拆掉（恒放行）⇒ 第 3 次仍 Accepted=true ⇒ 本用例变红。
|
||||
func TestWakeNow_WindowLimitIsTheGate(t *testing.T) {
|
||||
t.Parallel()
|
||||
w := newWakeThinker()
|
||||
b, audit := startSelfWake(t, Config{SelfWakeLimit: 2, SelfWakeWindow: time.Hour}, w)
|
||||
|
||||
for i := 1; i <= 2; i++ {
|
||||
receipt, err := b.WakeNow(ownerCtx(), SelfWakeRequest{Reason: "第几次自排"})
|
||||
if err != nil {
|
||||
t.Fatalf("第 %d 次 WakeNow: %v", i, err)
|
||||
}
|
||||
if !receipt.Accepted || receipt.Coalesced {
|
||||
t.Fatalf("第 %d 次收据 = %+v, want 排上一轮", i, receipt)
|
||||
}
|
||||
if receipt.Used != i || receipt.Limit != 2 || receipt.WindowSec != 3600 {
|
||||
t.Fatalf("第 %d 次收据 = %+v, want Used=%d Limit=2 WindowSec=3600", i, receipt, i)
|
||||
}
|
||||
w.waitFor(t, 1+i, 3*time.Second) // 槽必须空，下一次才测得到闸
|
||||
}
|
||||
|
||||
receipt, err := b.WakeNow(ownerCtx(), SelfWakeRequest{Reason: "再来一次"})
|
||||
if err != nil {
|
||||
t.Fatalf("第 3 次 WakeNow: %v", err)
|
||||
}
|
||||
if receipt.Accepted || receipt.Coalesced {
|
||||
t.Fatalf("第 3 次收据 = %+v, want 被上限拒绝", receipt)
|
||||
}
|
||||
if receipt.Used != 2 || receipt.Limit != 2 {
|
||||
t.Fatalf("第 3 次收据 = %+v, want Used==Limit==2（拒绝不占额度）", receipt)
|
||||
}
|
||||
if receipt.Reason == "" {
|
||||
t.Fatal("拒绝必须给可显示的文案（她要看得见拒绝才会停手）")
|
||||
}
|
||||
time.Sleep(50 * time.Millisecond)
|
||||
if got := w.selfWakeRounds(); got != 2 {
|
||||
t.Fatalf("自排轮数 = %d, want 2（被拒的那次不产生轮次）", got)
|
||||
}
|
||||
if got := audit.countOf(whatSelfWakeCapped); got != 1 {
|
||||
t.Fatalf("capped 审计 = %d 条, want 1（whats=%v）", got, audit.whats())
|
||||
}
|
||||
if got := audit.countOf(whatSelfWakeQueued); got != 2 {
|
||||
t.Fatalf("queued 审计 = %d 条, want 2（whats=%v）", got, audit.whats())
|
||||
}
|
||||
}
|
||||
|
||||
// TestWakeNow_CoalescesIntoPendingSlot 是 C.8 判据 3 + 13 的**可确定**版本：先把
|
||||
// worker 钉在一轮自排里（假 Thinker 只卡自排轮，启动轮在 tickLoop 上、不占 worker），
|
||||
// 再连发 3 次 —— 第 1 次入槽、第 2/3 次**合并**，而合并**一次都不调用** Allow()
|
||||
// （它成功即消耗 ⇒ 碰一下就白扣额度）。
|
||||
//
|
||||
// 与 C.8 判据 13 字面的一处**偏离（如实登记）**：它写「①queued 的 Used=1」，那是把
|
||||
// 「卡住的那一轮」当成**不消耗额度**的启动轮来算的；可启动轮跑在 tickLoop 上，**不占**
|
||||
// 自排 worker（C.1 明确心跳/事件轮可与自排并发）⇒ 钉住 worker 的那一轮必然是自排轮、
|
||||
// 必然已消耗 1 次额度。因此本用例把「合并扣不扣额度」钉在 **Used 恒为钉住后的值** 上，
|
||||
// 探针性质与判据 13 相同且更锋利（额度只有 2）。详见 `C1-2-report.md`。
|
||||
//
|
||||
// 变异：① 把过闸挪到探槽**之前**（G1 顺序颠倒）⇒ 第 3 次白扣额度后 capped ⇒ 红；
|
||||
//
|
||||
// ② 待跑槽改无界（每次都入队）⇒ 第 2/3 次不接受合并且跑出 3 轮 ⇒ 红。
|
||||
func TestWakeNow_CoalescesIntoPendingSlot(t *testing.T) {
|
||||
t.Parallel()
|
||||
w := newWakeThinker()
|
||||
w.holdSelfWake = make(chan struct{}) // 钉住 worker（只卡自排轮）
|
||||
b, audit := startSelfWake(t, Config{SelfWakeLimit: 2, SelfWakeWindow: time.Hour}, w)
|
||||
|
||||
// ① 先排一轮并等 worker 真的把它取走、卡在里面（槽因此是空的）。
|
||||
if got, err := b.WakeNow(ownerCtx(), SelfWakeRequest{Reason: "钉住 worker 的那一轮"}); err != nil || !got.Accepted {
|
||||
t.Fatalf("钉住用的 WakeNow = %+v err = %v, want 排上", got, err)
|
||||
}
|
||||
w.waitFor(t, 2, 3*time.Second) // 启动轮 + 被钉住的自排轮
|
||||
if got := w.selfWakeRounds(); got != 1 {
|
||||
t.Fatalf("自排轮数 = %d, want 1（worker 已取槽并卡在该轮里）", got)
|
||||
}
|
||||
|
||||
// ② 连发 3 次：第 1 次入槽（额度剩 1 ⇒ Used=2），第 2/3 次合并。
|
||||
first, err := b.WakeNow(ownerCtx(), SelfWakeRequest{Reason: "先到的那一轮"})
|
||||
if err != nil {
|
||||
t.Fatalf("WakeNow: %v", err)
|
||||
}
|
||||
if !first.Accepted || first.Coalesced || first.Used != 2 {
|
||||
t.Fatalf("入槽的那次收据 = %+v, want 排队且 Used=2", first)
|
||||
}
|
||||
for i, reason := range []string{"第二条理由", "第三条理由"} {
|
||||
got, err := b.WakeNow(ownerCtx(), SelfWakeRequest{Reason: reason})
|
||||
if err != nil {
|
||||
t.Fatalf("WakeNow #%d: %v", i+2, err)
|
||||
}
|
||||
if !got.Accepted || !got.Coalesced {
|
||||
t.Fatalf("第 %d 次收据 = %+v, want 与在队的那一轮合并", i+2, got)
|
||||
}
|
||||
if got.Used != 2 {
|
||||
t.Fatalf("第 %d 次收据 = %+v, want Used 恒 2（合并不消耗预算）", i+2, got)
|
||||
}
|
||||
}
|
||||
if got := audit.countOf(whatSelfWakeCapped); got != 0 {
|
||||
t.Fatalf("capped 审计 = %d 条, want 0（槽满只合并，不拒绝；whats=%v）", got, audit.whats())
|
||||
}
|
||||
if got := audit.countOf(whatSelfWakeCoalesced); got != 2 {
|
||||
t.Fatalf("coalesced 审计 = %d 条, want 2（whats=%v）", got, audit.whats())
|
||||
}
|
||||
|
||||
close(w.holdSelfWake) // 放行：被钉住的那一轮跑完，worker 再取走合并后的那一轮
|
||||
w.waitFor(t, 3, 3*time.Second)
|
||||
if got := w.selfWakeRounds(); got != 2 {
|
||||
t.Fatalf("自排轮数 = %d, want 2（被钉住的 + 合并后的那一轮）", got)
|
||||
}
|
||||
if got := w.total(); got != 3 {
|
||||
t.Fatalf("Think 总调用 = %d, want 3（启动轮 + 被钉住的 + 合并后的一轮）", got)
|
||||
}
|
||||
// 在队的那一轮必须用**先到**的理由（C.1 步骤 2）：后两条理由一个字都不该出现。
|
||||
summaries := w.selfWakeSummaries()
|
||||
want := SelfWakeSummaryPrefix + "先到的那一轮"
|
||||
if len(summaries) != 2 || summaries[1] != want {
|
||||
t.Fatalf("自排轮摘要 = %#v, want 第二条（合并后的那一轮）是 %q", summaries, want)
|
||||
}
|
||||
}
|
||||
|
||||
// TestSelfWakeGate_MergeNeverTouchesGate 是 G1（C.8 判据 13 的**确定**探针）：直接对闸
|
||||
// 连发 3 次（不启动 worker ⇒ 槽不会被取走）—— 槽满时**一次都不调用** Allow()，因此
|
||||
// 「本窗口已用」在三次请求之间恒定。若把过闸挪到探槽之前，第 2 次就会白扣一次额度
|
||||
// ⇒ Used 变成 2、第 3 次直接 capped ⇒ 本用例变红。
|
||||
//
|
||||
// 它同时覆盖容量 1：连发 3 次只得到 **1 个**待跑位（第 2/3 次合并），没有无界堆积。
|
||||
func TestSelfWakeGate_MergeNeverTouchesGate(t *testing.T) {
|
||||
t.Parallel()
|
||||
gate := newSelfWakeGate(2, time.Hour)
|
||||
|
||||
first, queued := gate.offer(SelfWakeRequest{Reason: "先到"})
|
||||
if !queued || !first.Accepted || first.Coalesced || first.Used != 1 {
|
||||
t.Fatalf("第 1 次 offer = %+v queued=%v, want 入槽且 Used=1", first, queued)
|
||||
}
|
||||
for i := 0; i < 2; i++ {
|
||||
got, queued := gate.offer(SelfWakeRequest{Reason: "后来的"})
|
||||
if queued {
|
||||
t.Fatalf("第 %d 次 offer 又入槽了：容量 1 被打破（%+v）", i+2, got)
|
||||
}
|
||||
if !got.Accepted || !got.Coalesced || got.Used != 1 {
|
||||
t.Fatalf("第 %d 次 offer = %+v, want 合并且 Used 恒 1", i+2, got)
|
||||
}
|
||||
}
|
||||
if got := gate.pendingDepth(); got != 1 {
|
||||
t.Fatalf("待跑槽占用 = %d, want 1", got)
|
||||
}
|
||||
|
||||
// worker 取走槽位后：额度还剩 1 ⇒ 下一次入槽（Used=2），再下一次槽满 ⇒ 仍合并。
|
||||
if _, ok := gate.take(); !ok {
|
||||
t.Fatal("take 应取到在队的那一轮")
|
||||
}
|
||||
again, queued := gate.offer(SelfWakeRequest{Reason: "取走后第二次"})
|
||||
if !queued || again.Used != 2 {
|
||||
t.Fatalf("取走后再 offer = %+v queued=%v, want 入槽且 Used=2", again, queued)
|
||||
}
|
||||
if got, _ := gate.offer(SelfWakeRequest{Reason: "再次满槽"}); !got.Coalesced || got.Used != 2 {
|
||||
t.Fatalf("满槽时 offer = %+v, want 合并且 Used=2", got)
|
||||
}
|
||||
// 额度用尽且槽空 ⇒ 拒绝，且**不占额度**（Allow() 返回 false 本就未消耗）。
|
||||
if _, ok := gate.take(); !ok {
|
||||
t.Fatal("take 应取到第二次入槽的那一轮")
|
||||
}
|
||||
capped, queued := gate.offer(SelfWakeRequest{Reason: "超额"})
|
||||
if queued || capped.Accepted || capped.Coalesced {
|
||||
t.Fatalf("超额 offer = %+v queued=%v, want 被拒", capped, queued)
|
||||
}
|
||||
if capped.Used != 2 || capped.Limit != 2 {
|
||||
t.Fatalf("超额 offer = %+v, want Used==Limit==2（拒绝不占额度）", capped)
|
||||
}
|
||||
if !strings.Contains(capped.Reason, "已达自我唤醒上限") {
|
||||
t.Fatalf("拒绝文案 = %q, want 可直接显示的中文", capped.Reason)
|
||||
}
|
||||
}
|
||||
|
||||
// TestWakeNow_ZeroLimitFallsBackToDefault 是 C.8 判据 14（G3 探针）：配
|
||||
// SelfWakeLimit=0（并按同一口径再试一个负值）交给 New ⇒ 闸拿到的是 **normalize 后**
|
||||
// 的默认 4/1h；每次等该轮真跑完再发 ⇒ 前 4 次 Accepted=true（Used=1…4），第 5 次
|
||||
// Accepted=false + capped（Used==Limit==4）。
|
||||
//
|
||||
// 变异：把**未 normalize** 的 ≤0 直达 NewRateLimiter ⇒ Allow() 恒 true 且不记录
|
||||
// ⇒ 第 5 次仍 Accepted=true（静默失败：无报错、无 WARN）⇒ 本用例变红。
|
||||
func TestWakeNow_ZeroLimitFallsBackToDefault(t *testing.T) {
|
||||
t.Parallel()
|
||||
for _, limit := range []int{0, -1} {
|
||||
t.Run("limit="+strconv.Itoa(limit), func(t *testing.T) {
|
||||
t.Parallel()
|
||||
w := newWakeThinker()
|
||||
b, audit := startSelfWake(t, Config{SelfWakeLimit: limit, SelfWakeWindow: 0}, w)
|
||||
|
||||
for i := 1; i <= defaultSelfWakeLimit; i++ {
|
||||
got, err := b.WakeNow(ownerCtx(), SelfWakeRequest{Reason: "回落默认后的第几次"})
|
||||
if err != nil {
|
||||
t.Fatalf("第 %d 次 WakeNow: %v", i, err)
|
||||
}
|
||||
if !got.Accepted {
|
||||
t.Fatalf("第 %d 次收据 = %+v, want 排上（≤0 必须回落默认，而不是无上限）", i, got)
|
||||
}
|
||||
if got.Limit != defaultSelfWakeLimit || got.WindowSec != int(defaultSelfWakeWindow/time.Second) {
|
||||
t.Fatalf("第 %d 次收据 = %+v, want Limit=%d WindowSec=%d",
|
||||
i, got, defaultSelfWakeLimit, int(defaultSelfWakeWindow/time.Second))
|
||||
}
|
||||
if got.Used != i {
|
||||
t.Fatalf("第 %d 次收据 = %+v, want Used=%d", i, got, i)
|
||||
}
|
||||
w.waitFor(t, 1+i, 3*time.Second)
|
||||
}
|
||||
|
||||
got, err := b.WakeNow(ownerCtx(), SelfWakeRequest{Reason: "超出默认上限"})
|
||||
if err != nil {
|
||||
t.Fatalf("第 %d 次 WakeNow: %v", defaultSelfWakeLimit+1, err)
|
||||
}
|
||||
if got.Accepted {
|
||||
t.Fatalf("第 %d 次收据 = %+v, want capped（默认上限是 %d）",
|
||||
defaultSelfWakeLimit+1, got, defaultSelfWakeLimit)
|
||||
}
|
||||
if got.Used != defaultSelfWakeLimit || got.Limit != defaultSelfWakeLimit {
|
||||
t.Fatalf("capped 收据 = %+v, want Used==Limit==%d", got, defaultSelfWakeLimit)
|
||||
}
|
||||
if got := audit.countOf(whatSelfWakeCapped); got != 1 {
|
||||
t.Fatalf("capped 审计 = %d 条, want 1（whats=%v）", got, audit.whats())
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
// TestSelfWakeGate_WindowExpiry 覆盖「窗口过期后额度回来」。session.RateLimiter 内部
|
||||
// 用 time.Now()（design/12 §N C.5 已登记的取舍：不为自排再自造第二套滑窗），因此这里
|
||||
// 用极小窗口 + 真实短等，只取下界判断（等过窗口必放行），不依赖上界精度。
|
||||
func TestSelfWakeGate_WindowExpiry(t *testing.T) {
|
||||
t.Parallel()
|
||||
const window = 120 * time.Millisecond
|
||||
w := newWakeThinker()
|
||||
b, _ := startSelfWake(t, Config{SelfWakeLimit: 1, SelfWakeWindow: window}, w)
|
||||
|
||||
if got, err := b.WakeNow(ownerCtx(), SelfWakeRequest{Reason: "第一次"}); err != nil || !got.Accepted {
|
||||
t.Fatalf("第一次收据 = %+v err = %v, want 排上", got, err)
|
||||
}
|
||||
w.waitFor(t, 2, 3*time.Second)
|
||||
if got, err := b.WakeNow(ownerCtx(), SelfWakeRequest{Reason: "窗口内第二次"}); err != nil || got.Accepted {
|
||||
t.Fatalf("窗口内第二次收据 = %+v err = %v, want 被上限拒绝", got, err)
|
||||
}
|
||||
|
||||
time.Sleep(3 * window)
|
||||
got, err := b.WakeNow(ownerCtx(), SelfWakeRequest{Reason: "窗口过了"})
|
||||
if err != nil {
|
||||
t.Fatalf("窗口过后 WakeNow: %v", err)
|
||||
}
|
||||
if !got.Accepted {
|
||||
t.Fatalf("窗口过后的收据 = %+v, want 额度已随窗口释放", got)
|
||||
}
|
||||
if got.Used != 1 {
|
||||
t.Fatalf("窗口过后的收据 = %+v, want Used=1（旧记录已过期）", got)
|
||||
}
|
||||
}
|
||||
|
||||
// TestSelfWakeAudit_FourActions 是 C.8 判据 11：注入 fake Auditor，四个动作各记一条
|
||||
// 对应的 what，且 who 恒为 backthink。
|
||||
func TestSelfWakeAudit_FourActions(t *testing.T) {
|
||||
t.Parallel()
|
||||
w := newWakeThinker()
|
||||
w.holdSelfWake = make(chan struct{})
|
||||
b, audit := startSelfWake(t, Config{SelfWakeLimit: 2, SelfWakeWindow: time.Hour}, w)
|
||||
|
||||
// ① 入槽并被 worker 取走（Used=1）② 槽空、额度剩 1 ⇒ 再入槽（Used=2）
|
||||
if _, err := b.WakeNow(ownerCtx(), SelfWakeRequest{Reason: "排一轮"}); err != nil {
|
||||
t.Fatalf("WakeNow: %v", err)
|
||||
}
|
||||
w.waitFor(t, 2, 3*time.Second)
|
||||
if _, err := b.WakeNow(ownerCtx(), SelfWakeRequest{Reason: "再排一轮"}); err != nil {
|
||||
t.Fatalf("WakeNow: %v", err)
|
||||
}
|
||||
// ③ 槽满 ⇒ 合并。
|
||||
if got, err := b.WakeNow(ownerCtx(), SelfWakeRequest{Reason: "槽满时再来"}); err != nil || !got.Coalesced {
|
||||
t.Fatalf("槽满时收据 = %+v err = %v, want 合并", got, err)
|
||||
}
|
||||
// 放行后等第二轮被取走（槽空、额度已用尽）⇒ 超额请求落到 capped。
|
||||
close(w.holdSelfWake)
|
||||
w.waitFor(t, 3, 3*time.Second)
|
||||
if got, err := b.WakeNow(ownerCtx(), SelfWakeRequest{Reason: "超额"}); err != nil || got.Accepted {
|
||||
t.Fatalf("超额收据 = %+v err = %v, want 被拒", got, err)
|
||||
}
|
||||
// 非主人环境 ⇒ denied。
|
||||
ctx := loop.WithSessionEnv(context.Background(), loop.SessionEnv{ActorEnv: restrict.ActorStranger})
|
||||
if _, err := b.WakeNow(ctx, SelfWakeRequest{Reason: "陌生人想唤醒"}); !errors.Is(err, ErrSelfWakeNotOwner) {
|
||||
t.Fatalf("err = %v, want ErrSelfWakeNotOwner", err)
|
||||
}
|
||||
|
||||
// 本用例的序列是确定的（worker 被 holdSelfWake 钉住、步进都有 waitFor 同步点）：
|
||||
// 两次真的入槽（Used=1、Used=2）⇒ 两条 queued；满槽一次 ⇒ 一条 coalesced；
|
||||
// 额度用尽后槽空 ⇒ 一条 capped；非主人 ⇒ 一条 denied。删掉任一审计调用 ⇒ 变红。
|
||||
want := map[string]int{
|
||||
whatSelfWakeQueued: 2,
|
||||
whatSelfWakeCoalesced: 1,
|
||||
whatSelfWakeCapped: 1,
|
||||
whatSelfWakeDenied: 1,
|
||||
}
|
||||
for what, n := range want {
|
||||
if got := audit.countOf(what); got != n {
|
||||
t.Fatalf("what=%q 记了 %d 条, want %d（全部=%v）", what, got, n, audit.whats())
|
||||
}
|
||||
}
|
||||
audit.mu.Lock()
|
||||
defer audit.mu.Unlock()
|
||||
for _, r := range audit.rows {
|
||||
if r.who != selfWakeAuditWho {
|
||||
t.Fatalf("who = %q, want %q", r.who, selfWakeAuditWho)
|
||||
}
|
||||
if r.result == "" {
|
||||
t.Fatalf("result 为空（what=%q）：审计行必须能直接显示", r.what)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// TestSelfWake_WorkerRecordsSelfWakeRow 走**生产路径**（WakeNow → worker → think →
|
||||
// Store）：落库行的 Trigger 是 SelfWakeSource，落库的是折算后的档位（C.8 判据 9 的
|
||||
// 自排一半）。C1-1 的用例直接构造 selfWakeTrigger，这里证明「排一次真的跑一轮」。
|
||||
func TestSelfWake_WorkerRecordsSelfWakeRow(t *testing.T) {
|
||||
t.Parallel()
|
||||
w := newWakeThinker()
|
||||
b, _ := startSelfWake(t, Config{}, w)
|
||||
|
||||
if got, err := b.WakeNow(ownerCtx(), SelfWakeRequest{Reason: "复盘今天的对话"}); err != nil || !got.Accepted {
|
||||
t.Fatalf("WakeNow = %+v err = %v, want 排上", got, err)
|
||||
}
|
||||
w.waitFor(t, 2, 3*time.Second)
|
||||
|
||||
// 轮次已进入 ≠ 记录已落库（记录发生在 Think 返回之后），且启动轮与自排轮分属两个
|
||||
// goroutine ⇒ 这里等那一行真的写进去，再断言它的形状。
|
||||
var selfWake *Record
|
||||
eventually(t, 3*time.Second, func() bool {
|
||||
recs, err := b.List(context.Background(), 5)
|
||||
if err != nil {
|
||||
t.Fatalf("List: %v", err)
|
||||
}
|
||||
for i := range recs {
|
||||
if recs[i].Trigger == SelfWakeSource {
|
||||
selfWake = &recs[i]
|
||||
return true
|
||||
}
|
||||
}
|
||||
return false
|
||||
})
|
||||
if selfWake.Urgency != "routine" {
|
||||
t.Fatalf("Record.Urgency = %q, want routine（未声明档位不上抬）", selfWake.Urgency)
|
||||
}
|
||||
}
|
||||
|
||||
// TestSetAuditSink_LateBinding 钉住审计出口是晚绑定的（组合根在网关组装后调用），
|
||||
// 且换出口不需要重建推送器（审计不在推送路径上）。
|
||||
func TestSetAuditSink_LateBinding(t *testing.T) {
|
||||
t.Parallel()
|
||||
w := newWakeThinker()
|
||||
b, _ := newBackthink(t, Config{}, Deps{Thinker: w, State: &mockState{}, Log: discardLog()})
|
||||
before := b.currentPusher()
|
||||
|
||||
audit := &fakeAuditor{}
|
||||
b.SetAuditSink(audit)
|
||||
if b.currentAuditor() != audit {
|
||||
t.Fatal("SetAuditSink 之后 currentAuditor 不是注入的那个出口")
|
||||
}
|
||||
if b.currentPusher() != before {
|
||||
t.Fatal("SetAuditSink 不该重建推送器（审计不在推送路径上）")
|
||||
}
|
||||
b.SetAuditSink(nil)
|
||||
if b.currentAuditor() != nil {
|
||||
t.Fatal("SetAuditSink(nil) 应摘下出口（审计缺失不阻塞业务）")
|
||||
}
|
||||
}
|
||||
@@ -25,6 +25,23 @@ func (b *Backthink) SetIMSink(im IMAdapter) {
|
||||
b.rebuildPusherLocked()
|
||||
}
|
||||
|
||||
// SetAuditSink 接入审计出口（网关，design/12 §N C.5）。它与上面两个 setter 同属
|
||||
// **晚绑定**：网关比 backthink 晚组装，构造时拿不到。它不重建推送器——审计不在推送
|
||||
// 路径上，换一个出口（或注入 nil）都不影响投递。
|
||||
func (b *Backthink) SetAuditSink(a Auditor) {
|
||||
b.mu.Lock()
|
||||
defer b.mu.Unlock()
|
||||
b.deps.Audit = a
|
||||
}
|
||||
|
||||
// currentAuditor 在生命周期锁下返回当前审计出口，因此在途的一轮自排拿到的是
|
||||
// 一致快照，而不是「一半旧一半新」的两个出口。
|
||||
func (b *Backthink) currentAuditor() Auditor {
|
||||
b.mu.Lock()
|
||||
defer b.mu.Unlock()
|
||||
return b.deps.Audit
|
||||
}
|
||||
|
||||
// rebuildPusherLocked 用当前槽位、频率门与推送出口替换
|
||||
// 推送器。调用方必须持有 b.mu。
|
||||
func (b *Backthink) rebuildPusherLocked() {
|
||||
|
||||
@@ -14,9 +14,21 @@ import (
|
||||
type TriggerKind int
|
||||
|
||||
const (
|
||||
// TriggerUnknown 是 TriggerKind 的**零值**：没有任何触发类别被声明。
|
||||
//
|
||||
// 它存在，是因为零值必须落在**最紧**的一档：此前零值等于 TriggerHeartbeat，
|
||||
// 于是「漏填 Kind」静默变成「拿最宽的豁免」（长期静默段免检）——fail-open
|
||||
// （design/12 §N C.4 的 G-A 修法，主人 2026-10-06 已批）。判定侧对未知触发
|
||||
// **不豁免、不放行**（decide.go 的 activityHeld 让它落在最后那条扣下上），
|
||||
// 推送侧也不把它当事件（照旧过 PushGate）。
|
||||
//
|
||||
// ⚠️ 它不是第四个真实触发：生产路径的每个 Trigger 都必须显式给出 Kind
|
||||
// （heartbeatTrigger / startupTrigger / triggerForSubject / selfWakeTrigger
|
||||
// 四处构造点，backthink.go 的 think 是唯一消费点）。
|
||||
TriggerUnknown TriggerKind = iota
|
||||
// TriggerHeartbeat 是周期性自主反思。它的主动推送受
|
||||
// PushGate 限流，使主人绝不会被空洞的反思刷屏。
|
||||
TriggerHeartbeat TriggerKind = iota
|
||||
TriggerHeartbeat
|
||||
// TriggerEvent 是真实事件（IoT 设备变化、提醒、与主人相关的
|
||||
// 系统事件）。它的推送绕过 PushGate——"该发就发"——只受
|
||||
// 同一事件去重约束。
|
||||
@@ -25,8 +37,10 @@ const (
|
||||
// 事件、也不是周期性心跳：推送既不绕 PushGate，也不享长期静默段的豁免——
|
||||
// 要求只是她说得出理由（见 decide.go 的 activityHeld）。
|
||||
//
|
||||
// 它是**追加**的第三个值：既有两个值的编号与标签逐字不变（改编号会
|
||||
// 静默改写已落库记录与在途请求的含义）。
|
||||
// 它是**追加**的第四个值：既有三个值的**标签与语义逐字不变**。G-A 修法让
|
||||
// TriggerUnknown 占住零值，三个值因此各后移一位 —— 这只影响进程内的比较：
|
||||
// Trigger.Kind 与 PushRequest.TriggerKind 都不落库、不上线（落库的是
|
||||
// Record.Trigger = tr.Source 字符串），所以没有存量含义被改写。
|
||||
TriggerSelfWake
|
||||
)
|
||||
|
||||
@@ -39,6 +53,8 @@ func (k TriggerKind) String() string {
|
||||
return "self_wake"
|
||||
case TriggerHeartbeat:
|
||||
return "heartbeat"
|
||||
case TriggerUnknown:
|
||||
return "unknown"
|
||||
default:
|
||||
return fmt.Sprintf("triggerkind(%d)", int(k))
|
||||
}
|
||||
|
||||
Reference in new issue
Block a user