Files
HomeAgent/internal/agent/core/agent.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

634 lines
25 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 core
import (
"context"
"log"
"strings"
"sync"
"time"
agentPkg "github.com/JianFeeeee/HomeAgent/internal/agent"
agentAPI "github.com/JianFeeeee/HomeAgent/internal/agent/api"
agentIO "github.com/JianFeeeee/HomeAgent/internal/agent/io"
"github.com/JianFeeeee/HomeAgent/internal/events"
"github.com/JianFeeeee/HomeAgent/internal/knowledge"
"github.com/JianFeeeee/HomeAgent/internal/memory"
"github.com/JianFeeeee/HomeAgent/internal/memory/document"
"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/plugin"
"github.com/JianFeeeee/HomeAgent/internal/tracker"
"github.com/JianFeeeee/HomeAgent/pkg/types"
)
// ContextEvent 和 RelevanceContext 定义在 context.go
// Agent — 单 agent,不区分会话/实例
//
// 并发现状(M3a 起):所有任务状态只由 **schedulerLoop goroutine** 独占读写,
// 因此不再有保护整轮执行的互斥量——挂起不能持锁(见 docs/zh/input-scheduler-design.md §8.1 I3)。
// 仍需跨 goroutine 保护的是:childMu/llmMu/lastInputMu/noMergeMu 与各子系统自己的锁;
// interceptLoop 只允许触碰 preemptionRequest 与 cancelLLM(经 llmMu)。
type Agent struct {
id types.AgentID
provider agentAPI.Provider
providerManager *agentAPI.ProviderManager
io *agentIO.IOManager
memory *memory.GraphDB
// graph 是本 agent 的**图记忆共同面**(根 = 同一个 GraphDB;驻留子 = LightMemory)。
// 整理面仍走 memory 字段(子为 nil ⇒ 既有的 nil 关卡自动禁用整理面)。
graph GraphMemory
// kernelSource/parentID/taskPrompt/dataDir:驻留子相关的层级信息(见 AgentConfig)。
kernelSource string
parentID string
taskPrompt string
dataDir string
// 驻留子(父侧):登记表 + 子侧钩子。
residentMu sync.Mutex
residents map[string]*residentChild
// 子侧:向父发消息(L3)与 contextfull 上报(父侧内核级事件)的钩子。
notifyParent func(text string)
onContextFull func()
ctxFullSignaled bool
// 子侧:inputch 处理表(子持有,父 pull)。
tableMu sync.Mutex
inputchTable []InputchRecord
inputchPending *InputchRecord
currentInputch string
indexer *memory.Indexer
tracker *tracker.Tracker
context *RelevanceContext
systemPrompt string
ctx context.Context
cancel context.CancelFunc
// 文档记忆(第二层)
docStore *document.Store
// 知识库
knowledge *knowledge.Store
// 人物特质与关系网
social *social.SocialStore
// 文本记忆(原始对话日志)
textMem *text.Memory
// 媒体存储句柄已随 CAS 淘汰(块自带负载,见 plan.md)。
// 人格设定(内容来自启动时载入的人格文件/配置项)
personality *agentPkg.Personality
// 人格落库面:首启门禁与 persona_set 工具使用(见 persona.go)。
// 为 nil 时门禁与工具都静默关闭(例如单测里不接配置的场景)。
personaStore PersonaStore
// 被授权的输出通道集合(空 = 完整授权,见 AgentConfig.AllowedOutputs)。
allowedOutputs []string
// toolResultWarnTokens 是「单条工具结果过大」的告警阈值(0 = 用默认)。
// ⚠️ 方案 B 只**统计与报告**,绝不裁剪(见 toolresult_budget.go 的理由)。
toolResultWarnTokens int
// toolResultReporter 报告超限;nil 时用 logReporter。
toolResultReporter toolResultReporter
// 插件注册表(用于 plgreload)
pluginReg *plugin.Registry
pluginDir string
// 定期心跳蒸馏
distillInterval time.Duration
// 三个独立心跳任务间隔
archiveInterval time.Duration // 冷文档归档
reviewInterval time.Duration // 关系复审
mergeInterval time.Duration // 实体合并检测
// 上下文裁剪:活跃上下文最大条数,超出按相关性裁剪
maxContextSize int
// overflowStat 统计上下文超页处理(触发次数、裁剪条数、终止次数)。
// 放 Agent 上而非全局:多实例并存时各自独立(跑分台就要实例隔离)。
overflowStat struct {
sync.Mutex
Checked int // 判据执行次数(诊断:区分「没触发」与「没执行」)
Triggered int // L4 上报次数
Aborted int // 裁不出东西的终止次数
Upstream int // 上游 ErrContextFull 次数
LastPruned int // 上次裁剪条数
LastRatio float64
LastEvents int
}
// ctxTuning 是上下文预算的可调参数(零值 = 历史默认)。
//
// 从配置读入(core.agent.context.*),使不同窗口的实例可以各自调优,
// 而不是所有实例共用一套写死的 0.8 / 600000 / 1:3。
ctxTuning ContextTuning
// 当前请求的输出通道(mutex 保护,process() 内独占)
// 阶段管道:插件消息流编辑
stageHost *StageHost
eventBus *events.Bus
pluginHealth *pluginHealthTracker
// 自循环输入通道:核心内部任务(记忆消歧、系统维护、子 Agent 通知),
// 不经过 IO 层。每条消息携带目标输出通道:
// "_consolidation_" = 记忆整理(无记忆路径,不写入上下文、不 emit 响应)
// 其他 = 正常处理(写入上下文、emit 响应到该通道)
selfInputCh chan selfInputMsg
// 子任务异步执行
childMu sync.Mutex
childNextID int64
// childTasks 记录子任务状态:运行中 / 结果 / 是否已交付。
//
// 为什么保留结果而不是“读到即删”:完成通知会写进持久上下文
// (formatMergedTimeline 每轮都重新注入),模型之后还会再查。若读到即删,
// 第二次查询就得到“不存在或已过期”这个**永久失败信号**——模型据此认为
// 任务未完成,会无限重试/汇报(实测单轮 35 次工具调用、持续 514 秒)。
childTasks map[string]*childTaskState
// childSeq 给完成的任务排个序,用于有界淘汰。
childSeq int64
// 输入调度器:就绪队列、任务抽象与快照(见 scheduler.go)。
// M2 起取代 eventLoop 的隐式 channel 排队。
sched *scheduler
// 工具轮次硬上限(0 = 不限);见 AgentConfig.MaxToolTurns。
maxToolTurns int
// offload 是积压任务自动转投的参数(见 offload.go)。
offload OffloadOptions
// 进行中的 LLM 请求取消函数,interceptLoop 可调用以在请求中打断
cancelLLM context.CancelFunc
llmMu sync.Mutex
// 模型思考模式(thinking/reasoning)
thinkingEnabled bool
pendingToolBlocks []interface{} // 插件工具通过 SetToolBlocks 注入的多模态块(type agentAPI.ContentBlock),process.go 消费后追加到 tool message
// 启动时间
startTime time.Time
// 当前轮次的非文本媒体数据(图片/音频),供 describe_image 等工具访问
pendingMedia map[string]interface{}
// pendingMediaBlocks 累积本轮已捕获的媒体块。
//
// 需要缓存而不是当场挂到事件上:媒体在 process() 执行期间被捕获,
// 而承载它的 ContextEvent 要等 process() 返回后才 Append——此刻还没有 owner。
// 由 schedulerLoop goroutine 独占读写。
pendingMediaBlocks []memory.MemoryBlock
// 当前输入是否为工具提醒/中断(以 system 角色注入,避免被当成用户消息)
interruptInput bool
// 非文本输入处理配置
inputCfg types.InputProcessingConfig
// noMergeMarkets 记录被标记"禁止合并"的实体对,key="entityA||entityB"(字典序),
// 每次 reorgGraph 扫描到对应实体对时计数减一,归零后自动移除。
noMergeMarkers map[string]int
noMergeMu sync.Mutex
// TerminalRegistry 是终端会话与命令历史的权威视图(“内核开,两个插件接”)。
// 内核订阅自己的事件总线归并而来;WebUI/CLI 经 KernelStatus 读取。
terminalReg *TerminalRegistry
// 输入去重:防 webui/GUI 断线重连导致的消息重放
// key=source+"|"+content, value=上次接收时间;短窗口内同内容丢弃
lastInput map[string]time.Time
lastInputMu sync.Mutex
// 词嵌入模型,用于实体语义相似度计算
embedder *memory.StaticEmbedder
// multimodalSpace 是统一多模态向量空间(可选)。实现可以是内嵌 ONNX,
// 也可以是外部 API 客户端;两者共享同一套 L0/L2/L3 向量缓存与检索基础设施。
multimodalSpace vector.MultimodalEmbedder
// embeddingProvider 是配置里指定的统一向量空间 provider 名;
// embeddingError 是打开/适配失败的原因(成功时为空)。
// 二者只用于状态报告:区分「没配」「配了但打不开」「已启用」。
embeddingProvider string
embeddingError string
// fusionCfg 控制文本路与视觉路的跨模态融合权重,可按模型实测结果配置。
fusionCfg CrossModalFusionConfig
// 技能索引提供者:由 skillmgr 插件实现,向 system prompt 注入轻量技能索引
skillIndex SkillIndexProvider
// usageLedger 累计**跨调用**的 token 用量与缓存命中。
//
// 为何是 Agent 级而不是 TaskFrame 级:「这个会话花了多少、缓存省了多少」
// 不是单次调用的属性,必须跨轮次、跨任务累积才有意义。
// 单次用量一直在 StageCtx 与 LLM chain 事件里,缺的正是这个落点。
usageLedger usageLedger
}
// SkillIndexProvider 提供已加载技能的精炼索引,供 buildSystemPrompt 注入。
// 实现方(skillmgr)需线程安全并快速返回(每次 LLM 调用都会调用)。
type SkillIndexProvider interface {
// SkillIndex 返回多行文本的技能索引,每行形如 "name vX.Y - description";
// 无技能时返回空串。
SkillIndex() string
}
type AgentConfig struct {
ID types.AgentID
SystemPrompt string
Provider agentAPI.Provider
ProviderManager *agentAPI.ProviderManager
IO *agentIO.IOManager
Memory *memory.GraphDB
// LightMemory 是**轻量内核**的图记忆装配(驻留子用;读 temp∪main,只写 temp)。
//
// 给了它就意味着这是轻量内核:`Memory` 必须为 nil,
// 于是记忆整理面(块/媒体/流水线/整理工具)全部不可达(见 memoryface.go)。
LightMemory *memory.LightMemory
Indexer *memory.Indexer
Tracker *tracker.Tracker
DocStore *document.Store
Knowledge *knowledge.Store
SocialStore *social.SocialStore
TextMemory *text.Memory
// 媒体存储句柄已随 CAS 淘汰(块自带负载,见 plan.md)。
MultimodalSpace vector.MultimodalEmbedder
// EmbeddingProvider / EmbeddingError 是向量空间的配置身份与打开失败原因,
// 供 healthcheck_kernel 状态报告区分「未配置 / 打开失败 / 已启用」。
EmbeddingProvider string
EmbeddingError string
FusionCfg CrossModalFusionConfig // 跨模态融合权重;零值用默认
Personality *agentPkg.Personality
PersonaStore PersonaStore // 人格设定的读写面(首启门禁 + persona_set 工具)
PluginReg *plugin.Registry
// KernelSource 是本 agent 的"上级"(驻留子的父)。
//
// 设计 §6.1:某个 agent 的 L4 只属于它的**内核** —— 根 agent 的内核是内核自身与
// 内核级插件;驻留子的内核是**父 agent**。因此子的 KernelSource = 父 ⇒ 只有父
// 能在子的阶梯上产生 L4(父的"发送消息")。
KernelSource string
// ParentID 是父 agent 的 id(空 = 根 agent)。子用它判断自己是不是驻留子。
ParentID string
// DataDir 是本 agent 的数据目录;创建驻留子时用它派生 temp 图记忆路径。
DataDir string
// TaskPrompt 是在固定提示词之上注入的**任务提示词**(驻留子创建时给定)。
TaskPrompt string
// AllowedOutputs 是本 agent **被授权的输出通道集合**(设计 §4.4 / R2)。
//
// nil 或空 = **完整授权**(默认);非空 = 白名单,只允许列出的输出通道。
// 父 agent 创建驻留子时用它收窄子的输出能力。
AllowedOutputs []string
PluginDir string
DistillInterval time.Duration
ArchiveInterval time.Duration // 冷文档归档间隔(L2→L3),0 则使用 DistillInterval
ReviewInterval time.Duration // 关系复审间隔,0 则使用 DistillInterval
MergeInterval time.Duration // 实体合并检测间隔,0 则使用 DistillInterval
MaxContextSize int // 活跃上下文最大条数,超出按相关性裁剪
// CtxTuning 是上下文预算的可调参数。零值 ⇒ 使用历史默认(与硬编码时代一致)。
//
// 为何必须能从配置传进来:这些阈值原本写死在 ComputeTokenBudget 里,
// 界面改不了、不同窗口的实例也没法各自调优(详见 ContextTuning 的注释)。
CtxTuning ContextTuning
ContextSavePath string // 上下文持久化路径,空则不持久化
EmbeddingModelPath string // 预训练词嵌入模型路径(word2vec 文本格式),空则不使用
Embedder *memory.StaticEmbedder // 共享词嵌入实例;nil 时按 EmbeddingModelPath 自建
StageHost *StageHost
EventBus *events.Bus
ThinkingEnabled bool
SkillIndexProvider SkillIndexProvider
InputProcessing types.InputProcessingConfig // 非文本输入处理配置
// MaxToolTurns 是单个任务允许的工具轮次上限(0 = 不限)。
// 设计文档 D6:主循环必须有硬上限,否则模型不停调用就永不完结。
MaxToolTurns int
// Offload 是「积压任务自动转投给驻留子」的参数(见 offload.go)。
//
// 默认关闭(Enabled=false):它让**内核替父做决策**,是设计 §7
//「决策在父的模型手里」的刻意例外,因此必须由部署方显式打开。
Offload OffloadOptions
}
func New(cfg AgentConfig) *Agent {
ctx, cancel := context.WithCancel(context.Background())
if cfg.DistillInterval <= 0 {
cfg.DistillInterval = 30 * time.Minute
}
if cfg.ArchiveInterval <= 0 {
cfg.ArchiveInterval = cfg.DistillInterval
}
if cfg.ReviewInterval <= 0 {
cfg.ReviewInterval = cfg.DistillInterval
}
if cfg.MergeInterval <= 0 {
cfg.MergeInterval = cfg.DistillInterval
}
if cfg.MaxContextSize <= 0 {
cfg.MaxContextSize = 30
}
embedder := cfg.Embedder
if embedder == nil {
embedder = memory.NewStaticEmbedder(strings.Split(cfg.EmbeddingModelPath, ",")...)
}
if cfg.DocStore != nil {
// TF-IDF 内置为 fallback,无需外部注入
}
if cfg.Knowledge != nil {
cfg.Knowledge.SetVectorizer(embedder)
cfg.Knowledge.ReindexWithVectorizer(embedder)
}
rc := NewRelevanceContext(cfg.ContextSavePath, embedder)
// 裁剪保留条数:从配置传入(0 时保持 NewRelevanceContext 的历史默认)。
if cfg.CtxTuning.ProtectedCount > 0 {
rc.protectedCount = cfg.CtxTuning.ProtectedCount
}
if cfg.StageHost != nil {
rc.SetToolDefLookup(cfg.StageHost.ToolDef)
}
if cfg.IO != nil {
rc.SetChannelDefLookup(cfg.IO.GetInputChannelDef)
}
// 注入稠密多模态向量空间(可选):配置后文档检索、L0 相关性裁剪、
// 跨模态检索全部共享同一向量空间,取代稀疏 fastText 语义路。
// 未配置时退化到 TF-IDF/fastText 稀疏检索,保持既有行为。
if cfg.MultimodalSpace != nil && cfg.MultimodalSpace.Loaded() {
rc.SetDenseSpace(cfg.MultimodalSpace)
if cfg.DocStore != nil {
cfg.DocStore.SetDenseSpace(cfg.MultimodalSpace)
cfg.DocStore.BuildDenseIndex(cfg.MultimodalSpace)
}
// 知识库接入同一多模态空间:媒体作为一等节点参与稠密召回,
// 于是「按图搜知识」「按文搜含图知识」成立。
//
// 稀疏两路(词向量 + TF-IDF)**保持启用**且仍是主召回路径:多模态
// 空间未配置时知识库行为与此前逐字一致(退化为 0.5/0.5 两路融合)。
if cfg.Knowledge != nil {
// 顺序要紧:先接线再重建。ReindexDense 会尝试从 .dense.json 缓存恢复,
// 恢复不了才重算,最后把结果落盘。若先重建后接线,首次启动算出的
// 向量会被丢掉而不落盘。
cfg.Knowledge.SetDenseSpace(cfg.MultimodalSpace)
// 媒体字节随知识条目自带(CAS 淘汰后):denseFor 直接读条目目录里的
// 媒体文件,无需 mediaGet/mediaPut 接线。
built, skipped := cfg.Knowledge.ReindexDense()
if built > 0 || skipped > 0 {
log.Printf("[knowledge] 多模态稠密索引: 新建 %d 跳过 %d(其余命中缓存)", built, skipped)
}
}
}
a := &Agent{
id: cfg.ID,
startTime: time.Now(),
provider: cfg.Provider,
providerManager: cfg.ProviderManager,
io: cfg.IO,
memory: cfg.Memory,
graph: graphMemoryOf(cfg),
kernelSource: cfg.KernelSource,
parentID: cfg.ParentID,
taskPrompt: cfg.TaskPrompt,
dataDir: cfg.DataDir,
indexer: cfg.Indexer,
tracker: cfg.Tracker,
context: rc,
systemPrompt: cfg.SystemPrompt,
ctx: ctx,
cancel: cancel,
docStore: cfg.DocStore,
knowledge: cfg.Knowledge,
social: cfg.SocialStore,
textMem: cfg.TextMemory,
// mediaStore 已随 CAS 淘汰:块自带负载。知识层的媒体也已自带字节
// (见 knowledge 的媒体侧车),不再需要外部 getter/putter 接线。
personality: cfg.Personality,
personaStore: cfg.PersonaStore,
allowedOutputs: cfg.AllowedOutputs,
pluginReg: cfg.PluginReg,
pluginDir: cfg.PluginDir,
distillInterval: cfg.DistillInterval,
archiveInterval: cfg.ArchiveInterval,
reviewInterval: cfg.ReviewInterval,
mergeInterval: cfg.MergeInterval,
maxContextSize: cfg.MaxContextSize,
ctxTuning: cfg.CtxTuning,
stageHost: cfg.StageHost,
skillIndex: cfg.SkillIndexProvider,
eventBus: cfg.EventBus,
selfInputCh: make(chan selfInputMsg, 64),
childTasks: make(map[string]*childTaskState),
sched: newScheduler(256),
maxToolTurns: cfg.MaxToolTurns,
offload: cfg.Offload,
pluginHealth: newPluginHealthTracker(),
thinkingEnabled: cfg.ThinkingEnabled,
inputCfg: cfg.InputProcessing,
embedder: embedder,
multimodalSpace: cfg.MultimodalSpace,
embeddingProvider: cfg.EmbeddingProvider,
embeddingError: cfg.EmbeddingError,
fusionCfg: cfg.FusionCfg,
noMergeMarkers: make(map[string]int),
lastInput: make(map[string]time.Time),
}
// 终端权威注册表只归**根 agent**(无 ParentID)。驻留子共用同一事件总线,
// 若每个子都建一份并订阅,一次工具调用会被 N+1 份重复记账;而终端本就是
// 内核级设备,不属于任何单个驻留子。
if cfg.ParentID == "" {
a.terminalReg = NewTerminalRegistry()
}
// 输入路由:inputch 是可分配资源,划给某个 agent 后输入**只**流向那个 agent
// (设计 §4.1「路由发生在进内核之前」)。io 层不认识 agent,所以在这里把路由器
// 注入进去:插件注入输入时先问它,被别的 agent 接管就不再进本内核队列。
if a.io != nil {
a.io.SetInputRouter(a.routeInputByOwner)
}
return a
}
// SetSkillIndexProvider 注入技能索引提供者(skillmgr 插件加载后由 main 接线)。
func (a *Agent) SetSkillIndexProvider(p SkillIndexProvider) { a.skillIndex = p }
func (a *Agent) Start() {
go a.schedulerLoop()
go a.interceptLoop()
go a.distillLoop()
go a.archiveLoop()
go a.mergeLoop()
go a.reviewLoop()
go a.offloadLoop()
a.subscribeTerminalRegistry()
a.reembedStaleMedia()
a.migrateLegacyGraphMedia()
log.Printf("[agent] %s started, waiting for IO interrupts", a.id)
}
// subscribeTerminalRegistry 让内核的终端/命令历史权威视图归并事件流。
//
// 内核自己发 EventToolCall(agent 路径),agentcli 发 EventTerminalOutput
// (含生命周期事件)。两者都进这份唯一真相,WebUI/CLI 不再各自推导。
func (a *Agent) subscribeTerminalRegistry() {
if a.eventBus == nil || a.terminalReg == nil {
return
}
a.eventBus.Subscribe(events.EventToolCall, func(ev *events.Event) {
a.terminalReg.OnToolCall(ev.Payload)
})
a.eventBus.Subscribe(events.EventTerminalOutput, func(ev *events.Event) {
a.terminalReg.OnTerminalOutput(ev.Payload)
})
}
func (a *Agent) Stop() {
// 父退出**必须**销毁全部驻留子(设计 §10 硬约束:子不得比父活得久、不留孤儿)。
a.StopResidents()
// 停机前给待办任务补终态。运行中的任务会经 cancel → LLM 失败 → emitResponse
// 自然拿到终态,但**从未运行**(排队/待处理)与**已挂起**的任务不会有任何人
// 回它们;带 ResponseCh 的同步注入方(cli / clawhubadapter 均无超时)会永久挂起
// (设计 §7 I5、§11.3 X2/X4)。必须在 cancel 之前做:cancel 会让调度器直接 return。
a.drainPendingInterrupts("agent_stopped")
a.cancel()
}
// drainPendingInterrupts 给排队/待处理/已挂起任务中带同步回执通道的调用方补一条
// skipped 终态(复用 emitSkippedReply:非阻塞写,不对外发 agent_output 事件)。
func (a *Agent) drainPendingInterrupts(reason string) {
if a.sched == nil {
return
}
pending := a.sched.pendingEvents()
if len(pending) == 0 {
return
}
for _, evt := range pending {
a.emitSkippedReply(evt, reason)
}
log.Printf("[agent] %s: 停机,%d 条待办任务已补 skipped 终态", a.id, len(pending))
}
// graphMemoryOf 决定本 agent 的图记忆共同面实现。
//
// - 轻量内核(给了 LightMemory):用 LightMemory,**整理面保持 nil**;
// - 完整内核:直接用主图库(*memory.GraphDB 天然满足 GraphMemory)。
func graphMemoryOf(cfg AgentConfig) GraphMemory {
if cfg.LightMemory != nil {
return cfg.LightMemory
}
if cfg.Memory == nil {
return nil
}
return cfg.Memory
}
// graphMem 返回本 agent 的图记忆**共同面**。
//
// `graph` 显式为 nil 时回落到 `memory` —— 这样"只设 memory 的构造"
// (大量既有测试直接用 Agent 字面量)照常工作,不需要同时维护两个字段。
// 轻量内核则显式设 graph=LightMemory 且 memory=nil:共同面走 LightMemory,
// 整理面因 memory==nil 而全部不可达。
func (a *Agent) graphMem() GraphMemory {
if a.graph != nil {
return a.graph
}
if a.memory == nil {
return nil
}
return a.memory
}
func (a *Agent) ID() types.AgentID { return a.id }
// IsOutputAllowed 报告某个输出通道是否被授权给本 agent。
//
// 默认(未配置白名单)= **完整授权**;这是"默认完整授权、父可收窄"的落点。
func (a *Agent) IsOutputAllowed(channel string) bool {
if len(a.allowedOutputs) == 0 {
return true
}
for _, c := range a.allowedOutputs {
if c == channel {
return true
}
}
return false
}
// ResolveOutputTarget 解析输出通道的投递目标(agent + inputch)。
//
// ok=false 表示该输出通道由传输层(device 通道,如 qq/webui)自行处理。
func (a *Agent) ResolveOutputTarget(channel string) (agentIO.OutputTarget, bool) {
if a.io == nil || a.io.ChannelRegistry() == nil {
return agentIO.OutputTarget{}, false
}
return a.io.ChannelRegistry().ResolveOutputTarget(channel)
}
// isDuplicateInput 判断是否为短窗口内的重复输入(防 webui/GUI 断线重连消息重放)。
// key=source+"|"+content;窗口内重复返回 true 并刷新时间戳(持续轰炸时保持拦截)。
const duplicateInputWindow = 10 * time.Second
func (a *Agent) isDuplicateInput(source, content string) bool {
a.lastInputMu.Lock()
defer a.lastInputMu.Unlock()
now := time.Now()
key := source + "|" + content
if last, ok := a.lastInput[key]; ok && now.Sub(last) < duplicateInputWindow {
a.lastInput[key] = now
return true
}
a.lastInput[key] = now
// 顺带清理过期项,防止 map 无限增长
for k, t := range a.lastInput {
if now.Sub(t) > duplicateInputWindow {
delete(a.lastInput, k)
}
}
return false
}
// IsDuplicateInput 导出包装,供测试验证去重行为。
func (a *Agent) IsDuplicateInput(source, content string) bool {
return a.isDuplicateInput(source, content)
}
// SelfInputChan 返回自循环输入通道(只读,供内部测试验证)
func (a *Agent) SelfInputChan() <-chan selfInputMsg {
return a.selfInputCh
}
// injectSelf 向自循环通道发送记忆整理类内部任务(无记忆路径)。
// 线程安全,不阻塞发送者(通道缓冲 64)。
func (a *Agent) injectSelf(task string) {
a.injectSelfChannel(selfInputMsg{
text: task,
channel: channelConsolidation,
})
}
// injectSelfChannel 向自循环通道发送一条带目标通道标志的消息。
// channel == "_consolidation_" 走无记忆整理路径;其他值走正常处理路径。
func (a *Agent) injectSelfChannel(msg selfInputMsg) {
select {
case a.selfInputCh <- msg:
default:
log.Printf("[agent] self input channel full, dropping task: %s", truncateStr(msg.text, 80))
}
}