feat(memory): 图数据库自报真实读写事件,取代「按工具名猜节点」

## 问题(生产实证)

星图的「跟随 agent 活动」此前靠**工具名词元猜节点**:把 `memory_recall`
拆成 [memory, recall] 去匹配实体名。而图谱里的实体名是**中文概念**
(小宅/对话/待命),工具名是英文 —— 两套命名空间永不交集。

实测(生产库)每个工具名命中 **0** 个节点:

    hint=memory_recall  → memory/recall  → 0
    hint=files_read     → files/read     → 0
    hint=knowledge_search → knowledge/search → 0

匹配失败后走「补齐」分支,按 mention_count 取前 8 个 ⇒ **无论 agent 在干什么,
点亮的都是同一批无关节点**(小宅/对话/待命/投递/图片/文档/插件/老大)。
这就是「星图没有读写反馈」的真身。

## 改法:让图数据库在真正读写处上报

新增 `events.EventMemoryAccess` + `GraphDB.AccessHook`,在读写入口埋点上报
**真实块 ID**(不再猜):

| 操作 | 埋点 | 上报 |
|---|---|---|
| 读 | RecallSorted | blocks |
| 写 | commit | blocks + **created**(真正新建的) |
| 删 | DeleteEntity | removed |
| 合 | MergeBlocks | merged(from→to) + removed |

三处关键设计:

1. **钩子用独立锁** `hookMu`。用 `g.mu` 会**自死锁** —— 埋点方法已持该锁,
   而 Go 的 RWMutex 不可重入。
2. **上报在锁外**。拆 `commit`→`commitLocked` 两层解决 `defer Unlock` 的时序:
   钩子内部可能要回调内存系统(如 SSE 订阅者补数据),持锁调用必死锁。
3. **区分「新建」与「更新」**(`putBlockTx` 返回 isNew)。用 SELECT 判存在性
   而非 `RETURNING (xmax = 0)` —— 后者在**事务 + ON CONFLICT** 下报
   `no such column: xmax`(实测:autocommit 可用、包进 tx 就不可用),
   而本函数只在事务里被调用。重复提交同一事实不会重复播「生长」动画。

接线:`bootstrap.newMainAgent` 里 `memDB.SetAccessHook(...)` → `events.Bus`。
只在 `evBus != nil` 时接,把「内核进程」与「独立 CLI 工具」
(homed-graph-migrate / memgc 也用 NewGraphDB)区分开 —— 后者不该为一次
批量迁移向不存在的订阅者发几万条事件。

## 验证

- `access_hook_test.go`:钉住 commit/recall/delete 三类埋点真的触发、
  且重复提交不误报新建
- `access_hook_race_test.go`:4 写 + 4 读 + 钩子热替换并发,`-race` 下无竞态、
  无死锁;**断言事件数 > 0**(否则「钩子从未触发」也会假通过)
- 该测试起初是 flaky 的(读者可能在写者落库前查询 ⇒ 空结果不报事件 ⇒
  断言随机失败)。修法是先播种确定数据再并发读,已连跑 3 次稳定
- 生产实例端到端:CLI 注入一句话 → SSE 收到 6 条 memory_access,
  带真实块 ID 与 op/created
This commit is contained in:
JianFeeeee
2026-10-08 10:41:07 +08:00
parent e6c1bfa6bf
commit 9036db9321
10 changed files with 568 additions and 81 deletions

View File

