diff --git a/internal/services/backthink/backthink.go b/internal/services/backthink/backthink.go index 8a71202..0d47e31 100644 --- a/internal/services/backthink/backthink.go +++ b/internal/services/backthink/backthink.go @@ -300,6 +300,11 @@ func (b *Backthink) onEvent(ctx context.Context, e bus.Event) error { Conclusion: "system event: low value, not worth waking a thinking pass", ShouldPush: false, PushResult: PushResult{Route: RouteNone, Detail: "downgraded"}, + // Urgency 留空（未知）：这一行**不是一轮思考**（结论即写明未唤醒思考）， + // 没有「该轮生效的档位」可落；宁可如实为未知，也不拿事件的档位冒充—— + // 与 P3+P4 的 fail-safe 同口径。（降级只发生在 shouldThink 判到 + // Import < High 的系统事件上，故它的档位恒为 routine，面板若要显示， + // 由既有下限推导即可。） }) return nil } @@ -372,6 +377,8 @@ func (b *Backthink) think(parent context.Context, tr Trigger) { Conclusion: fmt.Sprintf("thinking failed: %v", err), ShouldPush: false, PushResult: PushResult{Route: RouteNone, Detail: err.Error()}, + // Urgency 留空（未知）：这一轮没跑完，**没有任何档位生效**； + // 拿触发的档位冒充「该轮档位」就是给一行失败记录安它没有的事实。 }) return } @@ -412,6 +419,11 @@ func (b *Backthink) think(parent context.Context, tr Trigger) { Conclusion: recordConclusion, ShouldPush: shouldPush, PushResult: pushRes, + // X2a：把**该轮实际生效**的档位落进存档（面板此前只能给「下限」）。 + // 取数点就是上面决策真正吃进去的 roundImport —— 她自报的紧急度经 + // escalatedImport 上抬后已折叠在内，因此记下的是生效值；她自报得不比 + // 事件自带档位低时，生效值就是事件那一档（声明只能上抬，见 escalateImport）。 + Urgency: urgencyForImportance(roundImport).String(), }) if shouldPush { diff --git a/internal/services/backthink/store.go b/internal/services/backthink/store.go index 134012f..50ef393 100644 --- a/internal/services/backthink/store.go +++ b/internal/services/backthink/store.go @@ -5,6 +5,7 @@ import ( "database/sql" "fmt" "log/slog" + "strings" "time" "git.yeij.top/AskaEth/Cyrene_v2/internal/logger" @@ -17,7 +18,8 @@ const defaultListLimit = 50 // 思考记录存储的 DDL。at/created_at 遵循项目时间戳 // 约定（带时区的 RFC3339）；trigger 与 conclusion 是面板渲染的 // 自由文本字段；push_* 列携带分解后的 PushResult， -// 使 List 不必解析 blob。主键是项目标准的自增 id。 +// 使 List 不必解析 blob；urgency 是该轮**实际生效**的紧急度档位 +// （空串 = 未知，见 Record.Urgency）。主键是项目标准的自增 id。 const createRecordsSQL = ` CREATE TABLE IF NOT EXISTS backthink_records ( id INTEGER PRIMARY KEY AUTOINCREMENT, @@ -29,23 +31,38 @@ CREATE TABLE IF NOT EXISTS backthink_records ( push_route TEXT NOT NULL, push_detail TEXT NOT NULL DEFAULT '', push_error TEXT NOT NULL DEFAULT '', + urgency TEXT NOT NULL DEFAULT '', created_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ','now')) )` +// alterRecordsAddUrgencySQL 给**既有**库补上 urgency 列：CREATE TABLE IF NOT EXISTS +// 对已经存在的表不加列，缺了它每一条 INSERT 都会以「no such column」失败 +// （思考存档整条链路断掉）。它属于本模块自建的表的演进，因此**不开迁移号**、 +// 也不动 internal/db 的迁移记账（那是 schema 所有者那边的领域，见包注释与 NewStore）。 +const alterRecordsAddUrgencySQL = ` +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)` const insertRecordSQL = ` INSERT INTO backthink_records - (trigger, at, conclusion, should_push, push_pushed, push_route, push_detail, push_error) -VALUES (?, ?, ?, ?, ?, ?, ?, ?)` + (trigger, at, conclusion, should_push, push_pushed, push_route, push_detail, push_error, urgency) +VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)` const selectRecordsSQL = ` -SELECT id, trigger, at, conclusion, should_push, push_pushed, push_route, push_detail, push_error +SELECT id, trigger, at, conclusion, should_push, push_pushed, push_route, push_detail, push_error, urgency FROM backthink_records ORDER BY id DESC LIMIT ?` +// recordsTableName 是本模块自建表的物理名（DDL 与判列共用一处）。 +const recordsTableName = "backthink_records" + +// tableInfoSQL 列举一张表的列。pragma_table_info 是 SQLite 3.16+ 的表值函数， +// 因此表名也能走占位符，不必把名字拼进 SQL。 +const tableInfoSQL = `SELECT name FROM pragma_table_info(?)` + // Record 是一轮已持久化的思考（design/12 §4），由 Store.List // 与服务的 List 返回。Trigger 是来源子系统标签，At 是决策 // 时间戳，Conclusion 是运行时的输出，ShouldPush 是打扰决策， @@ -57,6 +74,19 @@ type Record struct { Conclusion string ShouldPush bool PushResult PushResult + // Urgency 是**该轮实际生效**的紧急度档位，用 internal/urgency 的词汇字符串 + // （routine / warning / critical）：面板因此能区分 high 与 critical，而此前 + // 推送成功行的 PushResult.Detail 已被投递说明覆盖、档位在存档里丢失。 + // + // 空串**如实表示「未知」**：本列出现之前落下的旧行，以及从未走到档位判定点的 + // 轮次（思考本身失败）都读作未知 —— 绝不反推、绝不编成 routine（与 P3+P4 的 + // fail-safe 同口径：宁可不显示，也不谎报）。 + // + // JSON 键名刻意是 `urgency`：WS 实时帧发出去的正是本结构体，前端 + // （wsParse.ts）已按 `r.urgency` 防御性读取，带上它实时行才会显示档位， + // 而不必等重开页面由历史拉取补齐。omitempty 让未知保持「键缺席」—— + // 与「例常」可区分。 + Urgency string `json:"urgency,omitempty"` } // Store 把思考记录持久化到 SQLite。它在 @@ -68,7 +98,8 @@ type Store struct { log *slog.Logger } -// NewStore 构建 Store，确保 backthink_records 表与索引存在。 +// NewStore 构建 Store，确保 backthink_records 表（含既有库缺的 urgency 列， +// 见 ensureRecordsUrgencyColumn）与索引存在。 // db 为 nil 时返回 ErrNilDB。ctx 用于 DDL。 func NewStore(ctx context.Context, db *sql.DB, log *slog.Logger) (*Store, error) { if db == nil { @@ -78,13 +109,74 @@ func NewStore(ctx context.Context, db *sql.DB, log *slog.Logger) (*Store, error) if _, err := db.ExecContext(ctx, createRecordsSQL); err != nil { return nil, fmt.Errorf("backthink: ensure backthink_records: %w", err) } + if err := ensureRecordsUrgencyColumn(ctx, db, log); err != nil { + return nil, err + } if _, err := db.ExecContext(ctx, createRecordsIndexSQL); err != nil { return nil, fmt.Errorf("backthink: ensure backthink_records index: %w", err) } return &Store{db: db, log: log}, nil } -// Append 持久化 r 并返回新记录 id。 +// ensureRecordsUrgencyColumn **幂等**地保证 backthink_records 带 urgency 列。 +// 既有部署的库是先建好的，CREATE TABLE IF NOT EXISTS 那句不会给它加列，所以这里 +// 先按 PRAGMA 判存在、缺了才 ALTER。 +// +// 判到「缺列」与真的执行 ALTER 之间可能有另一个进程同时启动（同库双开），那次 +// ALTER 会以 duplicate column 失败 —— 那是同一个目标的另一次成功，容忍它； +// **其余任何错误都照实上报**（吞掉会让「列其实没加上」静默过关，随后每条 INSERT +// 都失败，而启动日志里什么也看不到）。 +func ensureRecordsUrgencyColumn(ctx context.Context, db *sql.DB, log *slog.Logger) error { + exists, err := columnExists(ctx, db, recordsTableName, "urgency") + if err != nil { + return err + } + if exists { + return nil + } + if _, err := db.ExecContext(ctx, alterRecordsAddUrgencySQL); err != nil { + if isDuplicateColumnErr(err) { + return nil + } + return fmt.Errorf("backthink: add backthink_records.urgency: %w", err) + } + if log != nil { + log.Info("backthink records table upgraded", "column", "urgency", "reason", "existing database predates the urgency column") + } + return nil +} + +// columnExists 报告 table 是否已有 column。 +func columnExists(ctx context.Context, db *sql.DB, table, column string) (bool, error) { + rows, err := db.QueryContext(ctx, tableInfoSQL, table) + if err != nil { + return false, fmt.Errorf("backthink: inspect %s columns: %w", table, err) + } + defer rows.Close() + for rows.Next() { + var name string + if err := rows.Scan(&name); err != nil { + return false, fmt.Errorf("backthink: scan %s column: %w", table, err) + } + if name == column { + return true, nil + } + } + if err := rows.Err(); err != nil { + return false, fmt.Errorf("backthink: iterate %s columns: %w", table, err) + } + return false, nil +} + +// isDuplicateColumnErr 报告 err 是否为「这一列已经存在」。SQLite 的措辞是 +// "duplicate column name: xxx"，驱动原样带回该串，故按子串判定；判错方向的代价 +// 不对称（把真错误当重复会掩盖缺列），因此这里只认这一种措辞。 +func isDuplicateColumnErr(err error) bool { + return err != nil && strings.Contains(strings.ToLower(err.Error()), "duplicate column name") +} + +// Append 持久化 r 并返回新记录 id。r.Urgency 原样落库（空串 = 未知， +// 见 Record.Urgency）——存储层不替调用方猜一个档位。 func (s *Store) Append(ctx context.Context, r Record) (int64, error) { res, err := s.db.ExecContext(ctx, insertRecordSQL, r.Trigger, @@ -95,6 +187,7 @@ func (s *Store) Append(ctx context.Context, r Record) (int64, error) { string(r.PushResult.Route), r.PushResult.Detail, r.PushResult.Err, + r.Urgency, ) if err != nil { return 0, fmt.Errorf("backthink: insert record: %w", err) @@ -104,13 +197,15 @@ func (s *Store) Append(ctx context.Context, r Record) (int64, error) { return 0, fmt.Errorf("backthink: record id: %w", err) } if s.log != nil { - s.log.Debug("backthink record appended", "id", id, "trigger", r.Trigger, "should_push", r.ShouldPush, "route", r.PushResult.Route) + s.log.Debug("backthink record appended", "id", id, "trigger", r.Trigger, + "should_push", r.ShouldPush, "route", r.PushResult.Route, "urgency", r.Urgency) } return id, nil } // List 返回最近的记录，最新的在前。非正的 limit 选用 -// defaultListLimit。 +// defaultListLimit。urgency 列的空串如实读成「未知」（见 +// Record.Urgency），不做任何补齐或推断。 func (s *Store) List(ctx context.Context, limit int) ([]Record, error) { if limit <= 0 { limit = defaultListLimit @@ -143,6 +238,7 @@ func (s *Store) List(ctx context.Context, limit int) ([]Record, error) { &route, &r.PushResult.Detail, &r.PushResult.Err, + &r.Urgency, ); err != nil { return nil, fmt.Errorf("backthink: scan record: %w", err) } diff --git a/internal/services/backthink/store_urgency_test.go b/internal/services/backthink/store_urgency_test.go new file mode 100644 index 0000000..1e31847 --- /dev/null +++ b/internal/services/backthink/store_urgency_test.go @@ -0,0 +1,350 @@ +package backthink + +import ( + "context" + "database/sql" + "encoding/json" + "path/filepath" + "strings" + "testing" + "time" + + "git.yeij.top/AskaEth/Cyrene_v2/internal/db" + "git.yeij.top/AskaEth/Cyrene_v2/internal/urgency" +) + +// 本文件守住 X2a 的**落库面**：每一轮实际生效的紧急度档位要能被持久化并按原样读回， +// 既有库要能被幂等地补上该列，补列前的旧行必须读成「未知」而不是某个档位， +// 且实时帧靠的 JSON 键名就是 `urgency`。摘掉加列、把空串当成 routine、或改掉 JSON +// 键名，这里即变红。 + +// legacyRecordsDDL 是本列出现**之前**的建表语句（逐字照抄旧版 store.go 的 +// createRecordsSQL），把「既有部署的库」搬进测试：新代码必须能给这样的表补列、 +// 让补列前落下的旧行读成「未知」，而不是让整个思考存档写不进去。 +const legacyRecordsDDL = ` +CREATE TABLE backthink_records ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + trigger TEXT NOT NULL, + at TEXT NOT NULL, + conclusion TEXT NOT NULL, + should_push INTEGER NOT NULL, + push_pushed INTEGER NOT NULL DEFAULT 0, + push_route TEXT NOT NULL, + push_detail TEXT NOT NULL DEFAULT '', + push_error TEXT NOT NULL DEFAULT '', + created_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ','now')) +)` + +// insertLegacyRecordSQL 按旧形状（无 urgency 列）插入一行。 +const insertLegacyRecordSQL = ` +INSERT INTO backthink_records + (trigger, at, conclusion, should_push, push_pushed, push_route, push_detail, push_error) +VALUES (?, ?, ?, ?, ?, ?, ?, ?)` + +// newLegacyRecordsDB 建一个「旧版形状 + 一行旧记录」的库。 +func newLegacyRecordsDB(t *testing.T) *sql.DB { + t.Helper() + h := newTestDB(t) + ctx := context.Background() + if _, err := h.ExecContext(ctx, legacyRecordsDDL); err != nil { + t.Fatalf("create legacy table: %v", err) + } + if _, err := h.ExecContext(ctx, insertLegacyRecordSQL, + "heartbeat", "2026-09-06T10:00:00Z", "晚上好", boolInt(false), boolInt(false), + string(RouteNone), "held", "", + ); err != nil { + t.Fatalf("insert legacy row: %v", err) + } + return h +} + +// hasColumn 直接查 PRAGMA 判列是否已存在。 +func hasColumn(t *testing.T, db *sql.DB, name string) bool { + t.Helper() + rows, err := db.QueryContext(context.Background(), tableInfoSQL, recordsTableName) + if err != nil { + t.Fatalf("inspect columns: %v", err) + } + defer rows.Close() + for rows.Next() { + var col string + if err := rows.Scan(&col); err != nil { + t.Fatalf("scan column: %v", err) + } + if col == name { + return true + } + } + if err := rows.Err(); err != nil { + t.Fatalf("iterate columns: %v", err) + } + return false +} + +// TestStore_Urgency_RoundTripEveryLevel 覆盖验收①：写入→回读逐档往返（routine / +// warning / critical 各一条），三档互不塌缩。 +func TestStore_Urgency_RoundTripEveryLevel(t *testing.T) { + t.Parallel() + ctx := context.Background() + s, err := NewStore(ctx, newTestDB(t), discardLog()) + if err != nil { + t.Fatalf("NewStore: %v", err) + } + levels := []urgency.Level{urgency.LevelRoutine, urgency.LevelWarning, urgency.LevelCritical} + base := time.Date(2026, 10, 5, 12, 0, 0, 0, time.UTC) + for i, lv := range levels { + if _, err := s.Append(ctx, Record{ + Trigger: "iot.device_update", + At: base.Add(time.Duration(i) * time.Minute), + Conclusion: "结论", + ShouldPush: i > 0, + PushResult: PushResult{Route: RouteNone, Detail: "held"}, + Urgency: lv.String(), + }); err != nil { + t.Fatalf("Append(%s): %v", lv, err) + } + } + + recs, err := s.List(ctx, 0) + if err != nil { + t.Fatalf("List: %v", err) + } + if len(recs) != len(levels) { + t.Fatalf("len = %d, want %d", len(recs), len(levels)) + } + // 最新在前：第 i 条写入的记录在 len-1-i 位上。 + for i, lv := range levels { + got := recs[len(levels)-1-i].Urgency + if got != lv.String() { + t.Fatalf("round-trip %s = %q, want %q", lv, got, lv.String()) + } + } +} + +// TestStore_Urgency_LegacyDatabaseUpgraded 覆盖验收②③：既有库（旧形状 + 一行旧记录） +// 能被补上 urgency 列、同一库再跑一次 NewStore 依旧成功、数据不丢，且**旧行读成 +// 空串「未知」而不是 routine**。 +func TestStore_Urgency_LegacyDatabaseUpgraded(t *testing.T) { + t.Parallel() + ctx := context.Background() + db := newLegacyRecordsDB(t) + if hasColumn(t, db, "urgency") { + t.Fatalf("fixture: legacy table must not have the urgency column yet") + } + + s, err := NewStore(ctx, db, discardLog()) + if err != nil { + t.Fatalf("NewStore on a legacy database: %v", err) + } + if !hasColumn(t, db, "urgency") { + t.Fatalf("NewStore did not add backthink_records.urgency") + } + // 幂等：同一库再来一次（既有表 + 已有列）不得报错。 + if _, err := NewStore(ctx, db, discardLog()); err != nil { + t.Fatalf("second NewStore on the same database: %v", err) + } + + recs, err := s.List(ctx, 0) + if err != nil { + t.Fatalf("List: %v", err) + } + if len(recs) != 1 { + t.Fatalf("len = %d, want 1 (旧行必须还在)", len(recs)) + } + if recs[0].Trigger != "heartbeat" || recs[0].Conclusion != "晚上好" { + t.Fatalf("legacy row = %+v, want it preserved verbatim", recs[0]) + } + if recs[0].Urgency != "" { + t.Fatalf("legacy row urgency = %q, want 空串（未知）", recs[0].Urgency) + } + if recs[0].Urgency == urgency.LevelRoutine.String() { + t.Fatalf("legacy row was back-filled as %q; 未知绝不能编成 routine", recs[0].Urgency) + } + + // 补列之后，新行带档位写入并回读，与旧行并存。 + if _, err := s.Append(ctx, Record{ + Trigger: "heartbeat", At: time.Date(2026, 10, 5, 12, 0, 0, 0, time.UTC), + Conclusion: "新行", Urgency: urgency.LevelCritical.String(), + }); err != nil { + t.Fatalf("Append after upgrade: %v", err) + } + recs, err = s.List(ctx, 0) + if err != nil { + t.Fatalf("List after upgrade: %v", err) + } + if len(recs) != 2 { + t.Fatalf("len = %d, want 2", len(recs)) + } + if recs[0].Urgency != urgency.LevelCritical.String() { + t.Fatalf("new row urgency = %q, want %q", recs[0].Urgency, urgency.LevelCritical.String()) + } + if recs[1].Urgency != "" { + t.Fatalf("legacy row urgency after a new append = %q, want 空串", recs[1].Urgency) + } +} + +// TestEnsureRecordsUrgencyColumn_OnlyToleratesDuplicateColumn 守住幂等守卫的判据本身： +// 「列已存在」这条错误被容忍（那是一次成功重放），**其它错误一律不被容忍**——吞掉 +// 真错误会让缺列静默过关，随后每条 INSERT 都失败。 +func TestEnsureRecordsUrgencyColumn_OnlyToleratesDuplicateColumn(t *testing.T) { + t.Parallel() + ctx := context.Background() + db := newTestDB(t) + if _, err := NewStore(ctx, db, discardLog()); err != nil { + t.Fatalf("NewStore: %v", err) + } + + // 列已在：再执行一次加列语句必须失败，且措辞被判为「重复列」。 + _, dup := db.ExecContext(ctx, alterRecordsAddUrgencySQL) + if dup == nil { + t.Fatalf("adding an existing column succeeded; 幂等守卫的判据不再成立") + } + if !isDuplicateColumnErr(dup) { + t.Fatalf("second ALTER error = %v, want duplicate-column classification", dup) + } + // 负向对照：表不存在这类错误绝不能被当成「重复列」。 + _, other := db.ExecContext(ctx, `ALTER TABLE backthink_no_such_table ADD COLUMN urgency TEXT NOT NULL DEFAULT ''`) + if other == nil { + t.Fatalf("ALTER on a missing table succeeded unexpectedly") + } + if isDuplicateColumnErr(other) { + t.Fatalf("missing-table error %v misclassified as a duplicate column", other) + } + // 列已在时 ensure 本身是空操作（不报错、不重复 ALTER）。 + if err := ensureRecordsUrgencyColumn(ctx, db, discardLog()); err != nil { + t.Fatalf("ensure on an upgraded database: %v", err) + } +} + +// TestStore_Urgency_LegacyFileDatabaseUpgraded 用**真文件库**（生产部署的形态：WAL + +// 单写连接 + 跨进程重开）再走一遍旧库补列：主人手上的 data/cyrene.db 正是这个形态， +// 内存库覆盖不到它（WAL、DDL 落盘、关掉再开）。 +func TestStore_Urgency_LegacyFileDatabaseUpgraded(t *testing.T) { + t.Parallel() + ctx := context.Background() + path := filepath.Join(t.TempDir(), "cyrene.db") + + // 第一次「进程」：建旧形状的表并落一行旧记录。 + first, err := db.Open(ctx, path, nil) + if err != nil { + t.Fatalf("open file db: %v", err) + } + if _, err := first.ExecContext(ctx, legacyRecordsDDL); err != nil { + t.Fatalf("create legacy table: %v", err) + } + if _, err := first.ExecContext(ctx, insertLegacyRecordSQL, + "heartbeat", "2026-09-06T10:00:00Z", "晚上好", boolInt(false), boolInt(false), + string(RouteNone), "held", "", + ); err != nil { + t.Fatalf("insert legacy row: %v", err) + } + if err := first.Close(); err != nil { + t.Fatalf("close first handle: %v", err) + } + + // 第二次「进程」：新版本起来，必须自己把列补上。 + second, err := db.Open(ctx, path, nil) + if err != nil { + t.Fatalf("reopen file db: %v", err) + } + t.Cleanup(func() { _ = second.Close() }) + s, err := NewStore(ctx, second, discardLog()) + if err != nil { + t.Fatalf("NewStore on the upgraded file database: %v", err) + } + if !hasColumn(t, second, "urgency") { + t.Fatalf("NewStore did not add backthink_records.urgency to a file database") + } + // 同一库再来一次：幂等。 + if _, err := NewStore(ctx, second, discardLog()); err != nil { + t.Fatalf("second NewStore on the same file database: %v", err) + } + recs, err := s.List(ctx, 0) + if err != nil { + t.Fatalf("List: %v", err) + } + if len(recs) != 1 || recs[0].Conclusion != "晚上好" || recs[0].Urgency != "" { + t.Fatalf("legacy file row = %+v, want it preserved with an empty (unknown) urgency", recs) + } + + // 补列之后新行可写可读。 + if _, err := s.Append(ctx, Record{ + Trigger: "heartbeat", At: time.Date(2026, 10, 5, 12, 0, 0, 0, time.UTC), + Conclusion: "新行", Urgency: urgency.LevelWarning.String(), + }); err != nil { + t.Fatalf("Append after file upgrade: %v", err) + } + recs, err = s.List(ctx, 0) + if err != nil { + t.Fatalf("List after file upgrade: %v", err) + } + if len(recs) != 2 || recs[0].Urgency != urgency.LevelWarning.String() { + t.Fatalf("records after file upgrade = %+v, want the new round carrying warning", recs) + } +} + +// TestEnsureRecordsUrgencyColumn_ReportsRealFailures 守住「不许吞掉加列错误」：列**真的 +// 加不上**时（这里让 backthink_records 恰好是一个视图 ⇒ ALTER 必失败），ensure 必须把 +// 错误报出来，而不是假装成功——假装成功的后果是启动日志干净、之后每条 INSERT 都失败。 +func TestEnsureRecordsUrgencyColumn_ReportsRealFailures(t *testing.T) { + t.Parallel() + ctx := context.Background() + db := newTestDB(t) + if _, err := db.ExecContext(ctx, `CREATE VIEW backthink_records AS SELECT 1 AS id`); err != nil { + t.Fatalf("create view fixture: %v", err) + } + if hasColumn(t, db, "urgency") { + t.Fatalf("fixture: the view must not expose an urgency column") + } + err := ensureRecordsUrgencyColumn(ctx, db, discardLog()) + if err == nil { + t.Fatalf("ensure reported success although the column cannot be added") + } + if isDuplicateColumnErr(err) { + t.Fatalf("view-alter error %v misclassified as a duplicate column", err) + } +} + +// TestStore_Urgency_InsertAndSelectNameTheColumn 是一处结构性守卫：INSERT 与 SELECT +// 的列清单必须都含 urgency（列表顺序由列名对齐，靠扫描位置对不上就会读到错列的错—— +// 这里用「同一条记录两次读取逐字段相等」把两个语句绑在一起）。 +func TestStore_Urgency_InsertAndSelectNameTheColumn(t *testing.T) { + t.Parallel() + if !strings.Contains(insertRecordSQL, "urgency") { + t.Fatalf("insertRecordSQL 未列出 urgency 列：\n%s", insertRecordSQL) + } + if !strings.Contains(selectRecordsSQL, "urgency") { + t.Fatalf("selectRecordsSQL 未取 urgency 列：\n%s", selectRecordsSQL) + } + if !strings.Contains(createRecordsSQL, "urgency") { + t.Fatalf("createRecordsSQL 未含 urgency 列：\n%s", createRecordsSQL) + } +} + +// TestRecord_UrgencyJSONKeyIsUrgency 覆盖验收⑤：序列化后的键名**就是** `urgency` +// （WS 实时帧发的是同一个结构体，前端按 `r.urgency` 读），且「未知」保持键缺席， +// 好与前端的「例常」区分开。 +func TestRecord_UrgencyJSONKeyIsUrgency(t *testing.T) { + t.Parallel() + raw, err := json.Marshal(Record{ + ID: 7, Trigger: "heartbeat", At: time.Date(2026, 10, 5, 12, 0, 0, 0, time.UTC), + Conclusion: "结论", Urgency: urgency.LevelCritical.String(), + }) + if err != nil { + t.Fatalf("Marshal: %v", err) + } + if !strings.Contains(string(raw), `"urgency":"critical"`) { + t.Fatalf("marshalled record = %s, want the `urgency` key", raw) + } + if strings.Contains(string(raw), `"Urgency"`) { + t.Fatalf("marshalled record = %s, want ONLY the lowercase `urgency` key", raw) + } + + unknown, err := json.Marshal(Record{ID: 8, Trigger: "heartbeat"}) + if err != nil { + t.Fatalf("Marshal: %v", err) + } + if strings.Contains(string(unknown), "urgency") { + t.Fatalf("empty urgency must omit the key entirely, got %s", unknown) + } +} diff --git a/internal/services/backthink/trigger.go b/internal/services/backthink/trigger.go index 77b28d4..c4e2376 100644 --- a/internal/services/backthink/trigger.go +++ b/internal/services/backthink/trigger.go @@ -148,6 +148,28 @@ func importanceForUrgency(l urgency.Level) Importance { } } +// urgencyForImportance 是 importanceForUrgency 的**反向读法**：把一轮思考**实际生效** +// 的重要度还原成通用紧急度词汇，供存档携带精确档位（X2a；面板此前只能给出「下限」）。 +// +// 它复用同一张既有对应关系（Critical↔critical、High↔warning），不引入任何新的门限或 +// 折算规则：例行档对应 importanceForUrgency 的 Normal，而 ImportanceLow 只是「比例行 +// 还安静」的心跳/启动轮 —— urgency 词汇里没有更低的档，故与 Normal 同归 routine +// （无损：routine 本就表示「安静时段照常扣下」这一类）。按档位高低比较而不是逐值 +// 相等，是让将来插入更高档时它落在最高档、而不是静默掉回 routine。 +// +// 取数点在 backthink.go 的 think：入参是该轮**实际生效**的重要度 +// （escalatedImport 的结果，她自报的紧急度已折叠在内）。 +func urgencyForImportance(i Importance) urgency.Level { + switch { + case i >= ImportanceCritical: + return urgency.LevelCritical + case i >= ImportanceHigh: + return urgency.LevelWarning + default: + return urgency.LevelRoutine + } +} + // escalateImport 把声明的紧急度叠到基础重要度上。声明只能**上抬**、绝不压低，因此 // 主题级默认（如提醒的 High）与安全域事件既得的档位不会被一句 routine 抹掉； // 未声明（零值 LevelRoutine）时原样返回 base，没标紧急度的事件行为逐字不变。 diff --git a/internal/services/backthink/urgency_record_test.go b/internal/services/backthink/urgency_record_test.go new file mode 100644 index 0000000..d03e560 --- /dev/null +++ b/internal/services/backthink/urgency_record_test.go @@ -0,0 +1,272 @@ +package backthink + +import ( + "context" + "encoding/json" + "errors" + "strings" + "testing" + "time" + + "git.yeij.top/AskaEth/Cyrene_v2/internal/bus" + "git.yeij.top/AskaEth/Cyrene_v2/internal/iot" + "git.yeij.top/AskaEth/Cyrene_v2/internal/urgency" +) + +// 本文件守住 X2a 的**生产面**：存档里的 urgency 必须是**该轮实际生效**的档位 +// （取数点 backthink.go 的 think：roundImport = escalatedImport(tr.Import, 她自报)）， +// 逐档给出不同期望值——写死常量或漏接线在这里即变红；没跑完/没跑的轮次如实为未知。 + +// TestBackthink_RecordUrgency_EffectiveLevelPerRound 覆盖验收④：事件自带档位、 +// 她自报的上抬、以主题默认档入场的提醒，各落各的档，四档互不塌缩。 +func TestBackthink_RecordUrgency_EffectiveLevelPerRound(t *testing.T) { + t.Parallel() + routineDevice := iot.DeviceUpdate{Device: iot.Device{ID: "curtain", Type: "cover"}, Action: "opened"} + warnDevice := iot.DeviceUpdate{Device: iot.Device{ID: "freezer", Type: "sensor"}, Action: "offline", Severity: iot.SeverityWarning} + critDevice := iot.DeviceUpdate{Device: iot.Device{ID: "door", Type: "lock"}, Action: "opened", Severity: iot.SeverityCritical} + + cases := []struct { + name string + tr Trigger + conclusion string + want urgency.Level + }{ + { + name: "心跳_未自报_例行档", + tr: heartbeatTrigger(), + conclusion: "[我决定:push]\n我在呢，你忙你的", + want: urgency.LevelRoutine, + }, + { + name: "设备跃迁_主题默认档_例行", + tr: triggerForEvent(bus.Event{Subject: iot.DeviceUpdateSubject, Payload: routineDevice}), + conclusion: "[我决定:push]\n窗帘开了", + want: urgency.LevelRoutine, + }, + { + name: "设备跃迁_警告档", + tr: triggerForEvent(bus.Event{Subject: iot.DeviceUpdateSubject, Payload: warnDevice}), + conclusion: "[我决定:push]\n冷柜掉线了", + want: urgency.LevelWarning, + }, + { + name: "设备跃迁_安全域档", + tr: triggerForEvent(bus.Event{Subject: iot.DeviceUpdateSubject, Payload: critDevice}), + conclusion: "[我决定:push]\n门开了", + want: urgency.LevelCritical, + }, + { + name: "提醒_主题自带高档", + tr: triggerForEvent(bus.Event{Subject: SubjectReminderTrigger, Payload: "reminder due"}), + conclusion: "[我决定:push]\n该吃药了", + want: urgency.LevelWarning, + }, + { + name: "心跳_她自报warning_上抬", + tr: heartbeatTrigger(), + conclusion: "[我决定:push]\n[紧急:warning]\n饮水机快没水了", + want: urgency.LevelWarning, + }, + { + name: "心跳_她自报critical_上抬", + tr: heartbeatTrigger(), + conclusion: "[我决定:push]\n[紧急:critical]\n院里有人影", + want: urgency.LevelCritical, + }, + } + for _, tc := range cases { + tc := tc + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + ctx := context.Background() + b, _ := newBackthink(t, Config{}, Deps{ + Thinker: &mockThinker{text: tc.conclusion}, + State: &mockState{state: UserState{}}, + Pusher: &mockPusher{result: PushResult{Pushed: true, Route: RouteActiveClient}}, + Now: func() time.Time { return dayAt() }, + }) + b.think(ctx, tc.tr) + + recs, err := b.List(ctx, 1) + if err != nil { + t.Fatalf("List: %v", err) + } + if len(recs) != 1 { + t.Fatalf("len = %d, want 1", len(recs)) + } + if recs[0].Urgency != tc.want.String() { + t.Fatalf("record urgency = %q, want %q (trigger import = %s, conclusion = %q)", + recs[0].Urgency, tc.want.String(), tc.tr.Import, tc.conclusion) + } + }) + } +} + +// TestBackthink_RecordUrgency_ReminderHighBeatsLowerSelfReport 钉住「自报只能上抬」的 +// 那一半在存档里的表现：她在一轮高档事件里声明更低的档位，**生效档位仍是事件那一档** +// ——存档落的是生效值，不是她那句话。 +func TestBackthink_RecordUrgency_ReminderHighBeatsLowerSelfReport(t *testing.T) { + t.Parallel() + ctx := context.Background() + tr := triggerForEvent(bus.Event{Subject: SubjectReminderTrigger, Payload: "reminder due"}) + if tr.Import != ImportanceHigh { + t.Fatalf("fixture: reminder import = %s, want %s", tr.Import, ImportanceHigh) + } + b, _ := newBackthink(t, Config{}, Deps{ + Thinker: &mockThinker{text: "[我决定:push]\n[紧急:routine]\n该吃药了"}, + State: &mockState{state: UserState{}}, + Pusher: &mockPusher{result: PushResult{Pushed: true, Route: RouteActiveClient}}, + Now: func() time.Time { return dayAt() }, + }) + b.think(ctx, tr) + + recs, err := b.List(ctx, 1) + if err != nil { + t.Fatalf("List: %v", err) + } + if recs[0].Urgency != urgency.LevelWarning.String() { + t.Fatalf("record urgency = %q, want %q (声明的 routine 压不低事件自带的 high 档)", + recs[0].Urgency, urgency.LevelWarning.String()) + } +} + +// TestBackthink_RecordUrgency_ThinkingFailedStaysUnknown 守住 fail-safe：思考抛错的那 +// 一行**没有任何档位生效** ⇒ 如实留空（未知），不拿触发的档位冒充该轮档位。 +func TestBackthink_RecordUrgency_ThinkingFailedStaysUnknown(t *testing.T) { + t.Parallel() + ctx := context.Background() + // 触发本身的档位是 High：若实现把触发档位写进去，断言即红。 + tr := triggerForEvent(bus.Event{Subject: SubjectReminderTrigger, Payload: "reminder due"}) + if tr.Import != ImportanceHigh { + t.Fatalf("fixture: reminder import = %s, want %s", tr.Import, ImportanceHigh) + } + b, _ := newBackthink(t, Config{}, Deps{ + Thinker: &mockThinker{err: errors.New("thinker exploded")}, + State: &mockState{state: UserState{}}, + Now: func() time.Time { return dayAt() }, + }) + b.think(ctx, tr) + + recs, err := b.List(ctx, 1) + if err != nil { + t.Fatalf("List: %v", err) + } + if len(recs) != 1 { + t.Fatalf("len = %d, want 1", len(recs)) + } + if !strings.Contains(recs[0].Conclusion, "thinking failed") { + t.Fatalf("fixture: conclusion = %q, want the failure record", recs[0].Conclusion) + } + if recs[0].Urgency != "" { + t.Fatalf("失败行 urgency = %q, want 空串（未知）", recs[0].Urgency) + } +} + +// TestBackthink_RecordUrgency_DowngradedRowStaysUnknown 守住另一半 fail-safe：未唤醒 +// 思考的降级行**不是一轮思考**（结论即写明），没有「该轮生效的档位」可落，因此同样 +// 如实留空——它不是旧行，所以这条断言也说明「空白 ≠ 只在旧库里出现」。 +func TestBackthink_RecordUrgency_DowngradedRowStaysUnknown(t *testing.T) { + t.Parallel() + ctx := context.Background() + thinker := &mockThinker{text: "should not be called"} + b, _ := newBackthink(t, Config{}, Deps{ + Bus: bus.New(), + Thinker: thinker, + Now: func() time.Time { return dayAt() }, + }) + if err := b.onEvent(ctx, bus.Event{Subject: SubjectSystemEvent, Payload: SystemEvent{Summary: "member joined"}}); err != nil { + t.Fatalf("onEvent: %v", err) + } + if got := thinker.count(); got != 0 { + t.Fatalf("fixture: thinker calls = %d, want 0 (low-value system event downgraded)", got) + } + + recs, err := b.List(ctx, 1) + if err != nil { + t.Fatalf("List: %v", err) + } + if len(recs) != 1 { + t.Fatalf("len = %d, want 1", len(recs)) + } + if recs[0].Urgency != "" { + t.Fatalf("降级行 urgency = %q, want 空串（未知）", recs[0].Urgency) + } +} + +// TestBackthink_RecordUrgency_RealtimePayloadCarriesUrgency 守住「实时可见」那条链路： +// record() 把同一份 Record 发到 ws.backthink 上，ws 桥按原样成形为帧的 data，客户端 +// （wsParse.ts）按 `urgency` 读它 —— 因此帧载荷里必须既有该字段、序列化后键名也正确。 +func TestBackthink_RecordUrgency_RealtimePayloadCarriesUrgency(t *testing.T) { + t.Parallel() + ctx := context.Background() + bs := bus.New() + frames := make(chan any, 4) + if _, err := bs.Subscribe(ctx, "ws.backthink", func(_ context.Context, e bus.Event) error { + select { + case frames <- e.Payload: + default: + } + return nil + }); err != nil { + t.Fatalf("Subscribe: %v", err) + } + + b, _ := newBackthink(t, Config{}, Deps{ + Bus: bs, + Thinker: &mockThinker{text: "[我决定:push]\n门开了"}, + State: &mockState{state: UserState{}}, + Pusher: &mockPusher{result: PushResult{Pushed: true, Route: RouteActiveClient}}, + Now: func() time.Time { return dayAt() }, + }) + b.think(ctx, triggerForEvent(bus.Event{Subject: iot.DeviceUpdateSubject, Payload: iot.DeviceUpdate{ + Device: iot.Device{ID: "door", Type: "lock"}, Action: "opened", Severity: iot.SeverityCritical, + }})) + + var payload any + select { + case payload = <-frames: + case <-time.After(2 * time.Second): + t.Fatalf("no ws.backthink frame was published") + } + rec, ok := payload.(Record) + if !ok { + t.Fatalf("frame payload = %T, want backthink.Record", payload) + } + if rec.Urgency != urgency.LevelCritical.String() { + t.Fatalf("frame urgency = %q, want %q", rec.Urgency, urgency.LevelCritical.String()) + } + raw, err := json.Marshal(payload) + if err != nil { + t.Fatalf("Marshal: %v", err) + } + if !strings.Contains(string(raw), `"urgency":"critical"`) { + t.Fatalf("frame json = %s, want the `urgency` key the client reads", raw) + } +} + +// TestUrgencyForImportance_IsInverseOfImportanceForUrgency 钉住反向读法只是把既有 +// 对应关系读回来：档位高低单调（永不把 low/normal 抬成 warning），且与正向换算在 +// warning / critical 两档上互逆。 +func TestUrgencyForImportance_IsInverseOfImportanceForUrgency(t *testing.T) { + t.Parallel() + cases := []struct { + in Importance + want urgency.Level + }{ + {in: ImportanceLow, want: urgency.LevelRoutine}, + {in: ImportanceNormal, want: urgency.LevelRoutine}, + {in: ImportanceHigh, want: urgency.LevelWarning}, + {in: ImportanceCritical, want: urgency.LevelCritical}, + } + for _, tc := range cases { + if got := urgencyForImportance(tc.in); got != tc.want { + t.Fatalf("urgencyForImportance(%s) = %v, want %v", tc.in, got, tc.want) + } + } + // 正向换算回来必须逐档一致（两档互为逆），因此存档档位就是判定侧吃进去的档位。 + for _, lv := range []urgency.Level{urgency.LevelWarning, urgency.LevelCritical} { + if got := urgencyForImportance(importanceForUrgency(lv)); got != lv { + t.Fatalf("round-trip %v -> %v, want %v", lv, got, lv) + } + } +}