diff --git a/internal/config/runtimeconfig/backthink.go b/internal/config/runtimeconfig/backthink.go index 357cdfa..3594f88 100644 --- a/internal/config/runtimeconfig/backthink.go +++ b/internal/config/runtimeconfig/backthink.go @@ -34,6 +34,13 @@ const backthinkFileName = "backthink.json" // 但一旦某个家族换位置,共用常量会让另一处**静默**跟着搬家。 const backthinkConfigSubdir = "config" +// BackthinkConfigRelPath 是落点相对 `$DATA_DIR` 的路径(`config/backthink.json`)。 +// +// 单独导出它,是为了让面板能**如实**告诉主人「改哪个文件」,而不必把部署相关的 +// 绝对路径塞进接口形状(绝对路径随部署而变,由占用它的部署自己拼 `$DATA_DIR`)。 +// 它与 `DefaultBackthinkPath` 取自同一对常量(目录段 + 文件名),因此不会漂移。 +const BackthinkConfigRelPath = backthinkConfigSubdir + "/" + backthinkFileName + // ErrInvalidBackthink 是 backthink.json 结构不合法时的分类错误(errors.Is 可判类)。 var ErrInvalidBackthink = errors.New("runtimeconfig: invalid backthink config") @@ -42,6 +49,13 @@ var ErrInvalidBackthink = errors.New("runtimeconfig: invalid backthink config") // 字段名与 design/12 §N C.5 的规格逐字一致(camelCase 的 JSON 键,文件是给人看、 // 给人改的)。**没有版本字段**:这个家族只有这两个数字,等它真的长形状时再加版本, // 现在加一个永远为 0 的字段只是噪音。 +// +// ⚠️ **写路径未实现(切片 C2 如实登记)**:本文件至今只有读口。C.5 把「面板保存 + +// `SaveBackgroundTuning` + admin 端点」列为 C2,而面板写路径要经 design/09 §5 的 +// 面板声明机制(`Plugins.Declare` + 动作分发),那需要组合根接线(`cmd/cyrene`) +// 与一个 `backthink` 面板声明 —— 都不在 C2 的文件面内。因此 C2 只交付**读侧**: +// 运行期生效的旋钮经 `GET /api/v1/admin/backthink/self-wake` 可见(含代码默认值), +// 改动仍要**人手改这个文件 + 重启**(与 `QUIET_*` 同口径:启动期读一次)。 type BackgroundTuning struct { // SelfWakePerWindow 是每窗口允许的自排轮数。零/负 = 未声明或笔误 ⇒ // 由服务侧 normalize 回落代码默认(**没有「关闭上限」语义**)。 diff --git a/internal/services/backthink/backthink.go b/internal/services/backthink/backthink.go index 8cbd654..5d46183 100644 --- a/internal/services/backthink/backthink.go +++ b/internal/services/backthink/backthink.go @@ -44,7 +44,6 @@ import ( "time" "git.yeij.top/AskaEth/Cyrene_v2/internal/bus" - "git.yeij.top/AskaEth/Cyrene_v2/internal/iot" "git.yeij.top/AskaEth/Cyrene_v2/internal/logger" ) @@ -285,11 +284,11 @@ func (b *Backthink) List(ctx context.Context, limit int) ([]Record, error) { return b.store.List(ctx, limit) } -// triggers 返回本服务为事件驱动思考所监听的总线主题。 -// 它是一个方法(而非包级变量),因此不会出现包级全局 -// 切片,符合编码约束。 +// triggers 返回本服务为事件驱动思考所监听的总线主题。它是 +// `EventSubjects`(trigger_domain.go 的唯一真源)的转述 —— 订阅表与 admin +// 的 `?trigger=` 筛选白名单共用同一份,新增主题时两边同时生效。 func (b *Backthink) triggers() []string { - return []string{iot.DeviceUpdateSubject, SubjectReminderTrigger, SubjectSystemEvent} + return EventSubjects() } // subscribe 为每个触发主题注册一个处理器。每个订阅都 diff --git a/internal/services/backthink/selfwake_status.go b/internal/services/backthink/selfwake_status.go new file mode 100644 index 0000000..40377c3 --- /dev/null +++ b/internal/services/backthink/selfwake_status.go @@ -0,0 +1,70 @@ +package backthink + +import "time" + +// 本文件是自排闸的**只读快照**出口(design/12 §N C.5/C.2 的读侧,切片 C2)。 +// +// 为什么需要它:面板此前**看不到**「本窗口还能自排几次」—— 那个数字只出现在 +// 被拒绝那一轮的审计文案里(`backthink.self_wake.capped` 的 result),也就是 +// 「已经被拒一次」之后才第一次可见。C.5 的闸本来就维护了一份**只为上报**的镜像 +// (selfWakeGate.granted),本文件把它接出来,判定仍然只由 limiter 做 +// (§C.5:「判定始终由 limiter 做;两者用同一个时间源与同一个窗口」)。 +// +// 🔴 本文件**不做**任何写入、不碰滑窗件(`session.RateLimiter` 至今没有用量读口, +// 属别人的文件面):镜像在 selfWakeGate 里已经有了,读它即可。 + +// SelfWakeStatus 是自排闸的只读快照:本窗口已用多少、上限多少、窗口多长, +// 以及容量 1 的待跑槽是否被占用。 +// +// 它**不参与任何判定**(判定始终由滑窗与待跑槽做),因此读它永远不改变闸的状态。 +type SelfWakeStatus struct { + // Running 报告自排是否真的能排:未 Start / 已 Stop ⇒ false(此刻 WakeNow + // 会返回 ErrSelfWakeUnavailable)。面板据此显示「未运行」而不是「0 次」。 + Running bool + // Pending 报告容量 1 的待跑槽是否已占用(有一轮自排排着队还没跑)。 + Pending bool + // Used 是本窗口已用额度,与收据 `SelfWakeReceipt.Used` **同一口径** + // (同一次取数:过期的记录先被丢弃,再计数)。 + Used int + // Limit 与 WindowSec 是**进程生效**的上限与窗口(已 normalize;`≤0` 的那份 + // 配置不会走到这里,见 cfg.go 的 normalize 注释)。 + Limit int + WindowSec int + // DefaultLimit / DefaultWindowSec 是代码默认值;LimitIsDefault / WindowIsDefault + // 报告生效值**恰好等于**默认值。 + // + // ⚠️ 它**不是**「配置文件里没写这个键」的断言(与 config-facts 的 isDefault + // 同一条诚实口径:网关不读配置文件,也无从区分「配了 4」与「没配」)。 + DefaultLimit int + DefaultWindowSec int + LimitIsDefault bool + WindowIsDefault bool +} + +// SelfWakeStatus 取一张自排闸快照。第二个返回值为 false 表示**闸不在** +// (只有零值 / 未经 `New` 构造的 Backthink 会这样)—— 调用方**不得**把零值 +// 当成事实(不伪造「0 / 0」)。 +func (b *Backthink) SelfWakeStatus() (SelfWakeStatus, bool) { + gate, running := b.wakeGate() + if gate == nil { + return SelfWakeStatus{}, false + } + gate.mu.Lock() + // usedLocked 顺手丢弃过期记录,因此这里读到的与判定用的是同一个口径。 + used := gate.usedLocked() + pending := gate.pending != nil + limit, window := gate.limit, gate.window + gate.mu.Unlock() + + return SelfWakeStatus{ + Running: running, + Pending: pending, + Used: used, + Limit: limit, + WindowSec: int(window / time.Second), + DefaultLimit: defaultSelfWakeLimit, + DefaultWindowSec: int(defaultSelfWakeWindow / time.Second), + LimitIsDefault: limit == defaultSelfWakeLimit, + WindowIsDefault: window == defaultSelfWakeWindow, + }, true +} diff --git a/internal/services/backthink/selfwake_status_test.go b/internal/services/backthink/selfwake_status_test.go new file mode 100644 index 0000000..1bd3ac1 --- /dev/null +++ b/internal/services/backthink/selfwake_status_test.go @@ -0,0 +1,162 @@ +package backthink + +import ( + "context" + "testing" + "time" +) + +// 本文件守切片 C2 的**用量可见面**(design/12 §N C.5 的读侧):自排闸的只读快照 +// 必须与工具收据(SelfWakeReceipt)**同一口径**(同一个时间源、同一个窗口、 +// 同一个「消耗之后取值」),并且**读它不改变闸的状态**。 + +// TestSelfWakeStatus_MatchesReceiptAndTracksPendingSlot 是主判据: +// 排一轮后 `Used` 必须等于收据里的 Used,待跑槽随之变占用;额度用尽后 +// `Used` 停在 `Limit`(不会因为一次拒绝而虚增)。 +// +// 变异:把 SelfWakeStatus 的 Used 硬编码成 0(或改成 `len(granted)` 而不丢弃 +// 过期记录)⇒ 本用例在 `Used != receipt.Used` 处变红。 +func TestSelfWakeStatus_MatchesReceiptAndTracksPendingSlot(t *testing.T) { + t.Parallel() + // worker 卡住 ⇒ 排进去的那一轮不会把槽取走,`pending` 因此可确定地断言。 + w := newWakeThinker() + w.holdSelfWake = make(chan struct{}) + b, _ := startSelfWake(t, Config{SelfWakeLimit: 2, SelfWakeWindow: time.Hour}, w) + + // 启动即有一轮启动轮在跑(tickLoop 上的 TickImmediately),但它不占自排槽。 + before := mustStatus(t, b) + if !before.Running { + t.Fatal("已 Start 的服务其快照必须报告 Running=true") + } + if before.Used != 0 || before.Pending { + t.Fatalf("刚启动的快照 = %+v, want Used=0 / Pending=false", before) + } + if before.Limit != 2 || before.WindowSec != 3600 { + t.Fatalf("快照口径 = %+v, want Limit=2 / WindowSec=3600", before) + } + if before.LimitIsDefault { + t.Fatalf("上限 2 不该被算成默认值:%+v", before) + } + // 窗口仍取了代码默认(1 小时)⇒ WindowIsDefault 就是 true(这是「生效值恰好等于 + // 默认值」如实上报,而不是「配置文件没写」的断言)。 + if !before.WindowIsDefault { + t.Fatalf("窗口取了默认值却没标记:%+v", before) + } + if before.DefaultLimit != defaultSelfWakeLimit || before.DefaultWindowSec != int(defaultSelfWakeWindow/time.Second) { + t.Fatalf("默认值口径 = %+v, want %d / %d", before, + defaultSelfWakeLimit, int(defaultSelfWakeWindow/time.Second)) + } + + receipt, err := b.WakeNow(ownerCtx(), SelfWakeRequest{Reason: "排一轮"}) + if err != nil { + t.Fatalf("WakeNow: %v", err) + } + if !receipt.Accepted { + t.Fatalf("收据 = %+v, want Accepted", receipt) + } + after := mustStatus(t, b) + if after.Used != receipt.Used || after.Used != 1 { + t.Fatalf("快照 Used = %d, 收据 Used = %d, want 1(同一口径)", after.Used, receipt.Used) + } + if !after.Pending { + t.Fatalf("排入后待跑槽应报告占用:%+v", after) + } + + // 第二次撞上满槽 ⇒ 合并(不消耗额度),快照的 Used 不得变。 + if _, err := b.WakeNow(ownerCtx(), SelfWakeRequest{Reason: "再排一轮"}); err != nil { + t.Fatalf("WakeNow 合并: %v", err) + } + merged := mustStatus(t, b) + if merged.Used != 1 { + t.Fatalf("合并后 Used = %d, want 1(合并不额外占额度)", merged.Used) + } + if merged.Limit != 2 { + t.Fatalf("合并后 Limit = %d, want 2", merged.Limit) + } +} + +// TestSelfWakeStatus_StoppedServiceIsNotRunning 钉住「未运行」这一态: +// 服务在跑之前(以及 Stop 之后)快照必须如实报 Running=false,好让面板显示 +// 「此刻排不出新的一轮」而不是「已用 0 次」。 +// +// 变异:把 SelfWakeStatus 的 Running 直接写 true ⇒ 本用例变红。 +func TestSelfWakeStatus_StoppedServiceIsNotRunning(t *testing.T) { + t.Parallel() + b, _ := newBackthink(t, Config{}, Deps{Thinker: newWakeThinker(), State: &mockState{}}) + + st := mustStatus(t, b) + if st.Running { + t.Fatalf("未 Start 的快照 = %+v, want Running=false", st) + } + // 未运行时额度必然是 0,且上限已 normalize(面板仍能看到「上限 4/小时」)。 + if st.Used != 0 { + t.Fatalf("未运行时 Used = %d, want 0", st.Used) + } + if st.Limit != defaultSelfWakeLimit || !st.LimitIsDefault || !st.WindowIsDefault { + t.Fatalf("零配置的快照 = %+v, want 默认口径 %d/%v", st, + defaultSelfWakeLimit, defaultSelfWakeWindow) + } + + b.Start(context.Background()) + if !mustStatus(t, b).Running { + t.Fatal("Start 之后快照必须报告 Running=true") + } + if err := b.Stop(context.Background()); err != nil { + t.Fatalf("Stop: %v", err) + } + if mustStatus(t, b).Running { + t.Fatal("Stop 之后快照必须报告 Running=false") + } +} + +// TestSelfWakeStatus_ReadDoesNotConsumeBudget 钉住「读是纯读」:反复读快照 +// 不能让额度减少,也不能让待跑槽变化。 +// +// 变异:把快照实现改成调用 limiter.Allow()(成功即消耗)⇒ 本用例在 Used 处变红。 +func TestSelfWakeStatus_ReadDoesNotConsumeBudget(t *testing.T) { + t.Parallel() + b, _ := newBackthink(t, Config{SelfWakeLimit: 1, SelfWakeWindow: time.Hour}, + Deps{Thinker: newWakeThinker(), State: &mockState{}}) + b.Start(context.Background()) + t.Cleanup(func() { _ = b.Stop(context.Background()) }) + + for i := 0; i < 5; i++ { + if got := mustStatus(t, b).Used; got != 0 { + t.Fatalf("第 %d 次读快照后 Used = %d, want 0(读不改闸)", i+1, got) + } + } + // 读完之后额度仍然可用:唯一那一格必须能真排上一轮。 + receipt, err := b.WakeNow(ownerCtx(), SelfWakeRequest{Reason: "读完之后再排"}) + if err != nil { + t.Fatalf("WakeNow: %v", err) + } + if !receipt.Accepted || receipt.Used != 1 { + t.Fatalf("收据 = %+v, want Accepted 且 Used=1(读快照没有偷走额度)", receipt) + } +} + +// TestSelfWakeStatus_UnreadableOnlyWhenGateMissing 钉住第二个返回值的诚实语义: +// 正常构造的服务一定有闸(New 里建),只有零值 Backthink 才报告「看不到」。 +// +// 变异:把 `return SelfWakeStatus{}, false` 改成 `true` ⇒ 本用例变红 +// (调用方会把 0/0 当成真实用量)。 +func TestSelfWakeStatus_UnreadableOnlyWhenGateMissing(t *testing.T) { + t.Parallel() + if _, ok := (&Backthink{}).SelfWakeStatus(); ok { + t.Fatal("零值 Backthink 没有闸,必须报告 ok=false(不得伪造 0/0)") + } + b, _ := newBackthink(t, Config{}, Deps{Thinker: newWakeThinker(), State: &mockState{}}) + if _, ok := b.SelfWakeStatus(); !ok { + t.Fatal("经 New 构造的服务一定有闸,必须报告 ok=true") + } +} + +// mustStatus 取快照并在「闸不在」时直接失败(正常构造的服务不该走到那一步)。 +func mustStatus(t *testing.T, b *Backthink) SelfWakeStatus { + t.Helper() + st, ok := b.SelfWakeStatus() + if !ok { + t.Fatal("SelfWakeStatus 报告闸不在:正常构造的服务不该这样") + } + return st +} diff --git a/internal/services/backthink/store.go b/internal/services/backthink/store.go index 50ef393..3681f20 100644 --- a/internal/services/backthink/store.go +++ b/internal/services/backthink/store.go @@ -45,6 +45,14 @@ ALTER TABLE backthink_records ADD COLUMN urgency TEXT NOT NULL DEFAULT ''` const createRecordsIndexSQL = ` CREATE INDEX IF NOT EXISTS idx_backthink_records_at ON backthink_records(at)` +// createRecordsTriggerIndexSQL 支撑按触发源筛选的那条查询 +// (`WHERE trigger = ? ORDER BY id DESC`)。索引条目是 (trigger, rowid),而主键 +// `id` 就是 rowid ⇒ SQLite 能直接用一次索引范围扫描(反向)同时满足筛选与排序。 +// 它是**本片新增读路径**的配套:没有它,筛选就是全表扫描 + 排序(表按心跳每 15 分钟 +// 增长一行,一年就是三万余行)。 +const createRecordsTriggerIndexSQL = ` +CREATE INDEX IF NOT EXISTS idx_backthink_records_trigger ON backthink_records(trigger)` + const insertRecordSQL = ` INSERT INTO backthink_records (trigger, at, conclusion, should_push, push_pushed, push_route, push_detail, push_error, urgency) @@ -56,6 +64,19 @@ FROM backthink_records ORDER BY id DESC LIMIT ?` +// selectRecordsByTriggerSQL 与 selectRecordsSQL 逐字同形,只多一个按 +// `Record.Trigger` 的**服务端**筛选(参数化,不拼字符串)。 +// +// 为什么下推到 SQL 而不是「取一页再在内存里筛」:那样 `limit` 就变成「这一页里 +// 恰好匹配的条数」,面板会看到「自我唤醒 0 条」而库里其实有 —— 一个静默的假空值 +// (design/12 §N C.6 的服务端筛选要的正是「筛不到 ≠ 没发生」)。 +const selectRecordsByTriggerSQL = ` +SELECT id, trigger, at, conclusion, should_push, push_pushed, push_route, push_detail, push_error, urgency +FROM backthink_records +WHERE trigger = ? +ORDER BY id DESC +LIMIT ?` + // recordsTableName 是本模块自建表的物理名(DDL 与判列共用一处)。 const recordsTableName = "backthink_records" @@ -115,6 +136,9 @@ func NewStore(ctx context.Context, db *sql.DB, log *slog.Logger) (*Store, error) if _, err := db.ExecContext(ctx, createRecordsIndexSQL); err != nil { return nil, fmt.Errorf("backthink: ensure backthink_records index: %w", err) } + if _, err := db.ExecContext(ctx, createRecordsTriggerIndexSQL); err != nil { + return nil, fmt.Errorf("backthink: ensure backthink_records trigger index: %w", err) + } return &Store{db: db, log: log}, nil } @@ -207,9 +231,7 @@ func (s *Store) Append(ctx context.Context, r Record) (int64, error) { // defaultListLimit。urgency 列的空串如实读成「未知」(见 // Record.Urgency),不做任何补齐或推断。 func (s *Store) List(ctx context.Context, limit int) ([]Record, error) { - if limit <= 0 { - limit = defaultListLimit - } + limit = normListLimit(limit) if err := ctx.Err(); err != nil { return nil, err } @@ -218,8 +240,40 @@ func (s *Store) List(ctx context.Context, limit int) ([]Record, error) { return nil, fmt.Errorf("backthink: list records: %w", err) } defer rows.Close() + return scanRecords(rows, limit) +} - records := make([]Record, 0, limit) +// ListByTrigger 返回**某一触发源**的最近记录,最新的在前:语义与 List 逐字相同, +// 只多一个参数化的 `trigger = ?`(design/12 §N C.6 的服务端筛选)。 +// +// ⚠️ 它按**原始落库值**精确匹配(不模糊、不前缀):`Record.Trigger` 对事件轮是事件 +// 主题字符串,对内置轮是 `heartbeat` / `startup` / `self_wake` —— 合法取值由 +// `IsTriggerValue`(trigger_domain.go 的唯一真源)判定,本层不做二次裁剪。 +func (s *Store) ListByTrigger(ctx context.Context, trigger string, limit int) ([]Record, error) { + limit = normListLimit(limit) + if err := ctx.Err(); err != nil { + return nil, err + } + rows, err := s.db.QueryContext(ctx, selectRecordsByTriggerSQL, trigger, limit) + if err != nil { + return nil, fmt.Errorf("backthink: list records by trigger: %w", err) + } + defer rows.Close() + return scanRecords(rows, limit) +} + +// normListLimit 把非正的 limit 归一为 defaultListLimit(两条读路径同一口径)。 +func normListLimit(limit int) int { + if limit <= 0 { + return defaultListLimit + } + return limit +} + +// scanRecords 是两条读路径共用的行解码(抽出来是为了让「列序 / 空串语义」只有 +// 一处真源:再加第三条查询时不会悄悄错位)。 +func scanRecords(rows *sql.Rows, capacity int) ([]Record, error) { + records := make([]Record, 0, capacity) for rows.Next() { var ( r Record diff --git a/internal/services/backthink/trigger.go b/internal/services/backthink/trigger.go index e5d3cce..567054c 100644 --- a/internal/services/backthink/trigger.go +++ b/internal/services/backthink/trigger.go @@ -127,7 +127,7 @@ type Trigger struct { // heartbeatTrigger 是周期性、低重要度的触发。低重要度意味着 // 该轮很少打扰用户,遵守 "默认不发送 chat 类型消息"。 func heartbeatTrigger() Trigger { - return Trigger{Source: "heartbeat", Summary: "periodic autonomous reflection", Import: ImportanceLow, Kind: TriggerHeartbeat} + return Trigger{Source: HeartbeatSource, Summary: "periodic autonomous reflection", Import: ImportanceLow, Kind: TriggerHeartbeat} } // StartupSummary 是启动轮思考(裁决 07.4.4,design/12 §1)的触发摘要。它同时是 @@ -139,7 +139,7 @@ const StartupSummary = "process startup" // Kind 仍为 TriggerHeartbeat:启动轮的推送与心跳走完全相同的门与限频 // (PushGate / 打扰级联 / 活动门),进程反复重启也不会刷屏主人。 func startupTrigger() Trigger { - return Trigger{Source: "startup", Summary: StartupSummary, Import: ImportanceLow, Kind: TriggerHeartbeat} + return Trigger{Source: StartupSource, Summary: StartupSummary, Import: ImportanceLow, Kind: TriggerHeartbeat} } // triggerForEvent 把总线事件映射为 Trigger:主题给出可读摘要与主题级默认重要度, diff --git a/internal/services/backthink/trigger_domain.go b/internal/services/backthink/trigger_domain.go new file mode 100644 index 0000000..11086ec --- /dev/null +++ b/internal/services/backthink/trigger_domain.go @@ -0,0 +1,64 @@ +package backthink + +import ( + "context" + + "git.yeij.top/AskaEth/Cyrene_v2/internal/iot" +) + +// 本文件是「触发源取值域」的**唯一真源**:`Record.Trigger`(= `Trigger.Source` 字符串) +// 可能出现的取值,以及本服务订阅哪些事件主题。 +// +// 为什么要独立出来:admin 的 `GET /api/v1/admin/backthink?trigger=` **服务端筛选** +// (design/12 §N C.6,切片 C2)必须**如实拒绝未知取值**(400),而「未知」只能相对 +// 一个闭集判断。若在网关里另抄一份主题清单,新增订阅主题时就会出现「数据里有这个取值、 +// 筛选却报错」的静默漂移 —— 因此订阅表与筛选白名单从同一处取。 +// +// ⚠️ 与 design/12 §C.6「`trigger` 的取值域不封闭」**不矛盾**:那条约束的是**显示**口径 +// (客户端按字面值显示、不做闭集校验、不隐藏不认识的取值);这里定义的是**筛选白名单** +// —— 面向的是「筛选参数打错了要报错,而不是静默返回一个空列表」(静默的空列表会被 +// 读成「这段时间没有这一类的思考」,那正是本片要避免的假事实)。 + +// 内建触发标签。它们是 `Trigger.Source` 的字面值,也是 `Record.Trigger` 的落库值, +// 生产构造点分别是 heartbeatTrigger / startupTrigger(trigger.go)。 +const ( + // HeartbeatSource 是周期性自主反思(含启动轮之外的每次心跳)的标签。 + HeartbeatSource = "heartbeat" + // StartupSource 是进程启动后第一轮思考的标签(裁决 07.4.4)。 + StartupSource = "startup" +) + +// EventSubjects 返回本服务订阅的事件主题 —— **唯一真源**:`subscribe` 用它注册订阅, +// `TriggerFilterValues` 用它拼筛选白名单,因此新增一个订阅主题时两处同时生效 +// (不会出现「数据库里有新主题的记录、筛选却 400」的漂移)。 +func EventSubjects() []string { + return []string{iot.DeviceUpdateSubject, SubjectReminderTrigger, SubjectSystemEvent} +} + +// TriggerFilterValues 返回 `Record.Trigger` 可能出现的**全部取值**(闭集): +// 三个内建标签(心跳 / 启动 / 自排)+ 订阅的事件主题。 +// +// 它是 admin 筛选参数的白名单,也是「面板可选项」的真源;顺序稳定,便于报错文案 +// 与客户端下拉复用。 +func TriggerFilterValues() []string { + return append([]string{HeartbeatSource, StartupSource, SelfWakeSource}, EventSubjects()...) +} + +// IsTriggerValue 报告 v 是否是 `Record.Trigger` 可能出现的取值(白名单判定)。 +func IsTriggerValue(v string) bool { + for _, x := range TriggerFilterValues() { + if x == v { + return true + } + } + return false +} + +// ListByTrigger 按触发源读取记录(服务端筛选的读口,design/12 §N C.6 / 切片 C2)。 +// 它把筛选**下推**到存储层:`limit` 因此是「**匹配**的记录条数」,而不是 +// 「先取一页再筛掉不匹配的」(后者会给面板一个静默的假空值)。 +// +// 委托给 `*Store`(自己不认识 SQL);`List` 的语义、排序、空串口径一字不改。 +func (b *Backthink) ListByTrigger(ctx context.Context, trigger string, limit int) ([]Record, error) { + return b.store.ListByTrigger(ctx, trigger, limit) +} diff --git a/internal/services/backthink/trigger_domain_test.go b/internal/services/backthink/trigger_domain_test.go new file mode 100644 index 0000000..c148034 --- /dev/null +++ b/internal/services/backthink/trigger_domain_test.go @@ -0,0 +1,166 @@ +package backthink + +import ( + "context" + "testing" + "time" +) + +// 本文件守切片 C2 的两条读侧能力(design/12 §N C.5/C.6): +// 1. **触发源取值域**(trigger_domain.go 的唯一真源)—— 内建标签 + 订阅的事件主题, +// 它是 admin `?trigger=` 筛选白名单的依据; +// 2. **按触发源读取**(ListByTrigger 下推到 SQL)—— `limit` 必须是**匹配**条数, +// 而不是「先取一页再筛掉不匹配的」(后者会给面板一个静默的假空值)。 + +// TestTriggerFilterValues_CoversBuiltinsAndSubscribedSubjects 钉住白名单的**构成**: +// 三个内建标签(心跳 / 启动 / 自排)+ 本服务订阅的全部事件主题。 +// +// 变异:把 SelfWakeSource 从列表里摘掉 ⇒ 本用例变红(面板将筛不到自排轮, +// 而它恰恰是这一档最想看的);少列一个订阅主题 ⇒ 同红。 +func TestTriggerFilterValues_CoversBuiltinsAndSubscribedSubjects(t *testing.T) { + t.Parallel() + values := TriggerFilterValues() + set := map[string]bool{} + for _, v := range values { + if set[v] { + t.Fatalf("白名单里有重复取值 %q:%v", v, values) + } + set[v] = true + } + for _, want := range []string{HeartbeatSource, StartupSource, SelfWakeSource} { + if !set[want] { + t.Fatalf("白名单缺内建标签 %q:%v", want, values) + } + } + subjects := EventSubjects() + if len(subjects) == 0 { + t.Fatal("订阅主题表为空:白名单会漏掉全部事件轮") + } + for _, s := range subjects { + if !set[s] { + t.Fatalf("白名单缺订阅主题 %q:%v", s, values) + } + } + // 反向:白名单里的每个取值都必须真的可能出现在 Record.Trigger 上 + // (= 内建标签 或 订阅主题)—— 多列一个「不可能出现」的值会让筛选永远返回空。 + for _, v := range values { + if v == HeartbeatSource || v == StartupSource || v == SelfWakeSource { + continue + } + if !isEventSubject(v) { + t.Fatalf("白名单含一个不可能是触发源的取值 %q:%v", v, values) + } + } +} + +// isEventSubject 只在本测试里做反向核对(生产侧没有这个判定:白名单由 +// TriggerFilterValues 拼出来,不需要反查)。 +func isEventSubject(v string) bool { + for _, s := range EventSubjects() { + if s == v { + return true + } + } + return false +} + +// TestIsTriggerValue_RejectsUnknown 钉住白名单判定:已知取值通行,未知/空串一律拒。 +// +// 变异:把 IsTriggerValue 改成恒 true ⇒ 服务端筛选会「接受一切但筛不中」, +// 网关那条 400 的用例随之变红。 +func TestIsTriggerValue_RejectsUnknown(t *testing.T) { + t.Parallel() + for _, ok := range []string{HeartbeatSource, StartupSource, SelfWakeSource, "iot.device_update"} { + if !IsTriggerValue(ok) { + t.Fatalf("IsTriggerValue(%q) = false, want true", ok) + } + } + for _, bad := range []string{"", "event", "self-wake", "HEARTBEAT", "all", "iot"} { + if IsTriggerValue(bad) { + t.Fatalf("IsTriggerValue(%q) = true, want false", bad) + } + } +} + +// TestStore_ListByTrigger_IsServerSideFilter 钉住**下推**语义:筛选后 `limit` 数的是 +// 匹配条数(而不是「本页里恰好匹配的条数」)。 +// +// 变异:把 ListByTrigger 改回「List 之后内存筛」的退化实现(拿 limit 条再筛)⇒ +// 下面的 `limit=1` 断言会拿到 0 条(最新一条是心跳,被筛掉后没有别的自排)⇒ 变红。 +func TestStore_ListByTrigger_IsServerSideFilter(t *testing.T) { + t.Parallel() + ctx := context.Background() + s, err := NewStore(ctx, newTestDB(t), discardLog()) + if err != nil { + t.Fatalf("NewStore: %v", err) + } + base := time.Date(2026, 10, 6, 12, 0, 0, 0, time.UTC) + // 交替写入:心跳(最新)/ 自排 / 心跳 / 自排 / 心跳(最旧)。 + triggers := []string{HeartbeatSource, SelfWakeSource, HeartbeatSource, SelfWakeSource, HeartbeatSource} + for i, tr := range triggers { + if _, err := s.Append(ctx, Record{Trigger: tr, At: base.Add(time.Duration(i) * time.Minute)}); err != nil { + t.Fatalf("Append %d: %v", i, err) + } + } + + got, err := s.ListByTrigger(ctx, SelfWakeSource, 10) + if err != nil { + t.Fatalf("ListByTrigger: %v", err) + } + if len(got) != 2 { + t.Fatalf("自排记录 = %d 条, want 2(%+v)", len(got), got) + } + for _, r := range got { + if r.Trigger != SelfWakeSource { + t.Fatalf("筛选结果含非自排行:%+v", r) + } + } + // 最新的在前(与 List 同口径)。 + if got[0].ID != 4 || got[1].ID != 2 { + t.Fatalf("顺序 = %d,%d, want 4,2(最新在前)", got[0].ID, got[1].ID) + } + + // 🔴 下推的关键判据:limit 数的是**匹配**条数。 + one, err := s.ListByTrigger(ctx, SelfWakeSource, 1) + if err != nil { + t.Fatalf("ListByTrigger limit=1: %v", err) + } + if len(one) != 1 || one[0].ID != 4 { + t.Fatalf("limit=1 结果 = %+v, want 只有 id=4(匹配条数,不是先取一页再筛)", one) + } + + // 未出现过的合法取值 ⇒ 空列表(不是错误)。 + none, err := s.ListByTrigger(ctx, "reminder.trigger", 10) + if err != nil { + t.Fatalf("ListByTrigger(reminder.trigger): %v", err) + } + if len(none) != 0 { + t.Fatalf("无匹配时应返回空列表,得到 %+v", none) + } +} + +// TestBackthink_ListByTrigger_DelegatesToStore 钉住服务层的读口真的接到存储 +// (网关的可选接口 `backthinkTriggerLister` 靠的就是这个方法签名)。 +// +// 变异:把 `Backthink.ListByTrigger` 改成忽略 trigger 直接 List ⇒ 本用例变红。 +func TestBackthink_ListByTrigger_DelegatesToStore(t *testing.T) { + t.Parallel() + b, _ := newBackthink(t, Config{}, Deps{Thinker: newWakeThinker(), State: &mockState{}}) + ctx := context.Background() + for i := 0; i < 3; i++ { + if _, err := b.store.Append(ctx, Record{Trigger: HeartbeatSource, At: time.UnixMilli(int64(i))}); err != nil { + t.Fatalf("Append: %v", err) + } + } + if _, err := b.store.Append(ctx, Record{Trigger: SelfWakeSource, At: time.UnixMilli(99)}); err != nil { + t.Fatalf("Append: %v", err) + } + + got, err := b.ListByTrigger(ctx, SelfWakeSource, 50) + if err != nil { + t.Fatalf("ListByTrigger: %v", err) + } + if len(got) != 1 || got[0].Trigger != SelfWakeSource { + t.Fatalf("got = %+v, want 只有那条自排记录", got) + } +} diff --git a/internal/services/gateway/admin_backthink_filter.go b/internal/services/gateway/admin_backthink_filter.go new file mode 100644 index 0000000..ff565ab --- /dev/null +++ b/internal/services/gateway/admin_backthink_filter.go @@ -0,0 +1,133 @@ +package gateway + +import ( + "context" + "net/http" + "net/url" + "strconv" + "strings" + + "git.yeij.top/AskaEth/Cyrene_v2/internal/services/backthink" + + apperr "git.yeij.top/AskaEth/Cyrene_v2/internal/errors" +) + +// 本文件是 `GET /api/v1/admin/backthink` 的**服务端筛选**一侧(design/12 §N C.6, +// 切片 C2)。它从 admin_query.go 的 `adminListBackthink` 里把读侧抽出来,只做两件 +// 新事:① 解析并**校验** `?trigger=`(未知值如实 400);② 把筛选下推到存储层。 +// +// 三条口径(都在 C.6 与本片派工里点名,实现不得自行放宽): +// 1. **未知取值 ⇒ 400**,绝不静默忽略 —— 静默忽略会让面板在「显示的是全部」的同时 +// 声称「正在筛选自我唤醒」,那是一个假的筛选结果。 +// 2. **取值域 = 既有 + `self_wake`**:真源是 `backthink.TriggerFilterValues()` +// (内建标签 + 本服务订阅的事件主题),不在网关里另抄一份清单。 +// 3. **筛选下推**(不是「取一页再筛」):`limit` 是匹配条数,否则面板会看到 +// 「0 条」而库里其实有(同上,假空值)。 + +// backthinkTriggerLister 是按触发源下推筛选的**可选**读口:生产实现 +// `*backthink.Backthink` 满足它(下面有编译期断言钉住这一点)。 +// +// 为什么用可选接口而不是给 `BackthinkStore`(contracts.go,共享契约面)加一个方法: +// 加方法会让网关包内既有的测试替身(notifications_test.go 的 stubBackthinkFeed) +// 编译不过 —— 那属于别人的文件面。不满足本接口时退回「取一页再筛」并**如实记 WARN** +// (见 listBackthinkRecords),语义退化不静默发生。 +type backthinkTriggerLister interface { + ListByTrigger(ctx context.Context, trigger string, limit int) ([]backthink.Record, error) +} + +// 编译期钉住生产实现满足本可选口:形状一旦漂移,**编译期**就红,而不是等到运行期 +// 悄悄退化成「取一页再筛」(那条路径只有一条 WARN,肉眼很难发现)。 +var _ backthinkTriggerLister = (*backthink.Backthink)(nil) + +// parseBackthinkTrigger 解析并校验 `?trigger=`。返回空串 = 不过滤(参数缺席)。 +// +// 三个分支都刻意不静默: +// - 缺席 ⇒ 不过滤(既有行为,老客户端零影响); +// - **在场但为空** ⇒ 400:那多半是客户端把想筛选的值弄丢了,此时当作「不过滤」 +// 会让它以为自己在筛选(与「未知值静默忽略」是同一类假事实); +// - 不在白名单 ⇒ 400,并在文案里列出可选取值(客户端可直接显示)。 +func parseBackthinkTrigger(q url.Values) (string, error) { + if !q.Has("trigger") { + return "", nil + } + v := strings.TrimSpace(q.Get("trigger")) + if v == "" { + return "", apperr.Wrap(apperr.ErrBadRequest, "invalid_trigger", + "trigger 参数为空:要么省略该参数(返回全部),要么给一个合法取值:%s", + strings.Join(backthink.TriggerFilterValues(), " / ")) + } + if !backthink.IsTriggerValue(v) { + return "", apperr.Wrap(apperr.ErrBadRequest, "invalid_trigger", + "未知的触发源筛选 %q(可选值:%s)", + v, strings.Join(backthink.TriggerFilterValues(), " / ")) + } + return v, nil +} + +// listBackthinkRecords 取记录:给了 trigger 时优先**下推**筛选;实现不提供下推口时 +// 退回「取一页再筛」并记一条 WARN(那条 WARN 就是「本次口径退化了」的留痕)。 +func (s *Server) listBackthinkRecords(ctx context.Context, trigger string, limit int) ([]backthink.Record, error) { + if trigger == "" { + return s.backthink.List(ctx, limit) + } + if lister, ok := s.backthink.(backthinkTriggerLister); ok { + return lister.ListByTrigger(ctx, trigger, limit) + } + s.log.Warn("backthink trigger filter degraded: reader does not support pushdown", + "trigger", trigger, "limit", limit) + records, err := s.backthink.List(ctx, limit) + if err != nil { + return nil, err + } + return filterRecordsByTrigger(records, trigger), nil +} + +// filterRecordsByTrigger 是退化路径上的一页内筛选(下推不可用时才走这里)。 +func filterRecordsByTrigger(records []backthink.Record, trigger string) []backthink.Record { + out := make([]backthink.Record, 0, len(records)) + for _, r := range records { + if r.Trigger == trigger { + out = append(out, r) + } + } + return out +} + +// writeBackthinkList 是列表读侧的本体:取记录 → 补齐安静时段口径 → 下发。 +// 它由 `adminListBackthink`(admin_query.go 里的路由落点)在校验过筛选参数后调用。 +func (s *Server) writeBackthinkList(w http.ResponseWriter, r *http.Request, trigger string) { + out := make([]AdminBackthinkItem, 0) + if s.backthink == nil { + writeJSON(w, http.StatusOK, out) + return + } + records, err := s.listBackthinkRecords(r.Context(), trigger, backthinkListLimit) + if err != nil { + s.log.Error("list backthink", "err", err) + writeError(w, apperr.Wrap(apperr.ErrInternal, "backthink_failed", "failed to list backthink records")) + return + } + // 安静时段窗口取**运行时真正生效**的那一个(0/0 是「未填」而不是「没有安静时段」, + // 见 admin_config_facts.go 的 effectiveQuietHours —— 与 backthink 的 normalize 同口径), + // 否则 0/0 部署下会漏报真正的穿透。 + quietStart, quietEnd := s.effectiveQuietHours() + for _, rec := range records { + quiet := backthinkQuietOutcomeOf(rec, quietStart, quietEnd) + out = append(out, AdminBackthinkItem{ + ID: strconv.FormatInt(rec.ID, 10), + Trigger: rec.Trigger, + Conclusion: rec.Conclusion, + ShouldNotify: rec.ShouldPush, + Pushed: rec.PushResult.Pushed, + Route: string(rec.PushResult.Route), + Detail: rec.PushResult.Detail, + Bypassed: rec.PushResult.Bypassed, + At: rec.At.UnixMilli(), + // 紧急度(优先精确档位,旧行带下限标记)/ 是否穿透安静时段:只报存档可证的部分。 + Urgency: quiet.urgency, + UrgencyLowerBound: quiet.urgencyLowerBound, + QuietHoursBypassed: quiet.bypassed, + }) + } + writeJSON(w, http.StatusOK, out) +} diff --git a/internal/services/gateway/admin_backthink_filter_test.go b/internal/services/gateway/admin_backthink_filter_test.go new file mode 100644 index 0000000..1567599 --- /dev/null +++ b/internal/services/gateway/admin_backthink_filter_test.go @@ -0,0 +1,484 @@ +package gateway + +import ( + "context" + "encoding/json" + "net/http" + "net/url" + "strings" + "testing" + "time" + + "git.yeij.top/AskaEth/Cyrene_v2/internal/auth" + "git.yeij.top/AskaEth/Cyrene_v2/internal/db" + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/loop" + "git.yeij.top/AskaEth/Cyrene_v2/internal/services/backthink" +) + +// 本文件守切片 C2 的后端读侧(design/12 §N C.6 的服务端筛选 + C.5 的用量可见面): +// +// 1. `GET /api/v1/admin/backthink?trigger=` —— 白名单外的取值**如实 400** +// (绝不静默忽略:静默会让面板一边显示全部、一边声称在筛选); +// 2. 合法取值 ⇒ **下推**筛选(`limit` 是匹配条数); +// 3. `GET /api/v1/admin/backthink/self-wake` —— 只读快照的三态(未接线 / +// 未运行 / 运行中)必须可区分,且**不得**把「看不到」画成 0/0。 + +// triggerAwareFeed 是本片用的读侧替身:它实现可选口 ListByTrigger(下推筛选), +// 用来钉住「网关真的走了下推路径」;`pushdownCalls` 记录下推被调用的次数。 +type triggerAwareFeed struct { + records []backthink.Record + pushdownCalls int + err error +} + +func (f *triggerAwareFeed) List(_ context.Context, limit int) ([]backthink.Record, error) { + if f.err != nil { + return nil, f.err + } + if limit > 0 && limit < len(f.records) { + return append([]backthink.Record(nil), f.records[:limit]...), nil + } + return append([]backthink.Record(nil), f.records...), nil +} + +func (f *triggerAwareFeed) ListByTrigger(_ context.Context, trigger string, limit int) ([]backthink.Record, error) { + f.pushdownCalls++ + if f.err != nil { + return nil, f.err + } + out := make([]backthink.Record, 0, len(f.records)) + for _, r := range f.records { + if r.Trigger == trigger { + out = append(out, r) + } + } + if limit > 0 && limit < len(out) { + out = out[:limit] + } + return out, nil +} + +// getBackthink 以 owner 身份请求列表端点(query 是带前导 `?` 的查询串)。 +func getBackthink(t *testing.T, tsURL, query, tok string) (int, []byte, []string) { + t.Helper() + resp, raw := doReq(t, http.MethodGet, tsURL+"/api/v1/admin/backthink"+query, nil, tok) + if resp.StatusCode != http.StatusOK { + return resp.StatusCode, raw, nil + } + var list struct { + Data []AdminBackthinkItem `json:"data"` + } + if err := json.Unmarshal(raw, &list); err != nil { + t.Fatalf("decode: %v (%s)", err, raw) + } + out := make([]string, 0, len(list.Data)) + for _, it := range list.Data { + out = append(out, it.Trigger) + } + return resp.StatusCode, raw, out +} + +// sampleFeed 返回一份混合触发源的记录(最新在最前,与后端口径一致)。 +func sampleFeed() *triggerAwareFeed { + return &triggerAwareFeed{records: []backthink.Record{ + {ID: 5, Trigger: backthink.SelfWakeSource, At: time.UnixMilli(5000)}, + {ID: 4, Trigger: backthink.HeartbeatSource, At: time.UnixMilli(4000)}, + {ID: 3, Trigger: backthink.SelfWakeSource, At: time.UnixMilli(3000)}, + {ID: 2, Trigger: "iot.device_update", At: time.UnixMilli(2000)}, + {ID: 1, Trigger: backthink.HeartbeatSource, At: time.UnixMilli(1000)}, + }} +} + +// TestAdminBackthink_UnknownTriggerIs400 是 C.6 最要紧的一条:白名单外的 `trigger` +// 必须**如实 400**,绝不静默返回全部(那会让「正在筛选自我唤醒」变成假话)。 +// +// 变异:把 parseBackthinkTrigger 的未知值分支改成 `return "", nil`(静默忽略) +// ⇒ 本用例在「期望 400」处变红。 +func TestAdminBackthink_UnknownTriggerIs400(t *testing.T) { + t.Parallel() + feed := sampleFeed() + _, ts, jwt := buildEnv(t, nil, Deps{Backthink: feed}) + tok := signAny(t, jwt, "acct", auth.RoleOwner) + + for _, bad := range []string{"?trigger=event", "?trigger=self-wake", "?trigger=HEARTBEAT", "?trigger=all"} { + status, raw, _ := getBackthink(t, ts.URL, bad, tok) + if status != http.StatusBadRequest { + t.Fatalf("%s: status = %d, want 400(body=%s)", bad, status, raw) + } + if !strings.Contains(string(raw), `"invalid_trigger"`) { + t.Fatalf("%s: 400 必须带稳定的错误码 invalid_trigger:%s", bad, raw) + } + if !strings.Contains(string(raw), backthink.SelfWakeSource) { + t.Fatalf("%s: 文案应列出可选取值(含 self_wake):%s", bad, raw) + } + } + if feed.pushdownCalls != 0 { + t.Fatalf("非法取值不该触达存储:pushdownCalls = %d", feed.pushdownCalls) + } +} + +// TestAdminBackthink_EmptyTriggerIs400 钉住「在场但为空」的诚实口径:`?trigger=` +// 多半是客户端把想筛选的值弄丢了,此时当作「不过滤」会让它以为自己筛过了。 +// +// 变异:把空值分支合并进「缺席 ⇒ 不过滤」⇒ 本用例变红。 +func TestAdminBackthink_EmptyTriggerIs400(t *testing.T) { + t.Parallel() + _, ts, jwt := buildEnv(t, nil, Deps{Backthink: sampleFeed()}) + status, raw, _ := getBackthink(t, ts.URL, "?trigger=", signAny(t, jwt, "acct", auth.RoleOwner)) + if status != http.StatusBadRequest { + t.Fatalf("status = %d, want 400(body=%s)", status, raw) + } + if !strings.Contains(string(raw), `"invalid_trigger"`) { + t.Fatalf("400 必须带 invalid_trigger:%s", raw) + } +} + +// TestAdminBackthink_TriggerFilterPushesDown 钉住合法取值走**下推**:只回匹配行, +// 且值域里的事件主题同样可筛。 +// +// 变异:把 listBackthinkRecords 的下推分支删掉(一律走 List + 内存筛)⇒ +// pushdownCalls 断言变红(这也正是「口径退化」必须留痕的那条路径)。 +func TestAdminBackthink_TriggerFilterPushesDown(t *testing.T) { + t.Parallel() + feed := sampleFeed() + _, ts, jwt := buildEnv(t, nil, Deps{Backthink: feed}) + tok := signAny(t, jwt, "acct", auth.RoleOwner) + + status, raw, got := getBackthink(t, ts.URL, "?trigger="+backthink.SelfWakeSource, tok) + if status != http.StatusOK { + t.Fatalf("status = %d, body = %s", status, raw) + } + if len(got) != 2 || got[0] != backthink.SelfWakeSource || got[1] != backthink.SelfWakeSource { + t.Fatalf("筛选结果 = %v, want 两条自排", got) + } + if feed.pushdownCalls != 1 { + t.Fatalf("pushdownCalls = %d, want 1(筛选必须下推)", feed.pushdownCalls) + } + + // 事件主题也在白名单里(它是既有取值,不是新增的)。 + status, raw, got = getBackthink(t, ts.URL, "?trigger=iot.device_update", tok) + if status != http.StatusOK { + t.Fatalf("status = %d, body = %s", status, raw) + } + if len(got) != 1 || got[0] != "iot.device_update" { + t.Fatalf("筛选结果 = %v, want 一条 iot.device_update", got) + } +} + +// TestAdminBackthink_NoTriggerKeepsExistingShape 钉住老行为零改动:不给参数时 +// 返回全部(顺序与后端一致),且**不**调用下推口。 +// +// 变异:让缺参时也走 ListByTrigger ⇒ pushdownCalls 变红。 +func TestAdminBackthink_NoTriggerKeepsExistingShape(t *testing.T) { + t.Parallel() + feed := sampleFeed() + _, ts, jwt := buildEnv(t, nil, Deps{Backthink: feed}) + status, raw, got := getBackthink(t, ts.URL, "", signAny(t, jwt, "acct", auth.RoleOwner)) + if status != http.StatusOK { + t.Fatalf("status = %d, body = %s", status, raw) + } + if len(got) != 5 { + t.Fatalf("未筛选时应有 5 条,得到 %v", got) + } + if feed.pushdownCalls != 0 { + t.Fatalf("未筛选不该走下推口:pushdownCalls = %d", feed.pushdownCalls) + } +} + +// TestAdminBackthink_DegradedReaderStillFiltersOnePage 钉住**退化路径**:读侧替身 +// 不实现下推口时,仍要给出筛选结果(一页内筛),并如实记 WARN(WARN 无法在响应里 +// 断言,这里是「行为不静默失败」的探针)。 +// +// 变异:删掉退化分支 ⇒ 该路径会 panic(类型断言后直接调用)或返回全部 ⇒ 变红。 +func TestAdminBackthink_DegradedReaderStillFiltersOnePage(t *testing.T) { + t.Parallel() + // stubBackthinkFeed(notifications_test.go)只实现 List ⇒ 走退化路径。 + feed := &stubBackthinkFeed{records: sampleFeed().records} + _, ts, jwt := buildEnv(t, nil, Deps{Backthink: feed}) + status, raw, got := getBackthink(t, ts.URL, "?trigger="+backthink.SelfWakeSource, + signAny(t, jwt, "acct", auth.RoleOwner)) + if status != http.StatusOK { + t.Fatalf("status = %d, body = %s", status, raw) + } + if len(got) != 2 { + t.Fatalf("退化路径也应筛出两条自排,得到 %v", got) + } +} + +// statusReaderFeed 是自排状态读口的替身(网关只认这个可选口)。 +type statusReaderFeed struct { + triggerAwareFeed + st backthink.SelfWakeStatus + ok bool +} + +func (f *statusReaderFeed) SelfWakeStatus() (backthink.SelfWakeStatus, bool) { return f.st, f.ok } + +// getSelfWakeStatus 请求用量端点并解码。 +func getSelfWakeStatus(t *testing.T, tsURL, tok string) (int, []byte, AdminSelfWakeStatus) { + t.Helper() + resp, raw := doReq(t, http.MethodGet, tsURL+"/api/v1/admin/backthink/self-wake", nil, tok) + var env struct { + Data AdminSelfWakeStatus `json:"data"` + } + if resp.StatusCode == http.StatusOK { + if err := json.Unmarshal(raw, &env); err != nil { + t.Fatalf("decode: %v (%s)", err, raw) + } + } + return resp.StatusCode, raw, env.Data +} + +// TestAdminSelfWakeStatus_RunningSnapshot 钉住正常态:生效口径 + 本窗口用量 + +// 待跑槽 + 默认值标记一起下发。 +// +// 变异:把 Gate 里的 Used 换成 0(或漏配 Gate)⇒ 本用例变红。 +func TestAdminSelfWakeStatus_RunningSnapshot(t *testing.T) { + t.Parallel() + feed := &statusReaderFeed{ + st: backthink.SelfWakeStatus{ + Running: true, Pending: true, Used: 3, + Limit: 4, WindowSec: 3600, + DefaultLimit: 4, DefaultWindowSec: 3600, + LimitIsDefault: true, WindowIsDefault: true, + }, + ok: true, + } + _, ts, jwt := buildEnv(t, nil, Deps{Backthink: feed}) + status, raw, got := getSelfWakeStatus(t, ts.URL, signAny(t, jwt, "acct", auth.RoleOwner)) + if status != http.StatusOK { + t.Fatalf("status = %d, body = %s", status, raw) + } + if !got.Available || !got.Running { + t.Fatalf("status = %+v, want available+running", got) + } + if got.Reason != "" { + t.Fatalf("运行中不该带说明文案:%+v", got) + } + if got.Gate == nil { + t.Fatalf("运行中必须带 gate:%s", raw) + } + if got.Gate.Used != 3 || got.Gate.Limit != 4 || !got.Gate.Pending || got.Gate.WindowSec != 3600 { + t.Fatalf("gate = %+v, want used=3 limit=4 pending=true window=3600", got.Gate) + } + if !got.Gate.LimitIsDefault || !got.Gate.WindowIsDefault { + t.Fatalf("gate 默认值标记 = %+v, want 双 true", got.Gate) + } + if got.ConfigFile != "config/backthink.json" { + t.Fatalf("configFile = %q, want config/backthink.json", got.ConfigFile) + } + // 原始 JSON 也要真的带这些键(形状给客户端看,不只是 Go 侧结构)。 + for _, want := range []string{`"available":true`, `"running":true`, `"used":3`, `"pending":true`} { + if !strings.Contains(string(raw), want) { + t.Fatalf("响应缺 %s:%s", want, raw) + } + } +} + +// TestAdminSelfWakeStatus_HonestThreeStates 钉住「看不到 ≠ 0 次」这两态: +// 未接线 / 实现不提供读口 / 闸不在 ⇒ `available:false` 且 `gate` 缺席; +// 已接线但未运行 ⇒ `available:true, running:false` + 说明文案 + 配置面数字仍在。 +// +// 变异:把未接线一态改成返回 `gate:{used:0,limit:0}`(伪造 0/0)⇒ 变红。 +func TestAdminSelfWakeStatus_HonestThreeStates(t *testing.T) { + t.Parallel() + + t.Run("未接线", func(t *testing.T) { + _, ts, jwt := buildEnv(t, nil, Deps{}) + status, raw, got := getSelfWakeStatus(t, ts.URL, signAny(t, jwt, "acct", auth.RoleOwner)) + if status != http.StatusOK { + t.Fatalf("status = %d, body = %s", status, raw) + } + if got.Available || got.Gate != nil { + t.Fatalf("未接线时不得给 gate:%s", raw) + } + if got.Reason == "" { + t.Fatalf("未接线必须说明原因:%s", raw) + } + if strings.Contains(string(raw), `"gate"`) { + t.Fatalf("未接线时 gate 必须缺席:%s", raw) + } + }) + + t.Run("实现不提供读口", func(t *testing.T) { + // stubBackthinkFeed 不实现 SelfWakeStatus ⇒ 这一态。 + _, ts, jwt := buildEnv(t, nil, Deps{Backthink: &stubBackthinkFeed{}}) + _, raw, got := getSelfWakeStatus(t, ts.URL, signAny(t, jwt, "acct", auth.RoleOwner)) + if got.Available || got.Gate != nil || got.Reason == "" { + t.Fatalf("status = %+v, body = %s", got, raw) + } + }) + + t.Run("闸不在", func(t *testing.T) { + feed := &statusReaderFeed{ok: false} + _, ts, jwt := buildEnv(t, nil, Deps{Backthink: feed}) + _, raw, got := getSelfWakeStatus(t, ts.URL, signAny(t, jwt, "acct", auth.RoleOwner)) + if got.Available || got.Gate != nil || got.Reason == "" { + t.Fatalf("status = %+v, body = %s", got, raw) + } + }) + + t.Run("已接线未运行", func(t *testing.T) { + feed := &statusReaderFeed{ + st: backthink.SelfWakeStatus{Limit: 4, WindowSec: 3600, DefaultLimit: 4, DefaultWindowSec: 3600}, + ok: true, + } + _, ts, jwt := buildEnv(t, nil, Deps{Backthink: feed}) + _, raw, got := getSelfWakeStatus(t, ts.URL, signAny(t, jwt, "acct", auth.RoleOwner)) + if !got.Available || got.Running { + t.Fatalf("status = %+v, want available && !running", got) + } + if got.Gate == nil || got.Gate.Limit != 4 { + t.Fatalf("未运行时配置面数字仍应可见:%s", raw) + } + if got.Reason == "" { + t.Fatalf("未运行必须说明「此刻排不出新的一轮」:%s", raw) + } + }) +} + +// TestAdminSelfWakeStatus_RequiresOwnerOrAdmin 钉住档位:与列表端点同门同档, +// viewer 读不到(含她自排的成本口径)。匿名仍旧 401(fail-closed)。 +// +// 变异:把路由从 writeAuth 改成 viewerAuth ⇒ 本用例的 viewer 断言变红。 +func TestAdminSelfWakeStatus_RequiresOwnerOrAdmin(t *testing.T) { + t.Parallel() + feed := &statusReaderFeed{st: backthink.SelfWakeStatus{Running: true, Limit: 4, WindowSec: 3600}, ok: true} + _, ts, jwt := buildEnv(t, nil, Deps{Backthink: feed}) + + if status, _, _ := getSelfWakeStatus(t, ts.URL, signAny(t, jwt, "v", auth.RoleViewer)); status != http.StatusForbidden { + t.Fatalf("viewer status = %d, want 403", status) + } + if status, _, _ := getSelfWakeStatus(t, ts.URL, ""); status != http.StatusUnauthorized { + t.Fatalf("匿名 status = %d, want 401", status) + } + if status, _, _ := getSelfWakeStatus(t, ts.URL, signAny(t, jwt, "a", auth.RoleAdmin)); status != http.StatusOK { + t.Fatalf("admin status = %d, want 200", status) + } +} + +// TestParseBackthinkTrigger_Direct 直接钉住解析函数的两条边界(不经过 HTTP): +// 缺席 = 不过滤;带空白 = 先 trim 再判定(` self_wake ` 是合法取值)。 +// +// 变异:把 TrimSpace 去掉 ⇒ 第二组断言变红。 +func TestParseBackthinkTrigger_Direct(t *testing.T) { + t.Parallel() + got, err := parseBackthinkTrigger(url.Values{}) + if err != nil || got != "" { + t.Fatalf("缺席时 got=%q err=%v, want 空串 + nil", got, err) + } + q := url.Values{} + q.Set("trigger", " "+backthink.SelfWakeSource+" ") + got, err = parseBackthinkTrigger(q) + if err != nil { + t.Fatalf("trim 后合法:%v", err) + } + if got != backthink.SelfWakeSource { + t.Fatalf("got = %q, want %q(应先 trim)", got, backthink.SelfWakeSource) + } +} + +// fakeThinker 是给「真服务」用的最小 Thinker:只要它被调用就返回一句结论。 +// 它的存在让下面那条端到端用例不必依赖任何解释器或模型。 +type fakeThinker struct{} + +func (fakeThinker) Think(context.Context, string, string) (string, error) { + return "(测试结论)", nil +} + +// TestAdminBackthink_RealServiceSatisfiesBothReadPorts 是本片的**跨层闭环**: +// 用**真的** `*backthink.Backthink`(真存储 + 真滑窗 + 真待跑槽,内存 SQLite)当读侧, +// 经**真的** HTTP 路由读回来。它钉住三件只有真跑才成立的事: +// +// 1. 生产实现确实满足网关的两个**可选**读口(`backthinkTriggerLister` / +// `selfWakeStatusReader`)—— 两个 `var _ = (*backthink.Backthink)(nil)` 是编译期 +// 的第一道,这里是运行期的第二道(少了它,「可选口」会静默退化成「不提供读口」); +// 2. 服务端筛选对真存储真的按 `trigger = ?` 取(下推,不是拿一页再筛): +// 启动轮落的是 `startup`,自排轮落的是 `self_wake`,两者互不串门; +// 3. 用量快照与真闸同步:`WakeNow` 消耗一格后,端点报的 `used` 必须跟着变, +// 且**不是** 0/0 的假快照。 +func TestAdminBackthink_RealServiceSatisfiesBothReadPorts(t *testing.T) { + t.Parallel() + ctx := context.Background() + handle, err := db.Open(ctx, ":memory:", nil) + if err != nil { + t.Fatalf("open db: %v", err) + } + t.Cleanup(func() { _ = handle.Close() }) + + bt, err := backthink.New(backthink.Config{SelfWakeLimit: 2, SelfWakeWindow: time.Hour}, backthink.Deps{ + Thinker: fakeThinker{}, + DB: handle, + Log: nil, + }) + if err != nil { + t.Fatalf("backthink.New: %v", err) + } + bt.Start(ctx) + t.Cleanup(func() { _ = bt.Stop(context.Background()) }) + + _, ts, jwt := buildEnv(t, nil, Deps{Backthink: bt}) + tok := signAny(t, jwt, "acct", auth.RoleOwner) + + // 等启动轮落盘(异步),再用服务端筛选把它捞出来:`startup` 有、`self_wake` 还没有。 + waitFor(t, func() bool { + _, _, got := getBackthink(t, ts.URL, "?trigger="+backthink.StartupSource, tok) + return len(got) >= 1 + }, "启动轮落盘(?trigger=startup 应有一条)") + + status, raw, got := getBackthink(t, ts.URL, "?trigger="+backthink.SelfWakeSource, tok) + if status != http.StatusOK { + t.Fatalf("status = %d, body = %s", status, raw) + } + if len(got) != 0 { + t.Fatalf("自排还没发生,服务端筛选不该给出记录:%v", got) + } + + // 用量快照:真闸(limit=2),且此刻已用 0、在运行。 + _, raw, before := getSelfWakeStatus(t, ts.URL, tok) + if !before.Available || !before.Running || before.Gate == nil { + t.Fatalf("真服务的用量快照 = %+v, body = %s", before, raw) + } + if before.Gate.Used != 0 || before.Gate.Limit != 2 { + t.Fatalf("gate = %+v, want used=0 limit=2", before.Gate) + } + + // 真排一轮:主人环境 + 真服务 ⇒ 额度消耗一格。 + receipt, err := bt.WakeNow(loop.WithSessionEnv(ctx, loop.OwnerEnv()), + backthink.SelfWakeRequest{Reason: "端到端自排一轮"}) + if err != nil { + t.Fatalf("WakeNow: %v", err) + } + if !receipt.Accepted { + t.Fatalf("receipt = %+v, want Accepted", receipt) + } + + waitFor(t, func() bool { + _, _, rows := getBackthink(t, ts.URL, "?trigger="+backthink.SelfWakeSource, tok) + return len(rows) >= 1 + }, "自排轮落盘(?trigger=self_wake 应有一条)") + + _, raw, after := getSelfWakeStatus(t, ts.URL, tok) + if after.Gate == nil { + t.Fatalf("用量快照缺 gate:%s", raw) + } + if after.Gate.Used != 1 { + t.Fatalf("gate.Used = %d, want 1(真消耗一格;0 就是假快照)", after.Gate.Used) + } + if after.Gate.Limit != receipt.Limit { + t.Fatalf("端点上限 %d 与收据上限 %d 不一致", after.Gate.Limit, receipt.Limit) + } +} + +// waitFor 轮询 cond 直到为真或超时(异步落盘的同步点)。 +func waitFor(t *testing.T, cond func() bool, what string) { + t.Helper() + deadline := time.Now().Add(5 * time.Second) + for time.Now().Before(deadline) { + if cond() { + return + } + time.Sleep(10 * time.Millisecond) + } + t.Fatalf("等待超时:%s", what) +} diff --git a/internal/services/gateway/admin_backthink_selfwake.go b/internal/services/gateway/admin_backthink_selfwake.go new file mode 100644 index 0000000..9e3a390 --- /dev/null +++ b/internal/services/gateway/admin_backthink_selfwake.go @@ -0,0 +1,118 @@ +package gateway + +import ( + "net/http" + + "git.yeij.top/AskaEth/Cyrene_v2/internal/config/runtimeconfig" + "git.yeij.top/AskaEth/Cyrene_v2/internal/services/backthink" +) + +// 本文件是 `GET /api/v1/admin/backthink/self-wake` 的**只读用量可见面** +// (design/12 §N C.5 的读侧,切片 C2)。 +// +// 为什么要有它:C.5 的「本窗口已用 N / 上限 M」此前**只**出现在被拒那一轮的审计文案里 +// (`backthink.self_wake.capped` 的 result)—— 也就是「已经被拒过一次」才第一次看得见。 +// 面板要能提前告诉主人「还能自排几次」,就得有一个只读快照出口。 +// +// 它是**纯读**:不写库、不落审计、不碰闸(读快照永远不改变闸的状态),与列表端点同门 +// 同档(writeAuth = owner|admin;含她自排的成本口径与运行状态,不是给 viewer 的监控投影)。 +// +// ⚠️ 它**不是**写口:调这两个旋钮要改 `$DATA_DIR/config/backthink.json` + 重启 +// (启动期读一次,与 `QUIET_*` 同口径)。面板写路径未做,原因见 runtimeconfig +// `BackgroundTuning` 的注释(要经 design/09 §5 的面板声明机制,需组合根接线)。 + +// selfWakeStatusReader 是自排闸只读快照的**可选**读口:生产实现 +// `*backthink.Backthink` 满足它。用可选接口的理由同 backthinkTriggerLister +// (不改共享契约、不弄坏既有测试替身);不满足时如实回「本实现不提供读口」。 +type selfWakeStatusReader interface { + SelfWakeStatus() (backthink.SelfWakeStatus, bool) +} + +// 编译期钉住生产实现满足本可选口(形状漂移在编译期立刻显形,而不是等到运行期 +// 悄悄变成「本实现不提供自排状态读口」)。 +var _ selfWakeStatusReader = (*backthink.Backthink)(nil) + +// AdminSelfWakeGate 是自排闸的线上投影:**进程生效**的上限 / 窗口 + 本窗口用量 + +// 待跑槽占用,外带代码默认值(供面板区分「默认 / 已改」)。 +type AdminSelfWakeGate struct { + // Pending 报告容量 1 的待跑槽是否占用(有一轮自排排着队还没跑)。 + Pending bool `json:"pending"` + // Used 是本窗口已用额度(与工具收据 `SelfWakeReceipt.Used` 同一口径)。 + Used int `json:"used"` + // Limit / WindowSec 是**进程生效**的上限与窗口(已按「≤ 0 回落默认」normalize)。 + Limit int `json:"limit"` + WindowSec int `json:"windowSec"` + // DefaultLimit / DefaultWindowSec 是代码默认值;LimitIsDefault / WindowIsDefault + // 报告生效值**恰好等于**默认值。 + // + // ⚠️ 它**不是**「配置文件里没写这个键」的断言:网关**不读** `backthink.json` + // (与 config-facts 的 isDefault 同一条诚实口径 —— 无从不区分「配了 4」与「没配」)。 + DefaultLimit int `json:"defaultLimit"` + DefaultWindowSec int `json:"defaultWindowSec"` + LimitIsDefault bool `json:"limitIsDefault"` + WindowIsDefault bool `json:"windowIsDefault"` +} + +// AdminSelfWakeStatus 是 `GET /api/v1/admin/backthink/self-wake` 的响应。 +// +// 三态**必须可区分**(面板不得把「看不到」画成「0 次」): +// - `available:false` ⇒ 本进程给不出快照(服务未接线 / 实现不提供读口 / 闸不在), +// 此时 `gate` **缺席**、`reason` 说明原因; +// - `available:true, running:false` ⇒ 已接线但不在运行(未启动或已停止):`gate` +// 仍带配置面数字(上限/窗口/默认值),但此刻她排不出新的一轮 —— `reason` 说明; +// - `available:true, running:true` ⇒ 正常态,`reason` 缺席。 +type AdminSelfWakeStatus struct { + Available bool `json:"available"` + Running bool `json:"running"` + Reason string `json:"reason,omitempty"` + // ConfigFile 是「要改哪个文件」的相对落点(`config/backthink.json`,相对 `$DATA_DIR`): + // 绝对路径随部署而变,接口只给相对落点,由面板拼上下文说明。 + ConfigFile string `json:"configFile,omitempty"` + Gate *AdminSelfWakeGate `json:"gate,omitempty"` +} + +// adminSelfWakeStatus 下发自排闸的只读快照(见文件头注释)。 +func (s *Server) adminSelfWakeStatus(w http.ResponseWriter, r *http.Request) { + // 未接线:如实现状作答(与 adminListBackthink 返回空列表同口径:200 + 诚实说明, + // 不假装有一个 0/0 的闸)。 + if s.backthink == nil { + writeJSON(w, http.StatusOK, AdminSelfWakeStatus{ + Reason: "后台思考服务未接线:本进程不产生自排轮", + }) + return + } + reader, ok := s.backthink.(selfWakeStatusReader) + if !ok { + writeJSON(w, http.StatusOK, AdminSelfWakeStatus{ + Reason: "当前后端的后台思考实现不提供自排状态读口", + }) + return + } + st, ok := reader.SelfWakeStatus() + if !ok { + writeJSON(w, http.StatusOK, AdminSelfWakeStatus{ + Reason: "自排闸不在(后台思考服务未构造):没有可报的用量", + }) + return + } + + out := AdminSelfWakeStatus{ + Available: true, + Running: st.Running, + ConfigFile: runtimeconfig.BackthinkConfigRelPath, + Gate: &AdminSelfWakeGate{ + Pending: st.Pending, + Used: st.Used, + Limit: st.Limit, + WindowSec: st.WindowSec, + DefaultLimit: st.DefaultLimit, + DefaultWindowSec: st.DefaultWindowSec, + LimitIsDefault: st.LimitIsDefault, + WindowIsDefault: st.WindowIsDefault, + }, + } + if !st.Running { + out.Reason = "后台思考未在运行(未启动或已停止):此刻她排不出新的一轮" + } + writeJSON(w, http.StatusOK, out) +} diff --git a/internal/services/gateway/admin_query.go b/internal/services/gateway/admin_query.go index 56885e1..9910c03 100644 --- a/internal/services/gateway/admin_query.go +++ b/internal/services/gateway/admin_query.go @@ -3,7 +3,6 @@ package gateway import ( "encoding/json" "net/http" - "strconv" "strings" "time" @@ -363,41 +362,17 @@ func (s *Server) adminListUsers(w http.ResponseWriter, r *http.Request) { } // adminListBackthink 列出最近的后台思考记录(管理读)。 +// +// 它现在只做**筛选参数校验**并把读侧交给 writeBackthinkList +// (admin_backthink_filter.go,切片 C2):`?trigger=` 未知取值如实 400 +// (design/12 §N C.6 的服务端筛选),合法取值则下推到存储层取匹配条数。 func (s *Server) adminListBackthink(w http.ResponseWriter, r *http.Request) { - out := make([]AdminBackthinkItem, 0) - if s.backthink == nil { - writeJSON(w, http.StatusOK, out) - return - } - records, err := s.backthink.List(r.Context(), backthinkListLimit) + trigger, err := parseBackthinkTrigger(r.URL.Query()) if err != nil { - s.log.Error("list backthink", "err", err) - writeError(w, apperr.Wrap(apperr.ErrInternal, "backthink_failed", "failed to list backthink records")) + writeError(w, err) return } - // 安静时段窗口取**运行时真正生效**的那一个(0/0 是「未填」而不是「没有安静时段」, - // 见 admin_config_facts.go 的 effectiveQuietHours —— 与 backthink 的 normalize 同口径), - // 否则 0/0 部署下会漏报真正的穿透。 - quietStart, quietEnd := s.effectiveQuietHours() - for _, rec := range records { - quiet := backthinkQuietOutcomeOf(rec, quietStart, quietEnd) - out = append(out, AdminBackthinkItem{ - ID: strconv.FormatInt(rec.ID, 10), - Trigger: rec.Trigger, - Conclusion: rec.Conclusion, - ShouldNotify: rec.ShouldPush, - Pushed: rec.PushResult.Pushed, - Route: string(rec.PushResult.Route), - Detail: rec.PushResult.Detail, - Bypassed: rec.PushResult.Bypassed, - At: rec.At.UnixMilli(), - // 紧急度(优先精确档位,旧行带下限标记)/ 是否穿透安静时段:只报存档可证的部分。 - Urgency: quiet.urgency, - UrgencyLowerBound: quiet.urgencyLowerBound, - QuietHoursBypassed: quiet.bypassed, - }) - } - writeJSON(w, http.StatusOK, out) + s.writeBackthinkList(w, r, trigger) } // adminListRag 列出知识库文档,并在带 ?query= 时 diff --git a/internal/services/gateway/routes.go b/internal/services/gateway/routes.go index c36bd9d..01dc583 100644 --- a/internal/services/gateway/routes.go +++ b/internal/services/gateway/routes.go @@ -199,6 +199,13 @@ func (s *Server) registerAdminRoutes( mux.Handle("GET /api/v1/admin/llm/catalog", writeAuth(http.HandlerFunc(s.adminGetLlmCatalog))) mux.Handle("GET /api/v1/admin/users", writeAuth(http.HandlerFunc(s.adminListUsers))) mux.Handle("GET /api/v1/admin/backthink", writeAuth(http.HandlerFunc(s.adminListBackthink))) + // 自排闸只读快照(切片 C2,**v0.82**;design/12 §N C.5 的读侧):把 + // `$DATA_DIR/config/backthink.json` 的**进程生效值**(上限/窗口 + 代码默认值) + // 与「本窗口已用 N / 上限 M」(selfWakeGate 的只读镜像)下发给面板。 + // 只读、不审计、不写库;与上面的列表端点同门同档(writeAuth = owner|admin) + // —— 它是她自排的成本口径与运行状态,不是给 viewer 的监控投影。 + // ⚠️ 它**不是**写口:调这两个旋钮要改配置文件 + 重启(启动期读一次)。 + mux.Handle("GET /api/v1/admin/backthink/self-wake", writeAuth(http.HandlerFunc(s.adminSelfWakeStatus))) // 最近一轮链路图快照(**v0.71**):`graph_update` 只走实时帧、从不重放, // 前端图桶也不持久化 ⇒ 「轮次结束后才打开的页面」此前只能看到空态。 // 只读、不审计;与同一张图的 WS 下行帧同档(只发 owner|admin,