@ -13,33 +13,33 @@ import (
"syscall"
"time"
agentPkg "gitcode.com/JianFeeeee/HomeAgent/internal/agent"
agentAPI "gitcode.com/JianFeeeee/HomeAgent/internal/agent/api"
agentCore "gitcode.com/JianFeeeee/HomeAgent/internal/agent/core"
agentIO "gitcode.com/JianFeeeee/HomeAgent/internal/agent/io"
internalConfig "gitcode.com/JianFeeeee/HomeAgent/internal/config"
"gitcode.com/JianFeeeee/HomeAgent/internal/events"
"gitcode.com/JianFeeeee/HomeAgent/internal/ipc"
"gitcode.com/JianFeeeee/HomeAgent/internal/knowledge"
logpkg "gitcode.com/JianFeeeee/HomeAgent/internal/log"
luapkg "gitcode.com/JianFeeeee/HomeAgent/internal/lua"
"gitcode.com/JianFeeeee/HomeAgent/internal/memory"
"gitcode.com/JianFeeeee/HomeAgent/internal/memory/distill"
"gitcode.com/JianFeeeee/HomeAgent/internal/memory/document"
"gitcode.com/JianFeeeee/HomeAgent/internal/memory/media"
"gitcode.com/JianFeeeee/HomeAgent/internal/memory/pipeline"
"gitcode.com/JianFeeeee/HomeAgent/internal/memory/social"
"gitcode.com/JianFeeeee/HomeAgent/internal/memory/text"
"gitcode.com/JianFeeeee/HomeAgent/internal/memory/vector"
"gitcode.com/JianFeeeee/HomeAgent/internal/nlp"
"gitcode.com/JianFeeeee/HomeAgent/internal/plugin"
"gitcode.com/JianFeeeee/HomeAgent/internal/recovery"
sdk "gitcode.com/JianFeeeee/HomeAgent/internal/sdk"
"gitcode.com/JianFeeeee/HomeAgent/internal/supervisor"
"gitcode.com/JianFeeeee/HomeAgent/internal/tracker"
"gitcode.com/JianFeeeee/HomeAgent/pkg/embedding"
"gitcode.com/JianFeeeee/HomeAgent/pkg/generation"
"gitcode.com/JianFeeeee/HomeAgent/pkg/types"
agentPkg "github.com/JianFeeeee/HomeAgent/internal/agent"
agentAPI "github.com/JianFeeeee/HomeAgent/internal/agent/api"
agentCore "github.com/JianFeeeee/HomeAgent/internal/agent/core"
agentIO "github.com/JianFeeeee/HomeAgent/internal/agent/io"
internalConfig "github.com/JianFeeeee/HomeAgent/internal/config"
"github.com/JianFeeeee/HomeAgent/internal/events"
"github.com/JianFeeeee/HomeAgent/internal/ipc"
"github.com/JianFeeeee/HomeAgent/internal/knowledge"
logpkg "github.com/JianFeeeee/HomeAgent/internal/log"
luapkg "github.com/JianFeeeee/HomeAgent/internal/lua"
"github.com/JianFeeeee/HomeAgent/internal/memory"
"github.com/JianFeeeee/HomeAgent/internal/memory/distill"
"github.com/JianFeeeee/HomeAgent/internal/memory/document"
"github.com/JianFeeeee/HomeAgent/internal/memory/media"
"github.com/JianFeeeee/HomeAgent/internal/memory/pipeline"
"github.com/JianFeeeee/HomeAgent/internal/memory/social"
"github.com/JianFeeeee/HomeAgent/internal/memory/text"
"github.com/JianFeeeee/HomeAgent/internal/memory/vector"
"github.com/JianFeeeee/HomeAgent/internal/nlp"
"github.com/JianFeeeee/HomeAgent/internal/plugin"
"github.com/JianFeeeee/HomeAgent/internal/recovery"
sdk "github.com/JianFeeeee/HomeAgent/internal/sdk"
"github.com/JianFeeeee/HomeAgent/internal/supervisor"
"github.com/JianFeeeee/HomeAgent/internal/tracker"
"github.com/JianFeeeee/HomeAgent/pkg/embedding"
"github.com/JianFeeeee/HomeAgent/pkg/generation"
"github.com/JianFeeeee/HomeAgent/pkg/types"
)
// defaultSystemPrompt 是内置默认人格模板:不含版本号字面量,
@ -629,6 +629,54 @@ func newMainAgent(cfg *types.Config, cfgReg *internalConfig.ConfigRegistry, prov
InputProcessing: cfg.InputProcessing,
})
// 图记忆读写钩子 → 事件总线。
//
// ★ 为什么在这里接(而不是让各调用方自己发事件):
// 图读写散在 26 个方法与 180+ 个 db 调用点上,各调用方自报必漏发;
// 而**图库自己就知道它碰了哪些块 ID**。星图此前只能靠工具名猜
// 节点(把 "memory_recall" 拆词去匹配中文实体名,实测命中 0),
// 于是每次都点亮同一批无关节点——这正是「星图没有读写反馈」的真身。
//
// ★ 只在 evBus 非 nil 时接:它把「内核进程」与「独立 CLI 工具」区分开
// (homed-graph-migrate / memgc 等也用 NewGraphDB,但没有事件总线,
// 也不该为一次批量迁移向不存在的订阅者发几万条事件)。
//
// ★ 钩子**必须非阻塞**:它在记忆读写的同步路径上,发布到总线
// 是内存操作(各订阅者回调也都很短),实测开销可忽略。
if memDB != nil && evBus != nil {
memDB.SetAccessHook(func(ev memory.AccessEvent) {
payload := map[string]interface{}{
"op": ev.Op,
}
// 只在非空时带上字段:空数组会让前端白白重算一次。
if len(ev.Blocks) > 0 {
payload["blocks"] = ev.Blocks
}
if len(ev.Created) > 0 {
payload["created"] = ev.Created
}
if len(ev.Removed) > 0 {
payload["removed"] = ev.Removed
}
if len(ev.Merged) > 0 {
merged := make([]map[string]string, 0, len(ev.Merged))
for _, m := range ev.Merged {
merged = append(merged, map[string]string{"from": m.From, "to": m.To})
}
payload["merged"] = merged
}
if ev.Tool != "" {
payload["tool"] = ev.Tool
}
evBus.Publish(&events.Event{
Type: events.EventMemoryAccess,
Source: "main",
Payload: payload,
Timestamp: time.Now().Unix(),
})
})
}
// D4:把「设备是否已授权」的判据注入插件侧(toolImpl.CanUse 消费它)。
//
// 为什么必须在这里:判据要用 agent 自己的 allowedOutputs,而

View File

@ -33,6 +33,27 @@ const (
// 职责链:发现者(兼容层)→ 移交事件 → 归属者(skillmgr)加载管理。
EventSkillDetected EventType = "skill_detected"
// EventMemoryAccess 是**图记忆读写事件**:谁读了/写了哪几个记忆块。
//
// 为什么需要它(2026-10-08):星图此前的「跟随 agent 活动」靠**工具名猜节点**
// (把 "memory_recall" 拆成词元去匹配实体名)。而图谱里的实体名是中文概念
// (小宅/对话/待命),与英文工具名永不交集 —— 实测每个工具名命中 0 个节点,
// 于是每次都回退到「按 mention_count 取前 8 个」,即每次工具调用点亮的都是
// **同一批无关节点**。
//
// 正确的做法是让**图数据库自己在读写处上报**:它本来就知道自己碰了哪些块 ID。
// payload 约定:
//
// op: "recall" | "commit" | "merge" | "purge" | "delete"
// blocks: []string,本次实际读写到的块 ID
// created: []string,本次**新建**的块 ID(驱动「从小变大」生长动画)
// removed: []string,本次**删除/合入消失**的块 ID(驱动粒子消散)
// merged: [][2]string 或 []map,合入对(驱动「两节点消失又出现」)
// tool: 触发这次访问的工具名(可空,便于 UI 显示来源)
//
// 订阅者(webui SSE)据此按**真实块 ID** 精确高亮,不再猜。
EventMemoryAccess EventType = "memory_access"
EventAll EventType = "*"
)

View File

