Files
HomeAgent/internal/memory/block.go
JianFeeeee a68d6a428c feat: 模块化内核(编译期依赖图定序)+ 密钥防火墙 + 插件三态生命周期 (#10)
* feat(kernel): 模块化内核 + 编译期依赖图定序

把单体 init 序列拆成「模块自声明依赖 + 编译期生成依赖图 + 运行期只读图」。

## 为什么

原 cmd/homed/main.go 里 17 步 init 是**手写顺序**,依赖关系只存在于
注释和人脑里。实测踩到的坑(见 docs/zh/kernel-modularization-design.md):
 · nlp.defaultParser 被 initONNXParser 写、被 distiller 读,两者相距 60 行,
   只靠 ticker 10min 才没炸(隐式依赖,无声明)
 · 加一个模块要人肉找"该插在哪",插错的表现是运行期 nil 解引用
 · 关停是 defer ks.CloseAll(),而热更新路径 os.Exit(42) **跳过 defer**
   ⇒ docStore/distiller/textMem 的 flush 全部丢失(每次热更新都发生)

## 怎么改

- internal/kernelboot/:KernelState(一个结构体收全部包级共享变量,字段
  nil 安全)、Module 接口、graphfile(依赖图编解码 + 与二进制 hash 对账)、
  scheduler(按 BFS 层并行 Init)、syncbus(层边界写通知)、
  lifecycle(Quiesce/Resume 可逆伪关机,区别于一次性 CloseAll)
- internal/kernelmod/<name>/:每个模块一个文件夹,Requires/Provides 自声明;
  all/all.go 是唯一的 import 清单(Go 循环导入约束)
- cmd/homed-depgraph:编译期生成依赖图,Makefile 注入 digest
- 运行期只**读**图,不计算顺序(顺序是编译期决定的,源码里没有中央清单)

## 判据

kernelboot 8 条 + 图失败模式 7 条(全部拒绝并给准确错误)+ 模块目录数
与注册表一致性。变异测试 15/15 咬住,含:
 · 删掉 AddCloser 的锁 → -race 报数据竞争
 · Scheduler 同批跑去重 → 203 vs 200
 · 图 hash 不匹配 → 拒绝启动

## 顺带修的缺陷

热更新路径漏调模块关闭 ⇒ 显式 stopKernelModules(),并用 AST 判据钉住
"必须在 os.Exit 之前"(字符串检查在本仓已骗过我三次)。

实测:冷启动 1630ms(5 层/17 模块,原 17 步串行);生产 22 插件全部建链。

* feat(security): 密钥防火墙 + 配置加密 + close 纪律

## 密钥防火墙

模型只见别名、不见明文;插件工具调用入口(StageHost.ExecuteTool)做入向
回填。三维能力:
 · 出向:工具输出里的密钥被替换成 [[secret:name]] 别名后才进上下文
 · 入向:模型传别名,入口处填回真值(插件完全无感知)
 · 存储:config.db 里不存明文(0644 的库可读),敏感值独立 0600 文件

模式库「宁漏勿误伤」:宁可漏掉一次替换,也不把正常文本误当密钥。

## 配置加密

- internal/security/{codec,key,keyring,store,global}.go:AES-256-GCM,
  enc:v1: 前缀,密钥轮换(OldKeys 解码 + 新钥加密)
- harden.go:数据目录权限收紧
- embedded.go:支持内嵌主密钥(可选,仍保留外部 key 文件路径)
- EncryptExistingSecrets / AuditSecrets:启动时幂等迁移 + 审计

## close 纪律(用户明确要求:只要调 close 就要有前置检查)

internal/safeclose:Safe / SafeFunc。全仓 37 处裸 close(chan) → 0。
关键教训(都写进注释和判据):
 · `select { case <-ch: default: close(ch) }` **不是**并发安全的
   (两个 goroutine 可同时命中 default)—— healthcheck 原先是这种写法
 · `sync.Once` **不可重入**:X.Do(func(){ Safe(&X, ...) }) 会自锁,
   症状是 go test 直接挂住(3 个包 600s 超时),栈里只有嵌套的 Once.Do
 · map 里按请求持有的 channel(mcp/stdio.go、clawhubadapter/sidecar.go)
   改成 *pendingRPC 包装,每通道一个 once

## 判据

security 全量 + TestNoBareCloseOutsideSafeclose(AST 级扫描,含注释剥离)
+ TestNoNestedSelfOnceDeadlock(正则修正:\w+ 不匹配点号,捕获组要含 '.')。
变异 2/2 咬住。全量 52 包 ok。

* feat(plugin+deploy): 插件三态生命周期、数据目录迁移、发行包修复

## 插件三态生命周期(用户要求:自动启动 / 启用但不自动启动 / 禁用)

- Manifest 新增 requires_facilities / auto_start(*bool,用于区分
  "未声明"与"显式 false")
- plugin-host 模块(internal/kernelmod/pluginhost):用 kernelboot.Scheduler
  按依赖时机拉起,**不再手写两阶段**
- 运行态停启 StopPlugin/StartPlugin **可逆**:不写 disabled_plugins
  (不改用户决定)、不置 shuttingDown(之后还能加载)、会摘掉注册面
- 与 StopAll 的区别写进注释 —— StopAll 置 shuttingDown 后**永久无法加载**,
  那是进程退出用的,误用于"暂停"就是关了就死

## 数据目录迁移(伪关机 → 迁移 → 唤醒)

- kernelboot/lifecycle.go:Quiesce/Resume(可逆,区别于一次性 CloseAll),
  逆序暂停、逆序恢复、Pause 失败回滚
- kernelboot/migrate.go:BeginMigration/End 编排(先停插件 → 再停内核写者
  → 等在途归零)
- distiller/textmem 增 Pause/Resume 门(不是 cancel —— cancel 不可逆,
  恢复要重建整个对象)
- 模块 migration:体检 + 生成可执行迁移脚本

★ 迁移脚本的三个实测教训(都写成了判据):
 1. 无条件改 /etc/systemd/system ⇒ 演练把**生产** -data 改成了演练目录。
    加 --dry-run(零改动)+ 强制校验 systemd 当前 -data == OLD
 2. 本机没装 rsync,脚本在"服务已停、数据未搬"处硬失败 ⇒ cp -a 回退,
    并明确警告不保硬链接/ACL(生产 plugindev 有大量 git 硬链接)

## 发行包修复(全部经实测确认)

 1. package-linux.sh **完全漏打 depgraph.bin** ⇒ 内核 exit 1
    ("生产构建(内嵌 hash 非空)但读不到依赖图")
 2. build_homed 不走 Makefile ⇒ 单独修 Makefile 无效,必须同时生成图+注入
 3. build.sh 的 LDFLAGS 用**历史模块路径 gitcode.com** ⇒ -X 静默 no-op
    ⇒ --version 显示 unknown
 4. Makefile -ldflags 用**单引号** ⇒ 注入了字面 $GRAPH_HASH

## ★ build.sh 的版本号必须校验(真实事故)

原 `VERSION="$(git describe --tags --dirty)"`。无 tag 分支上它返回**分支名**
`backup/pre-identity-rewrite-20261005-51-g03d87a4e-dirty`,被注入 meta.Version
⇒ 插件握手判据 MajorVersion() 把整串当主版本 ⇒ **22 个子进程插件全部
握手失败**("主版本不匹配")。内核活着、systemd active、日志每条都像
插件的问题 —— 实际是发行包带着非法版本号。

修法:git describe 只在形如 vX.Y.Z 时采用,否则回退 meta.Version(权威源),
并对最终值做形态守卫。deploy/packaging/version_guard_test.go 3 条判据,
变异 2/2 咬住。

## deb 的 systemd unit 改为 guard 模式

ExecStart=/usr/bin/homed --role=guard --data /var/lib/homeagent
(worker exit 42 → guard 重拉,不经过 systemd、不计失败)
含 RestartSteps/RestartMaxDelaySec/TimeoutStopSec + 硬化
(NoNewPrivileges/ProtectSystem/ProtectHome/DevicePolicy/ReadWritePaths)。

## 其他

- internal/plugin/proc:非法 JSON 帧的报错**可定位**(内容摘要 + 可能原因 +
  修法)。生产实测 1038 条非法帧(插件往 stdout 打日志 = 污染协议通道)
- 示例插件 4 个 14 处 fmt.Printf → log.Printf
- hmapdev 模板同步修复;scaffold 判据剥离注释(模板注释里就含 fmt.Printf)

* chore: 收尾(主密钥轮换命令、IPC/知识库加固、SDK 对齐)

- cmd/homed-key-rotate:把 security 里的 RotateKey/RotateKeyFile/RekeySecrets
  串成一条**可执行**的运维路径。没有它,E-1 就只是库里的能力而不是
  "出事时能做的事",而轮换的价值恰恰在于能执行。
- cmd/homed:kmod_runtime(模块装配入口)、plugins_scheduled(failback
  白名单)、memcandidate、bootstrap_order_test / shutdown_flush_test
- internal/ipc、internal/devicebridge、internal/knowledge:配套加固
- third_party/homeagent-sdk/sdk:PluginMgrAPI 增三态能力(mgrStop/mgrStart/
  mgrIsRuntimeStopped)+ settings 对齐

判据:bootstrap_order_test(启动顺序不依赖手写序列)、
shutdown_flush_test(停机的两条路径必须对称 —— exit 42 会跳过 defer,
热更新路径下模块关闭/flush 曾全部丢失)。

* test(security): 测试夹具改用不可误认的占位值(修 push protection 拦截)

GitHub secret scanning 拦住了推送:三处测试夹具用了"看起来像真密钥"的
字面量:

  internal/security/prefilter_test.go:39
  internal/security/store_test.go:177
  internal/security/survey_test.go:20

它们全是**假密钥**(`AKIAIOSFODNN7EXAMPLE` 是 AWS 官方文档的标准示例,
其余是 `abcdefghij...` 这类可读占位符),但扫描器无法区分 —— 它只看形态。

★ 不绕过(不用 unblock URL、不加 allowlist):直接改夹具。
原因:一旦允许"假密钥可以通过",真密钥混进来时就没有最后一道门了。

改法:保留**模式库匹配所需的结构**,把可识别内容换成 FAKE:
  glpat-abcdefghijklmnopqrst       → glpat-FAKE-TEST-TOKEN-NOT-REAL
  sk-abcdefghij...012345           → sk-FAKE-TEST-KEY-NOT-REAL-0000000
  AKIAIOSFODNN7EXAMPLE             → AKIA0000FAKE0000TEST   (保持 20 字符)
  ASIAIOSFODNN7EXAMPLE             → ASIA0000FAKE0000TEST
  xoxb-123456789012-abcdefghijklmnop → xoxb-FAKE-TEST-TOKEN-NOT-REAL

★ 中途踩到一次:AWS 键的模式是 `\b(AKIA|ASIA)[0-9A-Z]{16}\b`(共 20 字符),
我第一次替换成 19 字符的 `AKIAFAKETESTKEY0000` ⇒ 模式匹配不上 ⇒
TestSurvey_WhatIsCaught / TestRecognizesCommonShapes /
TestPrefilterNeverBlocksRealMatches 三条当场变红。
判据正确地拒绝了"为了过扫描器而削弱覆盖"。

顺带核实:全仓 grep 无任何真实密钥泄漏。

* fix(packaging): package-linux.sh 的版本解析与 build.sh 统一

上一版只修了 build.sh,漏了 package-linux.sh —— 它仍只看 git describe:

  VERSION="${VERSION:-$(git describe --tags --dirty)}"

无 tag 分支上 git describe 返回**分支名**
(backup/pre-identity-rewrite-20261005-56-g98850e1),守卫会正确拒绝,
但要求人**手动传 VERSION** 才能打包。而 build.sh 早已能自动回退 meta.Version
⇒ 同一个版本,走 build.sh 打得出来、走 package-linux.sh 打不出来。

危害不只是"要手动传参":运维为了跑通会去掉守卫,而分支名一旦进
meta.Version 就是**插件协议主版本**的来源 ⇒ 22 个子进程插件全部握手失败
(这是真实发生过的事故,见 3efa4e5)。

修法:两个脚本用同一套规则(git describe 仅在 vX.Y.Z 形态时采用,
否则回退 meta.Version),并对最终值做形态守卫。

判据:TestBothPackagingScriptsShareVersionRules —— 逐个读两个脚本,
要求都能从 meta/meta.go 取权威版本、都校验 git describe 形态。
变异(把 package-linux.sh 改回旧规则)咬住并准确指出是哪个脚本。

实测:不传 VERSION 跑 package-linux.sh,输出
  [PKG] warn: git describe 给出非版本串(...),改用 meta.Version=1.4.0
  Version: 1.4.0

* fix(test): 跨仓判据在 CI 上应 skip 而非 fail(修 PR 双红)

CI 上 "Go build / vet / test" 与 "Build linux/amd64" 双红:

  --- FAIL: TestHmapdevTemplateImplementsAllPluginMgrMethods
      bridge_sync_test.go:170: 读模板: open
      ../../../third_party/homeagent-sdk/tools/hmapdev/templates/proc_main.go.tmpl:
      no such file or directory

## 根因:判据依赖了一个**被 gitignore 的路径**

`.gitignore:33` 有 `third_party/homeagent-sdk/tools/`(SDK 仓自治,本仓不跟踪)。
于是:
  · 本机(完整 SDK 检出)有该文件 ⇒ 判据通过
  · CI(只 checkout 本仓)没有该文件 ⇒ Fatalf ⇒ **必红**

我提交时只在本机跑测试、没看 CI,所以这个问题拖到 PR 才暴露。

## 修法

文件缺失时 `t.Skipf` 而不是 `t.Fatalf`。

★ 这不是"放弃检查":
  · 跨仓一致性由 **SDK 仓自己的判据**守着
    (tools/hmapdev/scaffold_stdout_test.go 等),那里才是模板的真实归属地
  · 本判据在本机开发时**仍然生效**(已验证:有文件时 PASS,缺文件时 SKIP)

## 验证

 · 本机(有文件)  → PASS(判据真跑)
 · 模拟 CI(移走 tools/)→ SKIP(不再 fail)
 · 全量 53 包 ok / 0 fail

## 顺带

同目录还有 e2e_template_test.go 也读 tools/,但它是 `//go:build` 门控的
e2e 判据,CI 默认不跑,无需处理。

---------

Co-authored-by: JianFeeeee <JianFeeeee@users.noreply.gitcode.com>
2026-10-17 19:18:24 +08:00

834 lines
31 KiB
Go
Raw Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

package memory
import (
"database/sql"
"encoding/json"
"fmt"
"strings"
"time"
)
// BlockModality 是一等记忆块的原生模态。
type BlockModality string
const (
BlockText BlockModality = "text"
BlockImage BlockModality = "image"
BlockVideo BlockModality = "video"
BlockAudio BlockModality = "audio"
)
// MemoryBlock 是 Context、Document、Graph 三层共同使用的记忆块值。
//
// 它不携带 Layer、Owner 或 RefCount:块当前由哪个层的容器持有,哪个层就是
// 唯一事实源。Context→Document→Graph 迁移的是这个值本身,不建立平行保活账本。
//
// ★★ 负载是 **union**:一个块要么是文本块(负载在 Text),要么是媒体块
// (负载在 Data,配 MIME/Width/Height)。两者是同一个 MemoryBlock 的两种取值,
// 不是「块 + 外挂字节层」。
//
// 为何不存在独立的字节存储:块已经是内容的载体(文本块自己带正文),媒体块
// 自当同理。把媒体内容拆到块外、只留一个 digest 指向外部 CAS,会凭空多出
// 一层需单独维护生命周期的东西(第二套向量、第二套召回路、引用账本、自己的
// db),并与「块是唯一记忆载体」直接冲突。内容随块同生同灭才是正解。
type MemoryBlock struct {
ID string `json:"id"`
Modality BlockModality `json:"modality"`
// Text 是文本块(Modality=text)的负载。
Text string `json:"text,omitempty"`
// Data 是媒体块(Modality=image/video/audio)的负载:内容的原始字节。
//
// 落盘形态随所在层而定:L0/L1 的块内联在 JSON 里(Data 序列化为 base64),
// L3 的块落 SQLite 的 BLOB 列。无论哪层,字节都不脱离块单独存在。
Data []byte `json:"data,omitempty"`
MIME string `json:"mime,omitempty"`
Size int64 `json:"size,omitempty"`
Width int `json:"width,omitempty"`
Height int `json:"height,omitempty"`
Vector []float64 `json:"vector,omitempty"`
Fingerprint string `json:"fingerprint,omitempty"`
Source string `json:"source,omitempty"`
Tool string `json:"tool,omitempty"`
// Scene 是这个块所属的场景键(可空)。
//
// 块是记忆流水线里最细的「子项目」:一段转写、一张图的描述、一份附件。
// 场景要贯穿到流水线底,就得从块开始——否则「QQ 那场对话里发过来的那张图」
// 在场面重现时永远拿不回来。块进 L3 时按 Scene 挂 scene_refs(kind='block'),
// 场景召回即可把它取回(见 GraphDB.RecallByScene)。
Scene string `json:"scene,omitempty"`
// SemanticType 是实体的语义类别(Person / Animal / Concept / Topic…),
// 对应旧 entities.type 与 Triple.SubjectType/ObjectType。
//
// ★ 2026-10-04 加入。此前 Triple 标注的类型只写旧表,块侧完全没有 ——
// 判据 TestGraphCommit_CarriesAllFields 抓到:
// Commit 标注的 (Person, Animal) 读回来是 ("block","block")。
SemanticType string `json:"semantic_type,omitempty"`
CreatedAt time.Time `json:"created_at"`
UpdatedAt time.Time `json:"updated_at"`
}
// MemoryBlockEdge 是 L3 中连接一等记忆节点的结构化语义边。
// source/target kind 当前允许 block、entity、sentence、document。
type MemoryBlockEdge struct {
ID int64 `json:"id"`
SourceKind string `json:"source_kind"`
SourceID string `json:"source_id"`
TargetKind string `json:"target_kind"`
TargetID string `json:"target_id"`
Type string `json:"type"`
CreatedAt time.Time `json:"created_at"`
// ★ 以下是「边作为独立单位」需要的属性列。
//
// 结构边(contains 等)不填这些;关系边填。
// 旧 relations 表有同名列 —— 块化后属性挂在**边**上,
// 而不是挂在一张独立的 relations 表上(那才是「边不是独立单位」的根源)。
Confidence float64 `json:"confidence,omitempty"`
SessionID string `json:"session_id,omitempty"`
TurnID int `json:"turn_id,omitempty"`
// Status 是 EdgeActive / EdgeDeleted / EdgeMerged。
// 空字符串表示结构边(无状态语义)。
Status string `json:"status,omitempty"`
// MergedInto 在 EdgeMerged 时指向目标块:源块被并进哪里,
// 历史边留在源块上不重定向。
MergedInto string `json:"merged_into,omitempty"`
}
// IsRelationEdge 判断该边是否为**关系边**(而非结构边)。
func (e MemoryBlockEdge) IsRelationEdge() bool {
return e.SessionID != "" || e.Status != "" || e.Confidence != 0
}
func validBlockModality(modality BlockModality) bool {
return modality == BlockText || modality == BlockImage || modality == BlockVideo || modality == BlockAudio
}
// PutMemoryBlocks 将完成 L2→L3 迁移的块写成 GraphDB 原生节点。
// 调用方只有在本事务成功后才能从 Document 删除这些块。
func (g *GraphDB) PutMemoryBlocks(blocks []MemoryBlock) error {
if len(blocks) == 0 {
return nil
}
g.mu.Lock()
defer g.mu.Unlock()
tx, err := g.db.Begin()
if err != nil {
return err
}
defer tx.Rollback()
for _, block := range blocks {
if block.ID == "" {
return fmt.Errorf("memory block id is required")
}
if !validBlockModality(block.Modality) {
return fmt.Errorf("memory block %s has invalid modality %q", block.ID, block.Modality)
}
vectorJSON, err := json.Marshal(block.Vector)
if err != nil {
return fmt.Errorf("marshal memory block %s vector: %w", block.ID, err)
}
now := time.Now()
if block.CreatedAt.IsZero() {
block.CreatedAt = now
}
_, err = tx.Exec(`INSERT INTO memory_blocks (
id, modality, text_content, data, mime, size, width, height,
vector, fingerprint, source, tool, scene, created_at, updated_at
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
ON CONFLICT(id) DO UPDATE SET
modality = excluded.modality,
text_content = excluded.text_content,
-- ★ 负载空值不覆盖:轻量读取路径(BlockByIDPrefix 等 lean 列 SELECT)
-- 取回的块不带 data/text,重新写回时若直接覆盖会把已存负载抹空。
data = CASE WHEN excluded.data IS NULL OR length(excluded.data) = 0
THEN memory_blocks.data ELSE excluded.data END,
mime = excluded.mime,
size = excluded.size,
width = excluded.width,
height = excluded.height,
vector = excluded.vector,
fingerprint = excluded.fingerprint,
source = excluded.source,
tool = excluded.tool,
-- 场景只在本次给了值时才覆盖:块可能先被写入、后被归档路径补挂场景,
-- 反过来「已挂场景的块被一次无场景的重写抹掉」是不可接受的静默降级。
scene = CASE WHEN excluded.scene != '' THEN excluded.scene ELSE memory_blocks.scene END,
-- ★ semantic_type 同样「空值不覆盖」:
-- 后写的三元组往往不带类型(Triple.SubjectType 可选),
-- 若让它覆盖已有类型,一次不带标注的写入就会抹掉语义类别。
semantic_type = CASE WHEN excluded.semantic_type != '' THEN excluded.semantic_type ELSE memory_blocks.semantic_type END,
updated_at = excluded.updated_at`,
block.ID, block.Modality, block.Text, block.Data, block.MIME,
block.Size, block.Width, block.Height, string(vectorJSON), block.Fingerprint,
block.Source, block.Tool, block.Scene, block.CreatedAt, now)
if err != nil {
return fmt.Errorf("put memory block %s: %w", block.ID, err)
}
// 场景引用与块同事务:块写进去了、引用丢了,这个块在场景里就永远取不回。
if block.Scene != "" {
if err := tagSceneRefTx(tx, block.Scene, "block", 0, block.ID, 1.0); err != nil {
return fmt.Errorf("tag scene for block %s: %w", block.ID, err)
}
}
}
return tx.Commit()
}
// PutDocumentNode 在 L3 登记一个文档节点,作为 document --contains--> block
// 结构边的端点。文档正文已蒸馏为实体/关系,这里只保留身份与摘要。
func (g *GraphDB) PutDocumentNode(id, summary string) error {
if id == "" {
return fmt.Errorf("document node id is required")
}
g.mu.Lock()
defer g.mu.Unlock()
_, err := g.db.Exec(`INSERT INTO documents (id, summary) VALUES (?, ?)
ON CONFLICT(id) DO UPDATE SET summary = excluded.summary`, id, summary)
return err
}
// MemoryBlocks 查询 Graph 层实际持有的一等记忆节点。
// BlocksChangedSince 返回自 since 之后**创建或更新过**的块(供星图轻量轮询)。
//
// ★ 为何需要专方法,而不是复用 GraphData() 再过滤(2026-10-08 实测):
//
// /memory/graph/pulse 本应是**轻量**端点,原实现却先调 GraphData()
// (全量 3190 块 + 2692 边)→ json.Marshal 约 3.3MB → 再反序列化 → 才按时间
// 过滤。实测它与全量端点**同价**(145ms vs 135ms),而它每 10s 被轮询一次。
//
// 这里直接用 SQL 在库侧过滤,响应体只含窗口内变动的块(几百字节~几 KB,
// 与全量差两个数量级)。
//
// limit 上限保护:窗口内也可能有大量块(如一次批量迁移),
// 而星图轮询只需知道“有新东西”,不需要全部。0 表示用默认上限。
func (g *GraphDB) BlocksChangedSince(since time.Time, limit int) ([]MemoryBlock, error) {
if limit <= 0 {
limit = 200
}
g.mu.RLock()
defer g.mu.RUnlock()
rows, err := g.db.Query(`SELECT `+blockColumns+`
FROM memory_blocks
WHERE created_at >= ? OR updated_at >= ?
ORDER BY updated_at DESC LIMIT ?`, since, since, limit)
if err != nil {
return nil, err
}
defer rows.Close()
var blocks []MemoryBlock
for rows.Next() {
block, err := scanBlockRow(rows)
if err != nil {
return nil, err
}
blocks = append(blocks, block)
}
return blocks, rows.Err()
}
func (g *GraphDB) MemoryBlocks() ([]MemoryBlock, error) {
g.mu.RLock()
defer g.mu.RUnlock()
rows, err := g.db.Query(`SELECT ` + blockColumns + `
FROM memory_blocks ORDER BY created_at, id`)
if err != nil {
return nil, err
}
defer rows.Close()
var blocks []MemoryBlock
for rows.Next() {
block, err := scanBlockRow(rows)
if err != nil {
return nil, err
}
blocks = append(blocks, block)
}
return blocks, rows.Err()
}
// scanBlockRow 读一行 memory_blocks。
//
// ★★ 列顺序与个数必须与**每个**调用它的 SELECT 完全一致
// -----------------------------------------------
// 共用扫描的代价是:SELECT 少列就整体读不出(不是某字段零值,而是
// 整行 Scan 失败 ⇒ 该路径静默返回空)。
//
// 实测踩过:MemoryBlocks() 的 SELECT 少了 scene 列(14 列),
// 而 block_recall.go / edge_entity.go 是 15 列,换成共用扫描后
// 三处一起读不出块 —— 症状是「召回返回未找到相关记忆」,
// 完全看不出是列数问题。
//
// ⇒ 改任何 memory_blocks 的 SELECT 时,**数一遍列**。
//
// ★ 提取成共用函数的原因
// ------------------
// BFSBlocks(edge_entity.go)也要读块。若它自己写一份 Scan,
// 将来加列时**两处都要改** —— 而我只加了 schema 列忘了扩读取端
// 已经犯过一次(属性静默变零值,被 TestEdge_承载关系属性 抓到)。
// 共用扫描让「加列漏改」不再可能。
// blockColumns 是 memory_blocks 的完整列清单,**必须与 scanBlockRow 的
// Scan 顺序逐列一致**。
//
// ★★★ 加列时的必读(2026-10-04 实测踩了 6 次)
//
// 加 semantic_type 那一次,**四处 SELECT 是手写列清单**(不是常量),
// 两处 Scan 也是手写 —— 于是:
//
// 「sql: expected 16 destination arguments in Scan, not 15」
//
// ★ 这个错**只在真跑 SQL 时暴露**,编译期完全无感;
//
// 而且它表现为「召回突然空了」而非「读失败」,
// 很容易被误判成召回逻辑坏了而查错方向。
//
// ⇒ 本文件的 SELECT 一律用 blockColumns,不要再手写。
// ⇒ 加列后必须做两件事:
// 1. `grep -rn 'FROM memory_blocks' --include='*.go'` 确认无手写清单
// 2. `grep -rn '&b.Scene,\|&scene,' --include='*.go'` 确认所有 Scan 已同步
//
// ★ 为什么要提成常量:2026-10-04 修过一次由此引发的生产 bug ——
//
// MemoryBlocks() 的 SELECT 漏了 scene 列,而它与 scanBlockRow 共用,
// 结果媒体与召回测试大面积变红(列数不匹配 ⇒ 读失败)。
//
// 单靠“记得同步”不可靠;提成常量后新增查询直接复用。
//
// blockColumns 是 memory_blocks 的**元数据**列清单,供所有常规读取复用。
//
// ★ 为何**不含 data**:媒体负载动辄数 MB(生产实测:图均 495KB、最大 2.7MB),
// 而 RecallBlocks 要全表扫描有向量的块。若把 data 列进来,每次召回都会把
// 几十 MB 字节读进内存。块拥有自己的负载(同一行、同一生命周期),
// 但负载**按需加载**(见 BlockData)——这与被淘汰的 CAS 有本质区别:
// 那里字节归另一个存储层所有,这里归块自己。
const blockColumns = `id, modality, text_content, mime, size, width, height,
vector, fingerprint, source, tool, scene, semantic_type, created_at, updated_at`
func scanBlockRow(rows *sql.Rows) (MemoryBlock, error) {
var b MemoryBlock
var vectorJSON string
var scene sql.NullString
if err := rows.Scan(&b.ID, &b.Modality, &b.Text,
&b.MIME, &b.Size, &b.Width, &b.Height, &vectorJSON,
&b.Fingerprint, &b.Source, &b.Tool, &scene, &b.SemanticType,
&b.CreatedAt, &b.UpdatedAt); err != nil {
return b, err
}
b.Scene = scene.String
if vectorJSON != "" && vectorJSON != "null" {
if err := json.Unmarshal([]byte(vectorJSON), &b.Vector); err != nil {
return b, fmt.Errorf("decode memory block %s vector: %w", b.ID, err)
}
}
return b, nil
}
func validGraphNodeKind(kind string) bool {
return kind == "block" || kind == "entity" || kind == "sentence" || kind == "document"
}
func graphNodeExists(tx *sql.Tx, kind, id string) (bool, error) {
var n int
var err error
switch kind {
case "block":
err = tx.QueryRow(`SELECT COUNT(*) FROM memory_blocks WHERE id = ?`, id).Scan(&n)
case "entity":
err = tx.QueryRow(`SELECT COUNT(*) FROM entities WHERE CAST(id AS TEXT) = ?`, id).Scan(&n)
case "sentence":
err = tx.QueryRow(`SELECT COUNT(*) FROM sentences WHERE CAST(id AS TEXT) = ?`, id).Scan(&n)
case "document":
err = tx.QueryRow(`SELECT COUNT(*) FROM documents WHERE id = ?`, id).Scan(&n)
default:
return false, fmt.Errorf("invalid graph node kind %q", kind)
}
return n == 1, err
}
// AddMemoryBlockEdge 建立 contains、depicts、derived_from 等原生图边。
// 端点必须是真实 Graph 节点,不能用 owner 字符串伪装关系。
func (g *GraphDB) AddMemoryBlockEdge(sourceKind, sourceID, targetKind, targetID, edgeType string) error {
if !validGraphNodeKind(sourceKind) || !validGraphNodeKind(targetKind) {
return fmt.Errorf("invalid memory block edge kinds %q -> %q", sourceKind, targetKind)
}
if sourceID == "" || targetID == "" || edgeType == "" {
return fmt.Errorf("memory block edge endpoints and type are required")
}
g.mu.Lock()
defer g.mu.Unlock()
tx, err := g.db.Begin()
if err != nil {
return err
}
defer tx.Rollback()
for _, endpoint := range []struct{ kind, id string }{{sourceKind, sourceID}, {targetKind, targetID}} {
exists, err := graphNodeExists(tx, endpoint.kind, endpoint.id)
if err != nil {
return err
}
if !exists {
return fmt.Errorf("%s graph node %s does not exist", endpoint.kind, endpoint.id)
}
}
// ★★ 结构边必须**显式查重** —— 不能再依赖 INSERT OR IGNORE
//
// 边表升格去掉了 UNIQUE(source_kind,source_id,target_kind,target_id,
// edge_type)(因为它让「同一对节点并存多条同类关系边」不可能)。
// 而 OR IGNORE 的去重**正是靠那个 UNIQUE 实现的** ——
// 约束一去,重复调用就会插出两条一样的 contains 边。
//
// 实测后果:distill 幂等与迁移幂等两个判据当场红了。
//
// ⇒ 结构边(contains 等)在这里显式查;关系边走
// AddRelationBlockEdge,它**刻意**不查重(可并存多条)。
var existing int
err = tx.QueryRow(`SELECT COUNT(*) FROM memory_block_edges
WHERE source_kind=? AND source_id=? AND target_kind=? AND target_id=?
AND edge_type=? AND COALESCE(session_id,'')=''`,
sourceKind, sourceID, targetKind, targetID, edgeType).Scan(&existing)
if err != nil {
return err
}
if existing > 0 {
return nil // 已存在:结构边幂等
}
if _, err = tx.Exec(`INSERT INTO memory_block_edges
(source_kind, source_id, target_kind, target_id, edge_type)
VALUES (?, ?, ?, ?, ?)`, sourceKind, sourceID, targetKind, targetID, edgeType); err != nil {
return err
}
return tx.Commit()
}
func (g *GraphDB) MemoryBlockEdges() ([]MemoryBlockEdge, error) {
g.mu.RLock()
defer g.mu.RUnlock()
// ★ 必须读全部 11 列 —— 只读旧 6 列会让关系边的属性
// (session_id / confidence / status)静默变成零值。
// 这正是第一版加列时漏掉的地方:写了列但没扩读取端。
rows, err := g.db.Query(`SELECT id, source_kind, source_id, target_kind, target_id,
edge_type, created_at, confidence, session_id, turn_id, status, merged_into
FROM memory_block_edges ORDER BY id`)
if err != nil {
return nil, err
}
defer rows.Close()
var edges []MemoryBlockEdge
for rows.Next() {
var edge MemoryBlockEdge
var sess, status, merged sql.NullString
var conf sql.NullFloat64
var turn sql.NullInt64
if err := rows.Scan(&edge.ID, &edge.SourceKind, &edge.SourceID, &edge.TargetKind,
&edge.TargetID, &edge.Type, &edge.CreatedAt,
&conf, &sess, &turn, &status, &merged); err != nil {
return nil, err
}
edge.Confidence = conf.Float64
edge.SessionID = sess.String
edge.Status = status.String
edge.MergedInto = merged.String
if turn.Valid {
edge.TurnID = int(turn.Int64)
}
edges = append(edges, edge)
}
return edges, rows.Err()
}
// BlocksForNode 返回与某个图节点通过任意边相连的一等记忆块。
// 例:sentence --contains--> block;entity --depicts--> block。
func (g *GraphDB) BlocksForNode(nodeKind, nodeID string) ([]MemoryBlock, error) {
g.mu.RLock()
defer g.mu.RUnlock()
rows, err := g.db.Query(`SELECT b.id, b.modality, b.text_content,
b.mime, b.size, b.width, b.height, b.vector, b.fingerprint, b.source, b.tool,
b.created_at, b.updated_at
FROM memory_block_edges e
JOIN memory_blocks b ON (
(e.source_kind = 'block' AND e.source_id = b.id AND e.target_kind = ? AND e.target_id = ?)
OR (e.target_kind = 'block' AND e.target_id = b.id AND e.source_kind = ? AND e.source_id = ?))
ORDER BY b.created_at, b.id`, nodeKind, nodeID, nodeKind, nodeID)
if err != nil {
return nil, err
}
defer rows.Close()
var blocks []MemoryBlock
for rows.Next() {
var block MemoryBlock
var vectorJSON string
if err := rows.Scan(&block.ID, &block.Modality, &block.Text,
&block.MIME, &block.Size, &block.Width, &block.Height, &vectorJSON,
&block.Fingerprint, &block.Source, &block.Tool, &block.CreatedAt,
&block.UpdatedAt); err != nil {
return nil, err
}
if vectorJSON != "" && vectorJSON != "null" {
if err := json.Unmarshal([]byte(vectorJSON), &block.Vector); err != nil {
return nil, fmt.Errorf("decode memory block %s vector: %w", block.ID, err)
}
}
blocks = append(blocks, block)
}
return blocks, rows.Err()
}
// addContainsEdgeTx 在事务内建一条**结构边**(原句块 → 关系边)。
//
// ★ 为什么不能用 AddRelationBlockEdge:它刻意不查重(同对节点可并存多条
//
// 同类型关系边,这是 52e4596 的设计)。结构边语义相反 —— 同一句与同一条
// 关系之间只应有一条 contains,重复插入会让 BFS 的邻接表里出现重复项。
//
// ★ 为什么不能用 addBlockEdgeTx:它是迁移专用的 INSERT OR IGNORE 版本,
//
// 端点校验走的是旧表。
//
// ★ 端点允许是**边 ID**:原句块指向的是关系边(memory_block_edges.id),
//
// 所以 targetKind 用 'edge'。graphNodeExists 已支持该 kind。
//
// ★ 幂等:重复写入静默跳过(返回 nil),让重试安全。
func addContainsEdgeTx(tx *sql.Tx, sentenceBlockID string, edgeID int64, weight float64) error {
if sentenceBlockID == "" || edgeID == 0 {
return nil
}
var n int
if err := tx.QueryRow(`SELECT COUNT(*) FROM memory_block_edges
WHERE source_kind='block' AND source_id=?
AND target_kind='edge' AND target_id=?
AND edge_type='contains' AND COALESCE(session_id,'')=''`,
sentenceBlockID, edgeID).Scan(&n); err != nil {
return err
}
if n > 0 {
return nil
}
_, err := tx.Exec(`INSERT INTO memory_block_edges
(source_kind, source_id, target_kind, target_id, edge_type,
confidence, status)
VALUES ('block', ?, 'edge', ?, 'contains', ?, 'active')`,
sentenceBlockID, edgeID, weight)
return err
}
// ═══════════════════════════════════════════════════════════════
// 读方切换用的薄包装(2026-10-04)
//
// 这三个函数存在的唯一理由:让 indexer / social / distill / Purge
// 能从旧表(entities/relations/sentences)迁到块体系。
//
// ★ 它们**刻意不做语义加工** —— 只做「按名找块」「枚举块名」
// 「取某块的邻居边」这三件最朴素的事。
// 任何排序、过滤、仲裁、召回打分都该在调用方做,
// 否则同一个语义会散落在两处,然后各自漂移。
//
// ★ 全部走 g.mu,所以调用方**不得**已持锁。
// ═══════════════════════════════════════════════════════════════
// BlockByText 按文本取块(精确匹配)。
//
// 用途对应旧表的 `WHERE name = ?` —— 旧表的 name 是实体名,
// 块侧的等价物是 text_content。
//
// ★ 为什么用 = 而不是 LIKE:旧表 name 是**规范化过的实体名**
//
// (Commit 走 TripleBlockID(t.Subject) 这样的内容派生 ID),
// 精确匹配能命中;模糊匹配会把「值班室分机号」与「值班室分机号(旧)」
// 混为一谈,而那正是仲裁要区分的东西。
//
// ★ 不区分 modality:媒体块的 text_content 常为空,
//
// 空串匹配不到任何真实查询(调用方不会传空串查)。
func (g *GraphDB) BlockByText(text string) (*MemoryBlock, error) {
text = strings.TrimSpace(text)
if text == "" {
return nil, nil
}
g.mu.Lock()
defer g.mu.Unlock()
blocks, err := g.blocksByTextTx(text)
if err != nil {
return nil, err
}
if len(blocks) == 0 {
return nil, nil
}
// 多个块同文本时取**最早**的:TripleBlockID 是内容派生的,
// 理论上一个文本一个块;重复只可能来自迁移的历史数据,
// 此时取最早的(它的 created_at 来自源实体,语义上更接近原意)。
b := blocks[0]
return &b, nil
}
// blocksByTextTx 按文本取全部同文本块(调用方已持锁)。
func (g *GraphDB) blocksByTextTx(text string) ([]MemoryBlock, error) {
rows, err := g.db.Query(
`SELECT `+blockColumns+` FROM memory_blocks
WHERE text_content = ? ORDER BY created_at ASC, id ASC`, text)
if err != nil {
return nil, err
}
defer rows.Close()
var out []MemoryBlock
for rows.Next() {
b, err := scanBlockRow(rows)
if err != nil {
return nil, err
}
out = append(out, b)
}
return out, rows.Err()
}
// BlockTextsLike 返回文本匹配 LIKE 的块(大小写不敏感,%value% 包裹)。
//
// 用途对应旧表的 `WHERE LOWER(name) LIKE ?`。
// ★ 调用方自己拼 LIKE 还是传裸子串?**传裸子串** ——
//
// 包裹在这里做,避免每个调用方各写各的 %…% 然后漏掉转义。
//
// ★ 不做 N 限制:调用方要么自己限,要么要全量(建索引类场景)。
//
// 但 ⚠️ 全表 LIKE 在大库上是 O(n) —— 调用方需自行评估。
func (g *GraphDB) BlockTextsLike(fragment string) ([]MemoryBlock, error) {
fragment = strings.TrimSpace(fragment)
if fragment == "" {
return nil, nil
}
pattern := "%" + escapeLike(fragment) + "%"
g.mu.Lock()
defer g.mu.Unlock()
rows, err := g.db.Query(
`SELECT `+blockColumns+` FROM memory_blocks
WHERE text_content LIKE ? ESCAPE '\'
ORDER BY created_at ASC, id ASC`, pattern)
if err != nil {
return nil, err
}
defer rows.Close()
var out []MemoryBlock
for rows.Next() {
b, err := scanBlockRow(rows)
if err != nil {
return nil, err
}
out = append(out, b)
}
return out, rows.Err()
}
// escapeLike 转义 LIKE 的元字符(% _ 与 ESCAPE 自身)。
//
// ★ 不转义的话,「50%」这类文本会变成通配符,
//
// 在 Purge 的删除路径上就是**误删**。
func escapeLike(s string) string {
r := strings.NewReplacer(`\`, `\\`, `%`, `\%`, `_`, `\_`)
return r.Replace(s)
}
// AllBlockTexts 返回全部块的文本(去重,按出现顺序)。
//
// 用途对应旧表的「全量实体名」—— 建向量索引(indexer)与
// 记忆整理(distill)都需要这个。
//
// ★ 去重:同名实体在旧表是**多行**(mention_count 不同),
//
// 而块侧的 TripleBlockID 是内容派生 ⇒ 同文本本应只有一个块。
// 去重让调用方不必处理重复,也顺带掩盖迁移期的历史重复。
//
// ★ 分批读取:全表取进内存在百万块级会炸。
//
// batchSize<=0 时用默认批大小。
func (g *GraphDB) AllBlockTexts(batchSize int) ([]string, error) {
if batchSize <= 0 {
batchSize = 5000
}
g.mu.Lock()
defer g.mu.Unlock()
var out []string
seen := make(map[string]bool)
var lastID string
for {
rows, err := g.db.Query(
`SELECT id, text_content FROM memory_blocks
WHERE id > ? AND text_content != ''
ORDER BY id LIMIT ?`, lastID, batchSize)
if err != nil {
return nil, err
}
n := 0
var nextID string
for rows.Next() {
var id, text string
if err := rows.Scan(&id, &text); err != nil {
rows.Close()
return nil, err
}
nextID = id
n++
if seen[text] {
continue
}
seen[text] = true
out = append(out, text)
}
rows.Close()
if err := rows.Err(); err != nil {
return nil, err
}
if n < batchSize {
break
}
lastID = nextID
}
return out, nil
}
// NeighbourEdgesOfBlock 返回以该块为端点的全部 active 关系边及其对端块。
//
// 用途对应旧表的「按实体名找它的所有关系」—— social 层按人/物聚合
// 社交关系时用它。
//
// ★ 双向:块作为 source 或 target 都算。
// ★ 只取非 deleted 的边(与全部读路径的 status 口径一致)。
//
// ★ 返回顺序按创建时间 —— 稳定的输出对测试与展示都重要,
//
// 而 SQLite 不保证无 ORDER BY 的行序。
func (g *GraphDB) NeighbourEdgesOfBlock(blockID string) ([]BlockNeighbour, []MemoryBlock, error) {
if blockID == "" {
return nil, nil, nil
}
g.mu.Lock()
defer g.mu.Unlock()
rows, err := g.db.Query(
`SELECT e.id, e.source_id, e.target_id, e.edge_type,
COALESCE(e.confidence, 0), COALESCE(e.session_id, ''),
COALESCE(e.turn_id, 0), COALESCE(e.status, ''),
COALESCE(e.created_at, ''),
COALESCE(other.id, ''), COALESCE(other.text_content, '')
FROM memory_block_edges e
LEFT JOIN memory_blocks other
ON other.id = CASE WHEN e.source_id = ? THEN e.target_id ELSE e.source_id END
WHERE (e.source_id = ? OR e.target_id = ?)
AND e.source_kind = 'block' AND e.target_kind = 'block'
AND COALESCE(e.status, '') != 'deleted'
ORDER BY e.created_at ASC, e.id ASC`, blockID, blockID, blockID)
if err != nil {
return nil, nil, err
}
defer rows.Close()
var edges []BlockNeighbour
var peers []MemoryBlock
seenPeer := make(map[string]bool)
for rows.Next() {
var nb BlockNeighbour
var otherID, otherText, created string
if err := rows.Scan(&nb.EdgeID, &nb.SourceID, &nb.TargetID, &nb.EdgeType,
&nb.Confidence, &nb.SessionID, &nb.TurnID, &nb.Status, &created,
&otherID, &otherText); err != nil {
return nil, nil, err
}
nb.IsOutgoing = nb.SourceID == blockID
nb.Peer = MemoryBlock{
ID: otherID,
Modality: BlockText,
Text: otherText,
CreatedAt: parseLegacyTime(created),
UpdatedAt: parseLegacyTime(created),
}
nb.RelationEdgeData = RelationEdgeData{
Confidence: nb.Confidence,
SessionID: nb.SessionID,
TurnID: nb.TurnID,
Status: nb.Status,
}
edges = append(edges, nb)
if otherText != "" && !seenPeer[otherID] {
seenPeer[otherID] = true
peers = append(peers, nb.Peer)
}
}
return edges, peers, rows.Err()
}
// blockSemanticType 返回块对外呈现的语义类型。
//
// ★ 为什么要这个函数而不是直接用 b.SemanticType:
//
// 语义类型是**可选标注**,绝大多数块没被标注过。
// 旧 entities 表在缺省时写 "Concept"(graph.go 里 subjType 默认值),
// 所以读侧(social 的 ListPersons、SDK 的实体类型)期待一个具体字符串。
// 返回空串会让下游的 == "Person" 之类的比较全部落空 ——
// 那不是「未标注」,那像是「数据坏了」。
//
// ★ 取值优先级:
// 1. 块自带标注(Person / Animal / Concept…)—— 直接用
// 2. 未标注 → "block"(说清「这是一个图节点,且没标类型」)
// ★ 不用 "Concept":那是旧表的默认值,
// 拿来兜底会让「未标注」与「确实是概念」无法区分。
func blockSemanticType(b MemoryBlock) string {
if t := strings.TrimSpace(b.SemanticType); t != "" {
return t
}
return "block"
}
// MemoryBlockCount 返回当前块总数。
//
// ★ 用途:判断「一次写入是否真的落库了」
//
// 停旧表双写后,Commit 的两个返回值恒为 0(它们数的是旧表行数),
// 于是「有没有写进去」失去了信号 ——
// 而 archiveColdDocs 正需要它来决定文档能否删除
// (没写进去就删 = 丢数据)。
//
// ★ 为什么用块数而不是「本次新增块数」:
//
// 新增数需要事务内计数,而 MemoryBlockCount 是事务外的简单计数。
// 调用方在写入前后各取一次相减即可,误差只来自并发写入 ——
// 归档是后台低频任务,这个精度足够。
func (g *GraphDB) MemoryBlockCount() (int, error) {
g.mu.RLock()
defer g.mu.RUnlock()
var n int
if err := g.db.QueryRow(`SELECT COUNT(*) FROM memory_blocks`).Scan(&n); err != nil {
return 0, err
}
return n, nil
}
// MemoryEdgeCount 返回关系边数(不含 contains 结构边)。
//
// ★ 为什么需要它(2026-10-05)
//
// 旧表停双写后 Commit 的 ec/rc 恒为 0(graph.go 里
// relationsCreated = 0 是写死的),所以「写了多少条关系」
// 唯一的可信来源只能是**实测边数差值**。
//
// 不含 contains:那是结构边(原句块 → 字段块 / 关系边),
// 混进来会让「关系数」随原句数量翻倍 —— 与 Introspect 的
// relation_count 口径保持一致。
func (g *GraphDB) MemoryEdgeCount() (int, error) {
g.mu.RLock()
defer g.mu.RUnlock()
var n int
if err := g.db.QueryRow(
`SELECT COUNT(*) FROM memory_block_edges
WHERE source_kind = 'block' AND target_kind = 'block'
AND edge_type != 'contains'`).Scan(&n); err != nil {
return 0, err
}
return n, nil
}