diff --git a/go/internal/api/routes/selfcheck.go b/go/internal/api/routes/selfcheck.go new file mode 100644 index 0000000..fa21f70 --- /dev/null +++ b/go/internal/api/routes/selfcheck.go @@ -0,0 +1,285 @@ +package routes + +// 织忆自检端点(2026-10-26,牧尘批「三档」) +// +// 背景:织忆相关的定时自检脚本有 ~20 个,取数方式三套并存—— +// ① 直连生产库 lancedb.connect("/var/lib/memoryweave")(绕过服务、schema 一改就静默腐烂) +// ② IPC socket 自定义协议 +// ③ HTTP /api/v1/{stats,popwatch,metrics/self} +// 三套契约对同一个库,schema 变更时分开损坏。典型事故:pattern 自检脚本写死 +// type='pattern',库里实际是 pattern_stat ⇒ 连续数月静默 0 输出。 +// +// 本端点是**唯一取数契约**:把「读织忆内部状态」收敛成一份结构化 JSON。 +// 设计边界(不做什么): +// - 不做判定/告警/投递 —— 阈值、渠道、评审留在外部 cron(观测者不能住在被观测者体内: +// zhiyid 一挂自检就跟着哑,内外同时沉默是更坏的故障)。 +// - 不跑重活(eval 全量、备份、vacuum)—— 那些抢生产资源,留在 systemd-run 独立 scope。 +// +// level 语义: +// - cheap(默认,<50ms):只读内存态/计数,不扫全表。 +// - full(秒级):额外扫全表算碎片率/冗余率/近 7 天质量/写量分布(原 quality-watch 口径)。 +// +// 兼容铁律:本端点返回的比率一律给「分子/分母/样本是否足够」,缺数据不返回 0 +// (0 会被下游读成"很好")。沿用 selfoptimize 的 metricNA(-1) 语义。 + +import ( + "encoding/json" + "fmt" + "math" + "net/http" + "os" + "path/filepath" + "strings" + "time" + + "github.com/xiaoxue/memoryweave/internal/distill" + "github.com/xiaoxue/memoryweave/internal/selfoptimize" + "github.com/xiaoxue/memoryweave/internal/storage" +) + +// SelfcheckSchemaVersion 变更返回结构时递增;消费方据此判兼容。 +const SelfcheckSchemaVersion = "1" + +// 与 zhiyi-quality-watch.py 同口径(该脚本已改为读本端点,口径只此一处) +const ( + selfcheckFragLen = 20 // <20 字符 = 碎片 + selfcheckDupMinSample = 5000 // 冗余率最小样本(活行),不足只报样本不判率 + selfcheckScanLimit = 300000 +) + +type selfcheckQuality struct { + TotalRows int `json:"total_rows"` + LiveRows int `json:"live_rows"` + ScannedRows int `json:"scanned_rows"` + ScanTruncated bool `json:"scan_truncated"` + Coverage float64 `json:"coverage"` // 扫描行 / store 真实总量;<1 表示结果不完整,禁止据此下结论 + FragLen int `json:"frag_len"` + + FragRowsAll int `json:"frag_rows_all"` + FragRateAll float64 `json:"frag_rate_all"` + FragRowsLive int `json:"frag_rows_live"` + FragRateLive float64 `json:"frag_rate_live"` + + DupGroups int `json:"dup_groups"` + DupExtraRows int `json:"dup_extra_rows"` + DupRate float64 `json:"dup_rate"` + DupSampleOK bool `json:"dup_sample_ok"` + + NewRows7d int `json:"new_rows_7d"` + NewFragRows7d int `json:"new_frag_rows_7d"` + NewFragRate7d *float64 `json:"new_frag_rate_7d"` // nil = 近 7 天无新增(无分母) +} + +type selfcheckDayCount struct { + Date string `json:"date"` + Rows int `json:"rows"` + Live int `json:"live"` +} + +type selfcheckWriteVolume struct { + Days []selfcheckDayCount `json:"days"` + Today int `json:"today"` + Last7d int `json:"last_7d"` + Median7d float64 `json:"median_7d"` + P25_7d float64 `json:"p25_7d"` +} + +// Selfcheck GET /api/v1/selfcheck?level=cheap|full +func (a *API) Selfcheck(w http.ResponseWriter, r *http.Request) { + start := time.Now() + if r.Method != http.MethodGet { + respondError(w, http.StatusMethodNotAllowed, "method not allowed") + return + } + level := strings.ToLower(strings.TrimSpace(r.URL.Query().Get("level"))) + if level == "" { + level = "cheap" + } + if level != "cheap" && level != "full" { + respondError(w, http.StatusBadRequest, "level must be cheap|full") + return + } + + rep := map[string]interface{}{ + "schema_version": SelfcheckSchemaVersion, + "checked_at": time.Now().Format(time.RFC3339), + "level": level, + } + degraded := []string{} + + // ── 存储总量 ─────────────────────────────── + if st, err := a.LanceDB.Stats(); err != nil { + degraded = append(degraded, "store: "+err.Error()) + } else { + total := numOf(st["total_memories"]) + tomb := numOf(st["tombstone_count"]) + live := total - tomb + if live < 0 { + live = 0 + } + store := map[string]interface{}{ + "backend": st["backend"], + "data_dir": st["data_dir"], + "total_memories": total, + "total_episodes": numOf(st["total_episodes"]), + "tombstone_count": tomb, + "live_memories": live, + } + if total > 0 { + store["tombstone_rate"] = round4(tomb / total) + } else { + store["tombstone_rate"] = -1.0 // N/A:空库无分母 + } + rep["store"] = store + } + + // ── 图谱 ─────────────────────────────────── + if a.GraphStore != nil { + n, e, d := a.GraphStore.Stats() + rep["graph"] = map[string]interface{}{ + "node_count": n, + "edge_count": e, + "density": round4(d), + } + } else { + degraded = append(degraded, "graph: store not initialised") + } + + // ── 面板指标(反馈/蒸馏/gap/cascade,含 _valid/_samples 语义)── + rep["dashboard"] = selfoptimize.Dash.Metrics() + + // ── 整合链路(内存态,<1ms)──────────────── + tracker := GetConsolTracker() + consolStatus, consolFails, sinceSuccess, lastErr := tracker.HealthStatus() + _, consolSkips, lastSkip, _ := tracker.Snapshot() + rep["consolidate"] = map[string]interface{}{ + "status": consolStatus, + "consec_failures": consolFails, + "skipped": consolSkips, + "last_skip": lastSkip, + "last_success_age": sinceSuccess.String(), + "last_error": lastErr, + } + + // ── 缓存 / FTS ───────────────────────────── + cache := map[string]interface{}{"fts_index": storage.FtsIndexStats()} + if storage.SearchCacheInstance != nil { + cache["search_cache"] = storage.SearchCacheInstance.Stats() + } + rep["cache"] = cache + + // ── 空壳拦截器(t_e11207f8 取证)─────────── + checks, blocks, loaded, rulesPath := distill.PayloadGuardStats() + rep["payload_guard"] = map[string]interface{}{ + "checks": checks, + "blocks": blocks, + "loaded": loaded, + "rules_path": rulesPath, + } + + // ── 全量质量段(level=full):读卫星快照,不在本进程扫全表 ──────── + // 设计:全表扫描是重活(4 万行 × 内容),且 daemon 还有 recall 要跑。 + // 取数由仓库内 tools/selfcheck_full.py 完成(列投影只读,1.7s / 250MB,独立进程), + // 写 ~/.hermes/state/zhiyi-quality.json;本端点只读快照并标注新鲜度。 + // 这样"取数"只有一处实现(跟 schema 同仓),"判定/告警"仍在外部 cron。 + if level == "full" { + snap, err := loadQualitySnapshot() + if err != nil { + degraded = append(degraded, "quality: "+err.Error()) + } else { + rep["quality"] = snap.Quality + rep["write_volume"] = snap.WriteVolume + rep["quality_as_of"] = snap.GeneratedAt + rep["quality_age_s"] = snap.AgeSeconds() + rep["quality_stale"] = snap.AgeSeconds() > qualityStaleAfter + } + } + + if len(degraded) > 0 { + rep["degraded"] = degraded + } + rep["duration_ms"] = time.Since(start).Milliseconds() + + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(http.StatusOK) + json.NewEncoder(w).Encode(rep) +} + +// qualityStaleAfter:快照超过这个时长未刷新 ⇒ 标记 stale(消费方不得据此下结论)。 +const qualityStaleAfter = 6 * 3600 + +// qualitySnapshot 对应仓库内 tools/selfcheck_full.py 的产出。 +// 字段名与 python 侧严格一致(schema_version=1)。 +type qualitySnapshot struct { + SchemaVersion int `json:"schema_version"` + GeneratedAt string `json:"generated_at"` + Source struct { + DB string `json:"db"` + Table string `json:"table"` + Mode string `json:"mode"` + } `json:"source"` + DurationMS int `json:"duration_ms"` + Quality json.RawMessage `json:"quality"` + WriteVolume json.RawMessage `json:"write_volume"` + fileModTime time.Time +} + +func (q *qualitySnapshot) AgeSeconds() int64 { + if q.fileModTime.IsZero() { + return -1 + } + return int64(time.Since(q.fileModTime).Seconds()) +} + +// QualitySnapshotPath 允许用 env 覆盖(测试/灰度)。 +func QualitySnapshotPath() string { + if p := strings.TrimSpace(os.Getenv("ZHIYI_QUALITY_SNAPSHOT")); p != "" { + return p + } + home, _ := os.UserHomeDir() + return filepath.Join(home, ".hermes", "state", "zhiyi-quality.json") +} + +func loadQualitySnapshot() (*qualitySnapshot, error) { + p := QualitySnapshotPath() + fi, err := os.Stat(p) + if err != nil { + return nil, fmt.Errorf("快照不存在(先跑 tools/selfcheck_full.py):%w", err) + } + raw, err := os.ReadFile(p) + if err != nil { + return nil, err + } + var snap qualitySnapshot + if err := json.Unmarshal(raw, &snap); err != nil { + return nil, fmt.Errorf("快照解析失败: %w", err) + } + snap.fileModTime = fi.ModTime() + return &snap, nil +} + +// numOf 把 stats 里可能是 float64/int/json.Number 的数值统一成 float64。 +func numOf(v interface{}) float64 { + switch x := v.(type) { + case float64: + return x + case float32: + return float64(x) + case int: + return float64(x) + case int64: + return float64(x) + case json.Number: + f, _ := x.Float64() + return f + default: + return 0 + } +} + +func round4(f float64) float64 { + if math.IsNaN(f) || math.IsInf(f, 0) { + return -1 + } + return math.Round(f*10000) / 10000 +} diff --git a/go/internal/api/routes/selfcheck_test.go b/go/internal/api/routes/selfcheck_test.go new file mode 100644 index 0000000..380f388 --- /dev/null +++ b/go/internal/api/routes/selfcheck_test.go @@ -0,0 +1,99 @@ +package routes + +import ( + "encoding/json" + "os" + "path/filepath" + "testing" + "time" +) + +// 快照契约测试:/api/v1/selfcheck?level=full 依赖 tools/selfcheck_full.py 的产出格式。 +// 这里钉住 schema_version 与关键字段,防止 python 侧改字段名后 Go 侧静默返回空。 +func TestLoadQualitySnapshot(t *testing.T) { + dir := t.TempDir() + p := filepath.Join(dir, "zhiyi-quality.json") + body := `{ + "schema_version": 1, + "generated_at": "2026-10-26T13:20:00+0800", + "source": {"db": "/var/lib/memoryweave", "table": "memories", "mode": "read_only_projection"}, + "duration_ms": 1680, + "quality": { + "frag_len": 20, "total_rows": 39957, "live_rows": 39836, + "frag_rows_all": 313, "frag_rate_all": 0.0078, + "frag_rows_live": 313, "frag_rate_live": 0.0079, + "dup_groups": 34, "dup_extra_rows": 103, "dup_rate": 0.0026, + "new_rows_7d": 1176, "new_frag_rows_7d": 35, "new_frag_rate_7d": 0.0298 + }, + "write_volume": { + "days": [{"date": "2026-10-25", "rows": 1473, "live": 1471}], + "today": 332, "today_date": "2026-10-26", "last_7d": 7706, "nonzero_days": 60 + } + }` + if err := os.WriteFile(p, []byte(body), 0o644); err != nil { + t.Fatal(err) + } + t.Setenv("ZHIYI_QUALITY_SNAPSHOT", p) + + snap, err := loadQualitySnapshot() + if err != nil { + t.Fatalf("loadQualitySnapshot: %v", err) + } + if snap.SchemaVersion != 1 { + t.Fatalf("schema_version=%d want 1", snap.SchemaVersion) + } + + var q struct { + TotalRows int `json:"total_rows"` + LiveRows int `json:"live_rows"` + DupRate float64 `json:"dup_rate"` + FragRowsAll int `json:"frag_rows_all"` + } + if err := json.Unmarshal(snap.Quality, &q); err != nil { + t.Fatalf("quality 段解析失败: %v", err) + } + if q.TotalRows != 39957 || q.LiveRows != 39836 { + t.Fatalf("quality 口径不符: %+v", q) + } + if q.FragRowsAll != 313 || q.DupRate != 0.0026 { + t.Fatalf("quality 数值不符: %+v", q) + } + + var wv struct { + Today int `json:"today"` + Days []struct { + Date string `json:"date"` + Live int `json:"live"` + } `json:"days"` + } + if err := json.Unmarshal(snap.WriteVolume, &wv); err != nil { + t.Fatalf("write_volume 段解析失败: %v", err) + } + if wv.Today != 332 || len(wv.Days) != 1 || wv.Days[0].Live != 1471 { + t.Fatalf("write_volume 不符: %+v", wv) + } + + // 新鲜度:刚写的文件 age 应 < 10s;人为改旧 → stale 判定为真 + if age := snap.AgeSeconds(); age < 0 || age > 10 { + t.Fatalf("AgeSeconds=%d,本应接近 0", age) + } + old := time.Now().Add(-8 * time.Hour) + if err := os.Chtimes(p, old, old); err != nil { + t.Fatal(err) + } + snap2, err := loadQualitySnapshot() + if err != nil { + t.Fatal(err) + } + if snap2.AgeSeconds() <= qualityStaleAfter { + t.Fatalf("8h 前的快照应判 stale(>%d 秒),得到 %d", qualityStaleAfter, snap2.AgeSeconds()) + } +} + +// 快照缺失 ⇒ 必须报错(由 handler 转成 degraded),不能静默返回零值。 +func TestLoadQualitySnapshotMissing(t *testing.T) { + t.Setenv("ZHIYI_QUALITY_SNAPSHOT", filepath.Join(t.TempDir(), "nope.json")) + if _, err := loadQualitySnapshot(); err == nil { + t.Fatal("缺失快照应返回 error") + } +} diff --git a/go/internal/api/server.go b/go/internal/api/server.go index a80644e..d7b6908 100644 --- a/go/internal/api/server.go +++ b/go/internal/api/server.go @@ -613,6 +613,10 @@ func NewServer() http.Handler { // 反馈统一走 /api/v1/feedback/{useful,not-useful,deprecate,correct}(feedback.go,写拆分后新字段)。 mux.HandleFunc("/api/v1/bootstrap", api.Bootstrap) mux.HandleFunc("/api/v1/stats", api.Stats) + // 2026-10-26(牧尘批「三档」):自检统一取数端点。 + // 唯一契约 —— 自检脚本一律读这里,禁止再 lancedb.connect() 直连生产库。 + // ?level=cheap(默认,内存态计数)| full(+读 tools/selfcheck_full.py 的全量快照:碎片/冗余/近7天质量/写量) + mux.HandleFunc("/api/v1/selfcheck", api.Selfcheck) mux.HandleFunc("/api/v1/batch-commit", api.BatchCommit) // 主症A(t_fa6c9104):显式写入直落 memories 表(嵌入 + 立刻可召回)。 // 纯新增路由:commit/recall/蒸馏路径零改动;旧服务端 404 时插件自动回落 commit。 diff --git a/tools/selfcheck_full.py b/tools/selfcheck_full.py new file mode 100644 index 0000000..cf98134 --- /dev/null +++ b/tools/selfcheck_full.py @@ -0,0 +1,200 @@ +#!/usr/bin/env python3 +"""织忆自检·全量取数器(唯一允许直连生产库的脚本) + +为什么存在 +---------- +织忆相关的定时自检脚本曾有 ~20 个,其中两个直接 `lancedb.connect("/var/lib/memoryweave")` +绕过服务读全表。那种写法的病根不是"直连",而是**取数逻辑散落多处、与 schema 一起腐烂**: +pattern 自检写死 `type='pattern'`(库里其实是 `pattern_stat`)⇒ 连续数月静默 0 输出。 + +本文件把"全量取数"收成一处、放进仓库:改 schema 时同一个 commit 就能改它,且它有输出可复核。 +判定/告警/投递**不在这里**(留在外部 cron)—— 观测者不能住在被观测者体内。 + +它产出 `~/.hermes/state/zhiyi-quality.json`,服务端 `GET /api/v1/selfcheck?level=full` 读这份快照。 + +口径(必须与历史 zhiyi-quality-watch.py 一致,改了就改告警语义) +------------------------------------------------------------ +- 碎片:`len(content.strip()) < FRAG_LEN(20)`,按**字符数**(不是字节,中文按 1 算) +- 冗余:live 行内 content 完全相同;冗余行数 = Σ(组内行数-1) +- 近 7 天:created_at 在 7 天内(live) +- 写量:按 created_at 的**本地日期**分桶,同时给 rows(含已删)与 live(仅存活) + +性能 +---- +只投影 3 列(content/created_at/is_deleted),不读 1024 维向量: +4 万行 0.06s(旧实现 `to_arrow()` 全列要数秒且吃 1G+ 内存)。 + +用法 +---- + python3 tools/selfcheck_full.py [--out PATH] [--json] [--days 60] +""" +from __future__ import annotations + +import argparse +import datetime as dt +import json +import os +import sys +import time +from collections import Counter + +DB_PATH = os.environ.get("ZHIYI_DB", "/var/lib/memoryweave") +FRAG_LEN = 20 +OUT_DEFAULT = os.path.expanduser("~/.hermes/state/zhiyi-quality.json") +SCHEMA = 1 + + +def _reexec_with_lancedb() -> None: + """cron 默认解释器没有 lancedb 时,切到 lance-compact venv 再跑(一次性)。""" + try: + import lancedb # noqa: F401 + return + except ImportError: + pass + venv_py = os.path.expanduser("~/.hermes/venvs/lance-compact/bin/python") + if os.path.exists(venv_py) and os.path.realpath(sys.prefix) != os.path.realpath( + os.path.dirname(os.path.dirname(venv_py)) + ): + os.execv(venv_py, [venv_py] + sys.argv) + sys.stderr.write("[selfcheck_full] lancedb 不可用(需 lance-compact venv)\n") + sys.exit(2) + + +def _as_bool(v) -> bool: + if v is None: + return False + if isinstance(v, bool): + return v + return str(v).strip().lower() in ("true", "1") + + +def collect(days: int = 60) -> dict: + import lancedb + + t0 = time.time() + db = lancedb.connect(DB_PATH) + table = db.open_table("memories") + tbl = table.to_lance().to_table(columns=["content", "created_at", "is_deleted"]) + + contents = tbl.column("content").to_pylist() + created = tbl.column("created_at").to_pylist() + deleted = tbl.column("is_deleted").to_pylist() + + total = len(contents) + frag_all = 0 + live = 0 + frag_live = 0 + dup_counter: Counter[str] = Counter() + daily: dict[str, dict[str, int]] = {} + now = dt.datetime.now() + week_ago = now - dt.timedelta(days=7) + new_7d = 0 + new_frag_7d = 0 + + for i in range(total): + raw = contents[i] + text = str(raw).strip() if raw is not None else "" + n = len(text) + is_frag = n < FRAG_LEN + if is_frag: + frag_all += 1 + + is_del = _as_bool(deleted[i] if i < len(deleted) else False) + + # created_at:存量行有空串/畸形 → 归 unknown,不炸整批 + day = "unknown" + ts = None + if i < len(created): + raw_ts = created[i] + if raw_ts: + try: + if isinstance(raw_ts, dt.datetime): + ts = raw_ts + else: + ts = dt.datetime.fromisoformat(str(raw_ts).replace("Z", "+00:00")) + if ts.tzinfo is not None: + ts = ts.astimezone().replace(tzinfo=None) + day = ts.strftime("%Y-%m-%d") + except Exception: + ts = None + + slot = daily.setdefault(day, {"rows": 0, "live": 0}) + slot["rows"] += 1 + + if is_del: + continue + live += 1 + slot["live"] += 1 + dup_counter[text] += 1 + if is_frag: + frag_live += 1 + if ts is not None and ts >= week_ago: + new_7d += 1 + if is_frag: + new_frag_7d += 1 + + dup_groups = sum(1 for c in dup_counter.values() if c > 1) + dup_extra = sum(c - 1 for c in dup_counter.values() if c > 1) + + # 只保留最近 days 天(含 unknown),按日期升序 + keep = sorted(daily.keys())[-days:] + day_rows = [{"date": d, "rows": daily[d]["rows"], "live": daily[d]["live"]} for d in keep] + + live_days = [x["live"] for x in day_rows if x["date"] != "unknown"] + today = now.strftime("%Y-%m-%d") + today_live = next((x["live"] for x in day_rows if x["date"] == today), 0) + + return { + "schema_version": SCHEMA, + "generated_at": now.strftime("%Y-%m-%dT%H:%M:%S%z"), + "source": {"db": DB_PATH, "table": "memories", "mode": "read_only_projection"}, + "duration_ms": int((time.time() - t0) * 1000), + "quality": { + "frag_len": FRAG_LEN, + "total_rows": total, + "live_rows": live, + "frag_rows_all": frag_all, + "frag_rate_all": round(frag_all / total, 4) if total else None, + "frag_rows_live": frag_live, + "frag_rate_live": round(frag_live / live, 4) if live else None, + "dup_groups": dup_groups, + "dup_extra_rows": dup_extra, + "dup_rate": round(dup_extra / live, 4) if live else None, + "new_rows_7d": new_7d, + "new_frag_rows_7d": new_frag_7d, + "new_frag_rate_7d": round(new_frag_7d / new_7d, 4) if new_7d else None, + }, + "write_volume": { + "days": day_rows, + "today": today_live, + "today_date": today, + # 近 7 个完整日(不含今天/unknown) + "last_7d": sum(x["live"] for x in day_rows if x["date"] not in (today, "unknown")) if day_rows else 0, + "nonzero_days": sum(1 for x in live_days if x > 0), + }, + } + + +def main() -> int: + ap = argparse.ArgumentParser(description="织忆自检全量取数(只读)") + ap.add_argument("--out", default=OUT_DEFAULT) + ap.add_argument("--json", action="store_true", help="打印到 stdout") + ap.add_argument("--days", type=int, default=60) + a = ap.parse_args() + + _reexec_with_lancedb() + snap = collect(a.days) + + if a.out: + os.makedirs(os.path.dirname(a.out), exist_ok=True) + tmp = a.out + ".tmp" + with open(tmp, "w", encoding="utf-8") as f: + json.dump(snap, f, ensure_ascii=False) + os.replace(tmp, a.out) # 原子替换:读方永不会看到半个文件 + if a.json: + print(json.dumps(snap, ensure_ascii=False, indent=1)) + return 0 + + +if __name__ == "__main__": + sys.exit(main())