@ -0,0 +1,124 @@
package memory
import (
"fmt"
"path/filepath"
"sync"
"testing"
"time"
)
// 钩子在并发读写下的竞态与死锁判据。
//
// ★ 为何必须专测这个:钩子用**独立锁 hookMu**(用 g.mu 会自死锁,因为
// commit/Recall 已持该锁),而订阅者回调又可能反过来访问图库。
// 这是典型的两把锁交叉场景,-race 是唯一可靠的判据。
func TestAccessHookConcurrentNoRace(t *testing.T) {
g, err := NewGraphDB(filepath.Join(t.TempDir(), "g.db"))
if err != nil {
t.Fatal(err)
}
defer g.Close()
var mu sync.Mutex
var commits, recalls int
// counting 钩子:**所有**装上去的钩子都必须计数。
//
// ★ 为什么不能只在第一个钩子里计数:本测试有一个 goroutine 在并发
// 替换钩子(验 SetAccessHook 与读写并发安全)。若替换进去的是 no-op,
// 计数会被清零,断言「事件数 > 0」就永远失败、而断言「==0」又等于没验。
// ⇒ 让每个钩子都计数,才能既测替换安全、又保证计数有意义。
counting := func(ev AccessEvent) {
mu.Lock()
switch ev.Op {
case "commit":
commits++
case "recall":
recalls++
}
mu.Unlock()
if ev.Op == "recall" {
_ = g.Path() // 只读、无锁调用:验证钩子内不持 g.mu
}
}
g.SetAccessHook(counting)
// ★ 先**播种**确定存在的数据,再开始并发读。
//
// 为什么必须播种:recall 的钩子在**没有匹配块**时不报事件(空事件不上报,
// 否则星图收到空数组只会白重算一次)。若读者与写者同时起跑,
// 读者很可能在写者落库前查询 ⇒ 全部空结果 ⇒ recall 事件为 0,
// 断言「recall > 0」随机失败。实测正是如此(同一份代码,一次 recall=3、
// 一次 recall=0)——那是 flaky 测试,不是产品缺陷。
for i := 0; i < 20; i++ {
if _, _, err := g.Commit([]Triple{{
Subject: fmt.Sprintf("seed_实体_%d", i),
Relation: "关联",
Object: fmt.Sprintf("seed_目标_%d", i),
SentenceText: fmt.Sprintf("seed 句子 %d", i),
}}, "seed", i); err != nil {
t.Fatal(err)
}
}
done := make(chan struct{})
var wg sync.WaitGroup
// 写者
for w := 0; w < 4; w++ {
wg.Add(1)
go func(id int) {
defer wg.Done()
for i := 0; i < 25; i++ {
_, _, err := g.Commit([]Triple{{
Subject: fmt.Sprintf("w%d_实体_%d", id, i),
Relation: "关联",
Object: fmt.Sprintf("w%d_目标_%d", id, i),
SentenceText: fmt.Sprintf("w%d 的句子 %d", id, i),
}}, fmt.Sprintf("s%d", id), i)
if err != nil {
t.Errorf("commit: %v", err)
return
}
}
}(w)
}
// 读者
for r := 0; r < 4; r++ {
wg.Add(1)
go func(id int) {
defer wg.Done()
for i := 0; i < 25; i++ {
// 查播种数据:保证有匹配 ⇒ recall 事件必然上报,断言不再依赖时序。
if _, err := g.Recall([]string{fmt.Sprintf("seed_实体_%d", i%20)}, nil, 2, ""); err != nil {
t.Errorf("recall: %v", err)
return
}
}
}(r)
}
// 钩子的更换(SetAccessHook 与读写并发)
wg.Add(1)
go func() {
defer wg.Done()
for i := 0; i < 20; i++ {
g.SetAccessHook(counting)
time.Sleep(2 * time.Millisecond)
}
}()
go func() { wg.Wait(); close(done) }()
select {
case <-done:
case <-time.After(60 * time.Second):
t.Fatal("60s 未完成 —— 钩子/读写路径疑似死锁")
}
mu.Lock()
c, r := commits, recalls
mu.Unlock()
t.Logf("并发完成:commit 事件=%d recall 事件=%d", c, r)
// ★ 必须断言事件真的上报过,否则「钩子从未触发」也会让本测试通过。
// 这正是「判据失效」的典型:测了「不崩」,没测「真的工作」。
if c == 0 || r == 0 {
t.Fatalf("钩子未上报事件(commit=%d recall=%d)——说明埋点没触发或钩子被覆盖", c, r)
}
}

View File

@ -0,0 +1,83 @@
package memory
import (
"path/filepath"
"testing"
)
// 钩子必须真的在 commit / recall / delete / merge 上触发,且 ID 真实。
func TestAccessHookFires(t *testing.T) {
dir := t.TempDir()
g, err := NewGraphDB(filepath.Join(dir, "g.db"))
if err != nil {
t.Fatal(err)
}
defer g.Close()
var got []AccessEvent
g.SetAccessHook(func(ev AccessEvent) { got = append(got, ev) })
// commit:应上报 written + created
if _, _, err := g.Commit([]Triple{{
Subject: "张三", Relation: "喜欢", Object: "咖啡",
SentenceText: "张三喜欢咖啡",
}}, "s1", 1); err != nil {
t.Fatal(err)
}
if len(got) == 0 || got[len(got)-1].Op != "commit" {
t.Fatalf("commit 未触发钩子: %+v", got)
}
last := got[len(got)-1]
t.Logf("commit: blocks=%d created=%d", len(last.Blocks), len(last.Created))
if len(last.Blocks) == 0 {
t.Fatal("commit 应上报写入的块 ID")
}
if len(last.Created) == 0 {
t.Fatal("首次写入应有新建块(驱动生长动画)")
}
// 重复提交同一三元组:不应再报 created(否则动画重复播)
got = nil
if _, _, err := g.Commit([]Triple{{
Subject: "张三", Relation: "喜欢", Object: "咖啡",
SentenceText: "张三喜欢咖啡",
}}, "s1", 1); err != nil {
t.Fatal(err)
}
if len(got) > 0 && len(got[len(got)-1].Created) != 0 {
t.Fatalf("重复写入不应报新建,实际 =%v", got[len(got)-1].Created)
}
t.Log("重复 commit 未误报新建 ✓")
// recall:应上报读到的块
got = nil
if _, err := g.Recall([]string{"咖啡"}, nil, 1, ""); err != nil {
t.Fatal(err)
}
found := false
for _, ev := range got {
if ev.Op == "recall" && len(ev.Blocks) > 0 {
found = true
t.Logf("recall: blocks=%v", ev.Blocks)
}
}
if !found {
t.Fatalf("recall 未上报块 ID: %+v", got)
}
// delete:应上报 removed
got = nil
if _, err := g.DeleteEntity("张三"); err != nil {
t.Fatal(err)
}
found = false
for _, ev := range got {
if ev.Op == "delete" && len(ev.Removed) > 0 {
found = true
t.Logf("delete: removed=%v", ev.Removed)
}
}
if !found {
t.Fatalf("delete 未上报被删块: %+v", got)
}
}

