mirror of
https://gitcode.com/JianFeeeee/HomeAgent.git
synced 2026-09-21 09:28:14 +00:00
Phase 0.1/1/3/6: 核心生产问题修复
Healthcheck 隔离 (Phase 0.1): - 新增 internal/sdk/selftest.go: VirtualInstance 完全隔离自检空间 - PluginSDK.Selftest()/SelftestReset() 暴露隔离实例 (含 mutex) - LLM 自检只读白名单 isSafeReadonlyTool 防写类工具污染生产 - 单测验证: healthcheck 后生产实例内容不变 + 无残留 - 存量清理: 删除 gotest/luatest 残留目录 GraphDB 去重 (Phase 1): - migrateRelationUnique: 启动自动重建 relations 表加 UNIQUE 约束并去重 - Commit 改为存在性检查, 重复三元组仅刷新 confidence 不重复插入 - 3 个 dedup 单测全绿 配置时长解析 (Phase 3): - parseDurationExtended 支持 2d/1w/3h 等人类可读单位 - GetDuration 全局生效, 防 2d 静默回退 30m Agentcli 通知风暴治理 (Phase 6): - 语义通知: 累积 notify_bytes(2KB) 或间隔 notify_interval(2s) 触发 - 生命周期即时通知: 启动/进程退出/EOF 立即通知 - 可配置 settings, 保留通知机制保证 agent 感知终端存在 - 运维止血: 已杀掉幽灵 PID 3716282 (bash git sparse clone 运行 16h) Plan.md: 新增设计意图备忘(插件即App/分层记忆), 更新各 Phase 进度
This commit is contained in:
root
@ -3,6 +3,7 @@ package memory
|
||||
import (
|
||||
"database/sql"
|
||||
"fmt"
|
||||
"strings"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
@ -103,7 +104,8 @@ func (g *GraphDB) initSchema() error {
|
||||
date_bucket TEXT,
|
||||
sentence_id INTEGER DEFAULT 0,
|
||||
FOREIGN KEY (source_id) REFERENCES entities(id),
|
||||
FOREIGN KEY (target_id) REFERENCES entities(id)
|
||||
FOREIGN KEY (target_id) REFERENCES entities(id),
|
||||
UNIQUE(source_id, target_id, relation_type, session_id)
|
||||
)`,
|
||||
`CREATE INDEX IF NOT EXISTS idx_entity_name ON entities(name)`,
|
||||
`CREATE INDEX IF NOT EXISTS idx_entity_type ON entities(type)`,
|
||||
@ -132,9 +134,64 @@ func (g *GraphDB) initSchema() error {
|
||||
// sentence_id 索引在迁移后创建，避免旧表缺少该列时失败
|
||||
tx.Exec(`CREATE INDEX IF NOT EXISTS idx_relation_sentence ON relations(sentence_id)`)
|
||||
|
||||
// 迁移4：为旧版 relations 表（无复合唯一约束）重建表以去重。
|
||||
// 旧表由 2026-07 之前的版本创建，缺少 UNIQUE(source_id, target_id, relation_type, session_id)，
|
||||
// 生产库累积了海量重复关系。这里检查 sqlite_master 中已建表的 DDL，
|
||||
// 若不含该约束则走"新建带约束表 → INSERT OR IGNORE 拷贝去重 → 换名"的官方 12 步迁移。
|
||||
if err := g.migrateRelationUnique(tx); err != nil {
|
||||
return fmt.Errorf("migrate relations unique: %w", err)
|
||||
}
|
||||
|
||||
return tx.Commit()
|
||||
}
|
||||
|
||||
// migrateRelationUnique 检测 relations 表是否带复合唯一约束，缺失则重建去重。
|
||||
// 必须在 initSchema 的同一个事务内调用（外键/索引均已存在时需先禁用外键再换名）。
|
||||
func (g *GraphDB) migrateRelationUnique(tx *sql.Tx) error {
|
||||
var ddl string
|
||||
err := tx.QueryRow(`SELECT sql FROM sqlite_master WHERE type = 'table' AND name = 'relations'`).Scan(&ddl)
|
||||
if err != nil {
|
||||
if err == sql.ErrNoRows {
|
||||
return nil // 表都不存在，无从迁移
|
||||
}
|
||||
return err
|
||||
}
|
||||
if strings.Contains(ddl, "UNIQUE") {
|
||||
return nil // 已是新 schema
|
||||
}
|
||||
|
||||
stmt := []string{
|
||||
`ALTER TABLE relations RENAME TO relations_old`,
|
||||
`CREATE TABLE relations (
|
||||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||||
source_id INTEGER NOT NULL,
|
||||
target_id INTEGER NOT NULL,
|
||||
relation_type TEXT NOT NULL,
|
||||
confidence REAL DEFAULT 1.0,
|
||||
status TEXT DEFAULT 'active',
|
||||
session_id TEXT,
|
||||
turn_id INTEGER DEFAULT 0,
|
||||
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
|
||||
updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
|
||||
date_bucket TEXT,
|
||||
sentence_id INTEGER DEFAULT 0,
|
||||
sentence_ref TEXT DEFAULT '',
|
||||
FOREIGN KEY (source_id) REFERENCES entities(id),
|
||||
FOREIGN KEY (target_id) REFERENCES entities(id),
|
||||
UNIQUE(source_id, target_id, relation_type, session_id)
|
||||
)`,
|
||||
`INSERT OR IGNORE INTO relations (id, source_id, target_id, relation_type, confidence, status, session_id, turn_id, created_at, updated_at, date_bucket, sentence_id, sentence_ref)
|
||||
SELECT id, source_id, target_id, relation_type, confidence, status, session_id, turn_id, created_at, updated_at, date_bucket, sentence_id, sentence_ref FROM relations_old`,
|
||||
`DROP TABLE relations_old`,
|
||||
}
|
||||
for _, s := range stmt {
|
||||
if _, err := tx.Exec(s); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (g *GraphDB) Commit(triples []Triple, sessionID string, turnID int) (int, int, error) {
|
||||
g.mu.Lock()
|
||||
defer g.mu.Unlock()
|
||||
@ -206,15 +263,34 @@ func (g *GraphDB) Commit(triples []Triple, sessionID string, turnID int) (int, i
|
||||
}
|
||||
}
|
||||
|
||||
_, err = tx.Exec(
|
||||
`INSERT INTO relations (source_id, target_id, relation_type, confidence, session_id, turn_id, date_bucket, sentence_id)
|
||||
VALUES (?, ?, ?, ?, ?, ?, ?, ?)`,
|
||||
sourceID, targetID, t.Relation, confidence, sessionID, turnID, dateBucket, sentenceID,
|
||||
)
|
||||
if err != nil {
|
||||
var existing int
|
||||
err = tx.QueryRow(
|
||||
`SELECT 1 FROM relations WHERE source_id = ? AND target_id = ? AND relation_type = ? AND session_id = ?`,
|
||||
sourceID, targetID, t.Relation, sessionID,
|
||||
).Scan(&existing)
|
||||
if err == sql.ErrNoRows {
|
||||
_, err = tx.Exec(
|
||||
`INSERT INTO relations (source_id, target_id, relation_type, confidence, session_id, turn_id, date_bucket, sentence_id)
|
||||
VALUES (?, ?, ?, ?, ?, ?, ?, ?)`,
|
||||
sourceID, targetID, t.Relation, confidence, sessionID, turnID, dateBucket, sentenceID,
|
||||
)
|
||||
if err != nil {
|
||||
return 0, 0, err
|
||||
}
|
||||
relationsCreated++
|
||||
} else if err != nil {
|
||||
return 0, 0, err
|
||||
} else {
|
||||
// 同一(会话内)三元组已存在：仅刷新置信度与时间戳，不重复计数
|
||||
_, err = tx.Exec(
|
||||
`UPDATE relations SET confidence = ?, updated_at = CURRENT_TIMESTAMP
|
||||
WHERE source_id = ? AND target_id = ? AND relation_type = ? AND session_id = ?`,
|
||||
confidence, sourceID, targetID, t.Relation, sessionID,
|
||||
)
|
||||
if err != nil {
|
||||
return 0, 0, err
|
||||
}
|
||||
}
|
||||
relationsCreated++
|
||||
}
|
||||
|
||||
if err := tx.Commit(); err != nil {
|
||||
|
||||
@ -1,6 +1,8 @@
|
||||
package memory
|
||||
|
||||
import (
|
||||
"database/sql"
|
||||
"path/filepath"
|
||||
"os"
|
||||
"testing"
|
||||
)
|
||||
@ -60,6 +62,135 @@ func TestCommitTriples(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
func TestCommitDedupSameSession(t *testing.T) {
|
||||
g := newTestGraph(t)
|
||||
defer os.Remove(g.dbPath)
|
||||
defer g.Close()
|
||||
|
||||
triple := []Triple{{Subject: "李四", Relation: "喜欢", Object: "篮球"}}
|
||||
|
||||
ec, rc, err := g.Commit(triple, "session_dup", 1)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if ec != 2 || rc != 1 {
|
||||
t.Fatalf("first commit: want 2/1, got %d/%d", ec, rc)
|
||||
}
|
||||
|
||||
// 同一会话重复 commit 同一三元组：关系不再新增
|
||||
_, rc, err = g.Commit(triple, "session_dup", 2)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if rc != 0 {
|
||||
t.Errorf("duplicate commit should not create relations again, got %d", rc)
|
||||
}
|
||||
|
||||
var cnt int
|
||||
if err := g.db.QueryRow(`SELECT COUNT(*) FROM relations`).Scan(&cnt); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if cnt != 1 {
|
||||
t.Errorf("expected exactly 1 relation after duplicate commit, got %d", cnt)
|
||||
}
|
||||
}
|
||||
|
||||
func TestCommitDedupDifferentSession(t *testing.T) {
|
||||
g := newTestGraph(t)
|
||||
defer os.Remove(g.dbPath)
|
||||
defer g.Close()
|
||||
|
||||
triple := []Triple{{Subject: "王五", Relation: "喜欢", Object: "足球"}}
|
||||
|
||||
for _, sess := range []string{"s1", "s2"} {
|
||||
if _, _, err := g.Commit(triple, sess, 0); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
var cnt int
|
||||
if err := g.db.QueryRow(`SELECT COUNT(*) FROM relations`).Scan(&cnt); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if cnt != 2 {
|
||||
t.Errorf("different sessions may repeat a triple, expected 2 relations, got %d", cnt)
|
||||
}
|
||||
}
|
||||
|
||||
func TestMigrateRelationUniqueDedupsOldTable(t *testing.T) {
|
||||
dir := t.TempDir()
|
||||
dbPath := filepath.Join(dir, "legacy.db")
|
||||
|
||||
// 构造旧版 schema：relations 无复合唯一约束，且塞入重复行
|
||||
db, err := sql.Open("sqlite3", dbPath)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
setup := []string{
|
||||
`CREATE TABLE entities (
|
||||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||||
name TEXT UNIQUE NOT NULL,
|
||||
type TEXT DEFAULT 'Concept',
|
||||
mention_count INTEGER DEFAULT 1,
|
||||
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
|
||||
updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
|
||||
)`,
|
||||
`CREATE TABLE sentences (
|
||||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||||
text TEXT UNIQUE NOT NULL,
|
||||
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
|
||||
)`,
|
||||
`CREATE TABLE relations (
|
||||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||||
source_id INTEGER NOT NULL,
|
||||
target_id INTEGER NOT NULL,
|
||||
relation_type TEXT NOT NULL,
|
||||
confidence REAL DEFAULT 1.0,
|
||||
status TEXT DEFAULT 'active',
|
||||
session_id TEXT,
|
||||
turn_id INTEGER DEFAULT 0,
|
||||
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
|
||||
updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
|
||||
date_bucket TEXT,
|
||||
sentence_id INTEGER DEFAULT 0,
|
||||
sentence_ref TEXT DEFAULT '',
|
||||
FOREIGN KEY (source_id) REFERENCES entities(id),
|
||||
FOREIGN KEY (target_id) REFERENCES entities(id)
|
||||
)`,
|
||||
`INSERT INTO entities (id, name) VALUES (1, '张三'), (2, '编程')`,
|
||||
`INSERT INTO relations (source_id, target_id, relation_type, session_id) VALUES (1, 2, '喜欢', 's'), (1, 2, '喜欢', 's')`,
|
||||
}
|
||||
for _, s := range setup {
|
||||
if _, err := db.Exec(s); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
db.Close()
|
||||
|
||||
// 用 NewGraphDB 打开，应触发 migrateRelationUnique：重建带约束表并去重
|
||||
g, err := NewGraphDB(dbPath)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
defer g.Close()
|
||||
|
||||
var cnt int
|
||||
if err := g.db.QueryRow(`SELECT COUNT(*) FROM relations`).Scan(&cnt); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if cnt != 1 {
|
||||
t.Errorf("expected 1 relation after migration dedup, got %d", cnt)
|
||||
}
|
||||
|
||||
// 再次提交重复三元组不应再新增
|
||||
_, rc, err := g.Commit([]Triple{{Subject: "张三", Relation: "喜欢", Object: "编程"}}, "s", 1)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if rc != 0 {
|
||||
t.Errorf("after migration, duplicate commit should add 0 relations, got %d", rc)
|
||||
}
|
||||
}
|
||||
|
||||
func TestCommitEmptyTriples(t *testing.T) {
|
||||
g := newTestGraph(t)
|
||||
defer os.Remove(g.dbPath)
|
||||
|
||||
Reference in New Issue
Block a user