View File

@ -463,13 +463,31 @@ type BlockNeighbour struct {
// 返回**重定向的边数**(不含被去重丢弃的重复边)。
// 幂等:源块不存在时返回 (0, nil)。
func (g *GraphDB) MergeBlocks(sourceText, targetText string) (int, error) {
n, srcID, dstID, err := g.mergeBlocksLocked(sourceText, targetText)
// ★ 上报必须在锁释放后(见 GraphDB.onAccess 的死锁说明)。
// 合入事件驱动星图的「两节点消失又出现」:源块被删、目标块接纳,
// 所以 Removed 给 src、Blocks 给 dst(前端据此播「消失→出现」时序)。
if err == nil && srcID != "" {
g.emitAccess(AccessEvent{
Op: "merge",
Blocks: []string{dstID},
Removed: []string{srcID},
Merged: []MergePair{{From: srcID, To: dstID}},
})
}
return n, err
}
// mergeBlocksLocked 是 MergeBlocks 的实体(持 g.mu),
// 额外返回源/目标块 ID 供外部上报(上报不能在持锁时做)。
func (g *GraphDB) mergeBlocksLocked(sourceText, targetText string) (int, string, string, error) {
sourceText = strings.TrimSpace(sourceText)
targetText = strings.TrimSpace(targetText)
if sourceText == "" || targetText == "" {
return 0, fmt.Errorf("merge requires both source and target")
return 0, "", "", fmt.Errorf("merge requires both source and target")
}
if sourceText == targetText {
return 0, fmt.Errorf("merge source and target are identical: %q", sourceText)
return 0, "", "", fmt.Errorf("merge source and target are identical: %q", sourceText)
}
g.mu.Lock()
@ -477,7 +495,7 @@ func (g *GraphDB) MergeBlocks(sourceText, targetText string) (int, error) {
src, err := g.blockByTextTxLocked(sourceText)
if err != nil {
return 0, err
return 0, "", "", err
}
if src == nil {
// ★ 必须报错,不能返回 (0, nil)。
@ -495,25 +513,25 @@ func (g *GraphDB) MergeBlocks(sourceText, targetText string) (int, error) {
// 旧 MergeEntities 在这里返回 error,是对的。
// 幂等由**调用方**判断(合并后不要再合并同一个源),
// 不该由存储层替它猜。
return 0, fmt.Errorf("merge source block %q does not exist", sourceText)
return 0, "", "", fmt.Errorf("merge source block %q does not exist", sourceText)
}
dst, err := g.blockByTextTxLocked(targetText)
if err != nil {
return 0, err
return 0, "", "", err
}
if dst == nil {
// ★ 目标不存在必须报错。
// 返回 0 会让调用方以为合并成功了,而实际上什么都没做 ——
// 那正是「静默失效」的一种。
return 0, fmt.Errorf("merge target block %q does not exist", targetText)
return 0, "", "", fmt.Errorf("merge target block %q does not exist", targetText)
}
if src.ID == dst.ID {
return 0, nil
return 0, "", "", nil
}
tx, err := g.db.Begin()
if err != nil {
return 0, err
return 0, "", "", err
}
defer tx.Rollback()
@ -528,7 +546,7 @@ func (g *GraphDB) MergeBlocks(sourceText, targetText string) (int, error) {
AND COALESCE(session_id,'') != ''`,
dst.ID, src.ID)
if err != nil {
return 0, fmt.Errorf("redirect source edges: %w", err)
return 0, "", "", fmt.Errorf("redirect source edges: %w", err)
}
nOut, _ := res.RowsAffected()
@ -538,7 +556,7 @@ func (g *GraphDB) MergeBlocks(sourceText, targetText string) (int, error) {
AND COALESCE(session_id,'') != ''`,
dst.ID, src.ID)
if err != nil {
return 0, fmt.Errorf("redirect target edges: %w", err)
return 0, "", "", fmt.Errorf("redirect target edges: %w", err)
}
nIn, _ := res.RowsAffected()
@ -567,7 +585,7 @@ func (g *GraphDB) MergeBlocks(sourceText, targetText string) (int, error) {
AND COALESCE(session_id,'') != ''
AND COALESCE(status,'') != 'deleted'`,
dst.ID, dst.ID, dst.ID, dst.ID); err != nil {
return 0, fmt.Errorf("dedupe self-loops: %w", err)
return 0, "", "", fmt.Errorf("dedupe self-loops: %w", err)
}
// ── ③ 清理指向源块的结构边 ────────────────────────────────
@ -580,12 +598,12 @@ func (g *GraphDB) MergeBlocks(sourceText, targetText string) (int, error) {
WHERE (source_kind='block' AND source_id=? AND COALESCE(session_id,'')='')
OR (target_kind='block' AND target_id=? AND COALESCE(session_id,'')='')`,
src.ID, src.ID); err != nil {
return 0, fmt.Errorf("drop structural edges of source: %w", err)
return 0, "", "", fmt.Errorf("drop structural edges of source: %w", err)
}
// ── ④ 删源块 ─────────────────────────────────────────────
if _, err := tx.Exec(`DELETE FROM memory_blocks WHERE id = ?`, src.ID); err != nil {
return 0, fmt.Errorf("delete source block: %w", err)
return 0, "", "", fmt.Errorf("delete source block: %w", err)
}
// ── ⑤ 场景引用同步 ───────────────────────────────────────
@ -596,13 +614,13 @@ func (g *GraphDB) MergeBlocks(sourceText, targetText string) (int, error) {
`UPDATE scene_refs SET ref_text = ?
WHERE kind = 'block' AND ref_text = ?`,
dst.ID, src.ID); err != nil {
return 0, fmt.Errorf("repoint scene refs: %w", err)
return 0, "", "", fmt.Errorf("repoint scene refs: %w", err)
}
if err := tx.Commit(); err != nil {
return 0, err
return 0, "", "", err
}
return int(nOut + nIn), nil
return int(nOut + nIn), src.ID, dst.ID, nil
}
// blockByTextTxLocked 按文本取最早的块(调用方已持锁)。

View File

@ -147,6 +147,80 @@ type GraphDB struct {
db *sql.DB
mu sync.RWMutex
dbPath string
// onAccess 是**图读写钩子**(可为 nil),由 hookMu 保护。
//
// ❗为什么钩子必须用**独立的**锁:读写钩子在 commit/Recall 这类方法的同步
// 路径上触发,而那些方法已持 g.mu。若钩子也去取 g.mu,就是**自死锁**
// (Go 的 sync.RWMutex 不可重入)。
//
// ❗为什么埋在图数据库而不是各调用方:图读写散在 26 个方法与 180+ 个
// db 调用点上,让每个调用方自己上报必然漏发。而**图数据库自己就知道
// 它碰了哪些块 ID** —— 这正是星图需要的(它此前靠工具名猜节点,实测命中 0)。
//
// 钩子只上报**事实**(op / 块 ID / 工具名),不关心消费者是谁;
// 内核把它接到 events.Bus,CLI 工具不接(nil 即静默跳过)。
hookMu sync.RWMutex
onAccess AccessHook
}
// AccessEvent 描述一次图读写。
//
// 用结构体而非多个参数:字段后续可能增加(如关系 ID),
// 而签名一旦发布就不好改(调用点数量多)。
//
// ❗要求:实现**必须非阻塞**。钩子虽在 GraphDB 的锁外调用,
// 但仍在业务调用的同步路径上 —— 慢钩子会直接拖慢记忆读写。
// 发布到事件总线是内存操作 + 各订阅者回调,实际开销极小。
type AccessEvent struct {
// Op 是本次操作类型:recall / commit / merge / purge / delete。
Op string
// Blocks 是本次实际读到的块 ID(recall)或写入的块 ID(commit)。
Blocks []string
// Created 是本次**新建**的块 ID(驱动星图「从小变大」生长动画)。
Created []string
// Removed 是被删除/取代而消失的块 ID(驱动粒子消散动画)。
Removed []string
// Merged 是合入对(from → to),驱动「两节点消失又出现」。
Merged []MergePair
// Tool 是触发本次访问的工具名(可空)。
Tool string
}
// MergePair 是一次合入的源与目标块 ID。
type MergePair struct {
From string `json:"from"`
To string `json:"to"`
}
// AccessHook 是钩子函数类型。
type AccessHook func(ev AccessEvent)
// SetAccessHook 安装读写钩子(传 nil 卸载)。
//
// 与 GraphDB 的构造解耦:NewGraphDB 的调用点有 7 处(含 4 个独立 CLI 工具),
// 而只有内核进程需要上报 —— 加构造参数会逼所有调用点改签名。
func (g *GraphDB) SetAccessHook(h AccessHook) {
g.hookMu.Lock()
g.onAccess = h
g.hookMu.Unlock()
}
// emitAccess 上报一次图访问。调用方**必须已释放 g.mu**(见 onAccess 的注释)。
//
// 钩子为 nil 时零开销(一次指针比较),因此 CLI 工具与测试无需任何改动。
func (g *GraphDB) emitAccess(ev AccessEvent) {
g.hookMu.RLock()
h := g.onAccess
g.hookMu.RUnlock()
if h == nil {
return
}
// 空事件不上报:星图收到空数组只会白白重算一次。
if len(ev.Blocks) == 0 && len(ev.Created) == 0 && len(ev.Removed) == 0 && len(ev.Merged) == 0 {
return
}
h(ev)
}
func NewGraphDB(dbPath string) (*GraphDB, error) {
@ -542,6 +616,24 @@ func (g *GraphDB) CommitWithMedia(triples []Triple, sessionID string, turnID int
}
func (g *GraphDB) commit(triples []Triple, sessionID string, turnID int, trackSentences bool) (map[string]string, int, int, error) {
sentenceIDs, ec, rc, written, created, err := g.commitLocked(triples, sessionID, turnID, trackSentences)
// ★ 上报必须在 commitLocked **返回之后**(即 g.mu 已释放)。
// 在 commitLocked 内部(持锁时)调钩子会自死锁 —— 钩子的订阅者
// 会回调到内存系统,而那些路径要取同一把 g.mu。
//
// 只上报成功的写入:失败的 commit 什么都没落库,报出去会让星图
// 高亮一批不存在的块。
if err == nil {
g.emitAccess(AccessEvent{Op: "commit", Blocks: written, Created: created})
}
return sentenceIDs, ec, rc, err
}
// commitLocked 是 commit 的实体(持 g.mu)。
//
// 拆出原因:外部上报需要在**锁释放后**发生,而 defer g.mu.Unlock()
// 只能在函数返回时生效 —— 把两者拆成两层是最直接的解法。
func (g *GraphDB) commitLocked(triples []Triple, sessionID string, turnID int, trackSentences bool) (map[string]string, int, int, []string, []string, error) {
g.mu.Lock()
defer g.mu.Unlock()
@ -550,9 +642,14 @@ func (g *GraphDB) commit(triples []Triple, sessionID string, turnID int, trackSe
sentenceIDs = make(map[string]string)
}
// writtenIDs / createdIDs 收集本次写入与新建的块 ID,供**锁释放后**
// 通过 emitAccess 上报给星图(不能在持锁时调钩子,见 GraphDB.onAccess)。
var writtenIDs []string
var createdIDs []string
tx, err := g.db.Begin()
if err != nil {
return nil, 0, 0, err
return nil, 0, 0, nil, nil, err
}
defer tx.Rollback()
@ -587,11 +684,13 @@ func (g *GraphDB) commit(triples []Triple, sessionID string, turnID int, trackSe
// (声明提前到循环开头,否则它只在 SentenceText != "" 的分支里可见)
var sentenceBlockID string
srcBlockID, dstBlockID, edgeID, err := putTripleBlocksTx(tx, t, sessionID, turnID)
srcBlockID, dstBlockID, edgeID, tripletCreated, err := putTripleBlocksTx(tx, t, sessionID, turnID)
if err != nil {
return nil, 0, 0, fmt.Errorf("block triple %s/%s: %w",
return nil, 0, 0, nil, nil, fmt.Errorf("block triple %s/%s: %w",
t.Subject, t.Relation, err)
}
writtenIDs = append(writtenIDs, srcBlockID, dstBlockID)
createdIDs = append(createdIDs, tripletCreated...)
subjType := t.SubjectType
if subjType == "" {
@ -640,9 +739,17 @@ func (g *GraphDB) commit(triples []Triple, sessionID string, turnID int, trackSe
// 判据 zz_blockcommit_test.go 的②盯的就是这条。
// ★ 同样留空时间:原句块的时序由迁移/蒸馏显式指定,
// 这里写 now 会覆盖既有值(实测抹平了迁移块的时序)。
if err := putBlockTx(tx, NewSentenceBlock(t.SentenceText,
time.Time{}, time.Time{})); err != nil {
return nil, 0, 0, fmt.Errorf("put sentence block: %w", err)
//
// ★ 用返回值区分「新建」与「更新」:只有真正新建的块才驱动星图的
// 「生长」动画。该句被后续引用时走 UPDATE,不算新块——
// 若也播生长,同一句话每次被提及都会再长一次,看着像新记忆。
createdSentenceBlock, err := putBlockTx(tx, NewSentenceBlock(t.SentenceText,
time.Time{}, time.Time{}))
if err != nil {
return nil, 0, 0, nil, nil, fmt.Errorf("put sentence block: %w", err)
}
if createdSentenceBlock {
createdIDs = append(createdIDs, SentenceBlockID(t.SentenceText))
}
sentenceBlockID = SentenceBlockID(t.SentenceText)
if sentenceIDs != nil {
@ -697,24 +804,24 @@ func (g *GraphDB) commit(triples []Triple, sessionID string, turnID int, trackSe
if sentenceBlockID != "" && edgeID != 0 {
if err := addContainsEdgeTx(tx, sentenceBlockID, edgeID,
confidence); err != nil {
return nil, 0, 0, fmt.Errorf("contains %s->%d: %w",
return nil, 0, 0, nil, nil, fmt.Errorf("contains %s->%d: %w",
sentenceBlockID, edgeID, err)
}
}
for _, sc := range effectiveScenes(t) {
if err := tagSceneTripleTx(tx, sc, srcBlockID, dstBlockID,
edgeID, confidence); err != nil {
return nil, 0, 0, err
return nil, 0, 0, nil, nil, err
}
}
}
}
if err := tx.Commit(); err != nil {
return nil, 0, 0, err
return nil, 0, 0, nil, nil, err
}
return sentenceIDs, entitiesCreated, relationsCreated, nil
return sentenceIDs, entitiesCreated, relationsCreated, writtenIDs, createdIDs, nil
}
func validEntityName(name string) bool {
@ -820,6 +927,31 @@ func (g *GraphDB) Recall(keywords []string, seedEntities []string, depth int, se
// 判据 TestMemoryRecall_指纹不匹配的块被跳过 当场抓到这一点:
// 块路因无空间被跳过 → 走兜底 → 兜底把旧空间的块原样返回。
func (g *GraphDB) RecallSorted(keywords []string, seedEntities []string, depth int, sessionFilter, fingerprint string, mode SortMode) (*RecallResult, error) {
result, err := g.recallSortedLocked(keywords, seedEntities, depth, sessionFilter, fingerprint, mode)
// ★ 上报必须在锁释放**之后**(见 GraphDB.onAccess 的死锁说明)。
// 读事件驱动星图的「被查询提亮」:这些块就是本次召回真正取回的。
if err == nil && result != nil {
seen := make(map[string]struct{}, len(result.Entities))
ids := make([]string, 0, len(result.Entities))
for _, e := range result.Entities {
// 只报有块键的实体:旧实现下部分实体来自非块路径(无 blockKey),
// 报出去星图也找不到对应节点。
if e.blockKey == "" {
continue
}
if _, dup := seen[e.blockKey]; dup {
continue
}
seen[e.blockKey] = struct{}{}
ids = append(ids, e.blockKey)
}
g.emitAccess(AccessEvent{Op: "recall", Blocks: ids})
}
return result, err
}
// recallSortedLocked 是 RecallSorted 的实体(持 g.mu)。
func (g *GraphDB) recallSortedLocked(keywords []string, seedEntities []string, depth int, sessionFilter, fingerprint string, mode SortMode) (*RecallResult, error) {
g.mu.RLock()
defer g.mu.RUnlock()
@ -1730,18 +1862,29 @@ func (g *GraphDB) MergeEntities(sourceName, targetName string) (int, error) {
// 按 name 查 entities.id 会把「旧表里叫这个名字的行」
// 与「图里文本相同的块」当成两回事 —— 那正是本缺陷的成因。
func (g *GraphDB) DeleteEntity(name string) (DeleteResult, error) {
res, removed, err := g.deleteEntityLocked(name)
// ★ 上报必须在锁释放后(见 GraphDB.onAccess 的死锁说明)。
// 删除事件驱动星图的「粒子消散」—— removed 就是真正被删掉的块。
if err == nil {
g.emitAccess(AccessEvent{Op: "delete", Removed: removed})
}
return res, err
}
// deleteEntityLocked 是 DeleteEntity 的实体(持 g.mu),额外返回被删的块 ID。
func (g *GraphDB) deleteEntityLocked(name string) (DeleteResult, []string, error) {
g.mu.Lock()
defer g.mu.Unlock()
tx, err := g.db.Begin()
if err != nil {
return DeleteResult{}, err
return DeleteResult{}, nil, err
}
defer tx.Rollback()
name = strings.TrimSpace(name)
if name == "" {
return DeleteResult{}, fmt.Errorf("name 不能为空")
return DeleteResult{}, nil, fmt.Errorf("name 不能为空")
}
// ★ 按块文本精确匹配(不是 LIKE):DeleteEntity 的承诺是「删除这一个实体」,
@ -1750,20 +1893,20 @@ func (g *GraphDB) DeleteEntity(name string) (DeleteResult, error) {
rows, err := tx.Query(
`SELECT id FROM memory_blocks WHERE text_content = ? ORDER BY id`, name)
if err != nil {
return DeleteResult{}, err
return DeleteResult{}, nil, err
}
var blockIDs []string
for rows.Next() {
var id string
if err := rows.Scan(&id); err != nil {
rows.Close()
return DeleteResult{}, err
return DeleteResult{}, nil, err
}
blockIDs = append(blockIDs, id)
}
if err := rows.Err(); err != nil {
rows.Close()
return DeleteResult{}, err
return DeleteResult{}, nil, err
}
rows.Close()
@ -1777,7 +1920,7 @@ func (g *GraphDB) DeleteEntity(name string) (DeleteResult, error) {
// ★ 返回 0 而不是 error 也可以,但那样工具层只能回「已删除 0 个」——
// 模型分不清「删了但本来就没有」和「条件写错了」,
// 于是会重试或改口径乱猜。明确报错更有用。
return DeleteResult{}, fmt.Errorf("块 %q 不存在,删除未执行(可能已删除,或这个名字是关系文本而非端点块)", name)
return DeleteResult{}, nil, fmt.Errorf("块 %q 不存在,删除未执行(可能已删除,或这个名字是关系文本而非端点块)", name)
}
ph := placeholders(len(blockIDs))
@ -1798,7 +1941,7 @@ func (g *GraphDB) DeleteEntity(name string) (DeleteResult, error) {
edgeRes, err := tx.Exec(
`DELETE FROM memory_block_edges WHERE `+endpointCond, edgeArgs...)
if err != nil {
return DeleteResult{}, err
return DeleteResult{}, nil, err
}
edgesDeleted, _ := edgeRes.RowsAffected()
@ -1806,7 +1949,7 @@ func (g *GraphDB) DeleteEntity(name string) (DeleteResult, error) {
blockRes, err := tx.Exec(
`DELETE FROM memory_blocks WHERE id IN (`+ph+`)`, args...)
if err != nil {
return DeleteResult{}, err
return DeleteResult{}, nil, err
}
blocksDeleted, _ := blockRes.RowsAffected()
@ -1838,21 +1981,21 @@ func (g *GraphDB) DeleteEntity(name string) (DeleteResult, error) {
if _, err := tx.Exec(
`DELETE FROM relations WHERE source_id IN (SELECT id FROM entities WHERE name = ?)
OR target_id IN (SELECT id FROM entities WHERE name = ?)`, name, name); err != nil {
return DeleteResult{}, err
return DeleteResult{}, nil, err
}
if _, err := tx.Exec(`DELETE FROM entities WHERE name = ?`, name); err != nil {
return DeleteResult{}, err
return DeleteResult{}, nil, err
}
if err := tx.Commit(); err != nil {
return DeleteResult{}, err
return DeleteResult{}, nil, err
}
// ④ 摘场景引用。必须在 Commit 之后 —— purgeStaleSceneRefsLocked 走 g.db,
// 而这里的事务还没提交。
g.purgeStaleSceneRefsLocked()
return DeleteResult{Blocks: int(blocksDeleted), Edges: int(edgesDeleted)}, nil
return DeleteResult{Blocks: int(blocksDeleted), Edges: int(edgesDeleted)}, blockIDs, nil
}
// DeleteResult 报告一次删除实际删掉了什么。

View File

@ -124,10 +124,10 @@ func (g *GraphDB) MigrateLegacyTextEntities(embed EntityEmbedder) (MigrateResult
blockID := legacyEntityBlockID(e.id, e.name)
src := NewSentenceBlock(e.name, e.createdAt, e.updatedAt)
srcBlockID := src.ID
if err := putBlockTx(tx, src); err != nil {
if _, err := putBlockTx(tx, src); err != nil {
return res, fmt.Errorf("put sentence block for entity %d: %w", e.id, err)
}
if err := putBlockTx(tx, MemoryBlock{
if _, err := putBlockTx(tx, MemoryBlock{
ID: blockID,
Modality: BlockText,
Text: e.name,
@ -244,7 +244,7 @@ func (g *GraphDB) MigrateLegacyTextEntities(embed EntityEmbedder) (MigrateResult
return res, err
}
sb := NewSentenceBlock(text, time.Time{}, time.Time{})
if err := putBlockTx(tx, sb); err != nil {
if _, err := putBlockTx(tx, sb); err != nil {
_ = sentRows.Close()
return res, fmt.Errorf("put sentence block: %w", err)
}
@ -406,12 +406,23 @@ func ensureSentenceTx(tx *sql.Tx, text string) (int64, error) {
return id, nil
}
func putBlockTx(tx *sql.Tx, b MemoryBlock) error {
// ★ 返回值:第一个是**是否真的新建了行**(供星图「生长」动画区分新建/更新),
// 第二个是 error。
//
// 为何要区分:星图的「生长」动画只该在**真·新块**上播。若把「写过了」
// 也算新建,重复提交同一事实(LLM 高频重试)会让旧块反复长出来 ——
// 看着像新记忆,实际是同一个东西。
//
// 判据走「先 SELECT 存在性」而不是 `RETURNING (xmax = 0)`:后者在
// **事务 + ON CONFLICT** 组合下报 `no such column: xmax`
// (实测:autocommit 下可用,套上 tx.Begin() 后不可用),而本函数只在事务里被调用。
// 代价是多一次主键点查,开销可忽略。
func putBlockTx(tx *sql.Tx, b MemoryBlock) (bool, error) {
vectorJSON := ""
if len(b.Vector) > 0 {
raw, err := json.Marshal(b.Vector)
if err != nil {
return err
return false, err
}
vectorJSON = string(raw)
}
@ -455,6 +466,26 @@ func putBlockTx(tx *sql.Tx, b MemoryBlock) error {
if !explicitTime {
tsUpdate = "updated_at = memory_blocks.updated_at"
}
// ★ 先用 SELECT 判定「是否已存在」,而不是靠 RETURNING (xmax = 0)。
//
// 为什么不用 xmax:它在**事务 + ON CONFLICT** 组合下直接报
// `no such column: xmax`(实测:同样的语句在 autocommit 下可用,
// 一句 tx.Begin() 包起来后就不可用)。而本函数**只在事务里**被调用
// (调用方全部持 tx)⇒ xmax 在这里永远不可用。
//
// 代价:多一次主键查找。这是索引点查,开销可忽略;
// 而它换来的是可靠的「新建 vs 更新」判据 —— 星图的生长动画依赖它,
// 报错就等于整个记忆写入失败(比多一次点查贵得多)。
var existed int
switch err := tx.QueryRow(`SELECT 1 FROM memory_blocks WHERE id = ?`, b.ID).Scan(&existed); err {
case nil:
// 已存在 ⇒ 本次是更新,不算新建。
case sql.ErrNoRows:
// 不存在 ⇒ 本次会新建。
default:
return false, err
}
_, err := tx.Exec(`INSERT INTO memory_blocks
(id, modality, text_content, payload_digest, mime, size, width, height,
vector, fingerprint, source, tool, scene, semantic_type, created_at, updated_at)
@ -471,7 +502,10 @@ func putBlockTx(tx *sql.Tx, b MemoryBlock) error {
b.ID, b.Modality, b.Text, b.PayloadDigest, b.MIME, b.Size, b.Width, b.Height,
vectorJSON, b.Fingerprint, b.Source, b.Tool, b.Scene, b.SemanticType,
created, updated)
return err
if err != nil {
return false, err
}
return existed == 0, nil
}
// ★ 返回 (error, 是否真的新增了一行)。第二个返回值是给报告口径用的 ——

View File

@ -138,7 +138,7 @@ func seedSceneRefs(t *testing.T, g *GraphDB) {
{ID: "blk_src_manual", Modality: BlockText, Text: "手工建的原句块",
Source: SentenceBlockSource},
} {
if err := putBlockTx(tx, b); err != nil {
if _, err := putBlockTx(tx, b); err != nil {
t.Fatal(err)
}
}

View File

@ -108,13 +108,19 @@ func TripleBlockID(name string) string {
// 于是 Commit 只能传旧表的 relationID/entityID,
// 写出一批退场后会悬空的引用。
//
// 返回值顺序:sourceBlockID, targetBlockID, edgeID(无块时为空/0)。
// 返回值顺序:sourceBlockID, targetBlockID, edgeID, createdBlockIDs(无块时为空/0)。
//
// ★ sessionID / turnID 来自 commit() 的参数而非 Triple ——
//
// Triple 是「三元组内容」,会话是「这次写入的上下文」,
// 两者本就不该混在一个结构里。
func putTripleBlocksTx(tx *sql.Tx, t Triple, sessionID string, turnID int) (string, string, int64, error) {
func putTripleBlocksTx(tx *sql.Tx, t Triple, sessionID string, turnID int) (string, string, int64, []string, error) {
// created 收集本次**真正新建**的块 ID(已存在则为 UPDATE,不算新建)。
//
// ★ 为什么要细分到每个块:星图的「生长」动画只该在真·新块上播。
// 若按「本次 commit 有写入」就算新建,重复提交同一事实(LLM 高频重试)
// 会让旧块反复「长出来」—— 看着像新记忆,实际是同一个东西。
var created []string
// ★ SemanticType 必须带(2026-10-04)
//
// Triple.SubjectType / ObjectType 此前只写旧 entities.type,
@ -139,8 +145,12 @@ func putTripleBlocksTx(tx *sql.Tx, t Triple, sessionID string, turnID int) (stri
SemanticType: strings.TrimSpace(t.ObjectType),
}
for _, b := range []MemoryBlock{src, dst} {
if err := putBlockTx(tx, b); err != nil {
return "", "", 0, err
isNew, err := putBlockTx(tx, b)
if err != nil {
return "", "", 0, nil, err
}
if isNew {
created = append(created, b.ID)
}
}
// ★ status 必须显式写 'active'(2026-10-04)
@ -179,10 +189,10 @@ func putTripleBlocksTx(tx *sql.Tx, t Triple, sessionID string, turnID int) (stri
ORDER BY id LIMIT 1`,
src.ID, dst.ID, strings.TrimSpace(t.Relation), sessionID).Scan(&existingEdge)
if err == nil {
return src.ID, dst.ID, existingEdge, nil
return src.ID, dst.ID, existingEdge, created, nil
}
if err != sql.ErrNoRows {
return "", "", 0, err
return "", "", 0, nil, err
}
res, err := tx.Exec(
@ -211,8 +221,8 @@ func putTripleBlocksTx(tx *sql.Tx, t Triple, sessionID string, turnID int) (stri
VALUES ('block', ?, 'block', ?, ?, ?, 'active', ?, ?)`,
src.ID, dst.ID, strings.TrimSpace(t.Relation), t.Confidence, sessionID, turnID)
if err != nil {
return "", "", 0, err
return "", "", 0, nil, err
}
edgeID, err := res.LastInsertId()
return src.ID, dst.ID, edgeID, err
return src.ID, dst.ID, edgeID, created, err
}

View File

@ -1,6 +1,6 @@
package sdk
import "gitcode.com/JianFeeeee/HomeAgent/internal/events"
import "github.com/JianFeeeee/HomeAgent/internal/events"
// 事件类型/事件类型别名:插件经 SDK 订阅/发布内核事件,
// 无需直接 import internal/events(内核事件总线仅通过 SDK 暴露)。
@ -23,4 +23,10 @@ const (
// 客户端可选订做真逐 token 渲染;聚合事件仍照常发布,旧订阅者不受影响。
EventReasoningDelta = events.EventReasoningDelta
EventContentDelta = events.EventContentDelta
// EventMemoryAccess 是图记忆的真实读写事件(哪个块被读/写/删/合)。
//
// 星图靠它做「反馈」:此前只能拿 tool_call 的工具名猜节点,而工具名是英文、
// 实体名是中文概念(实测命中 0),只能退化成点亮一批固定节点。
EventMemoryAccess = events.EventMemoryAccess
)