mirror of
https://gitcode.com/JianFeeeee/HomeAgent.git
synced 2026-11-06 01:07:35 +00:00
fix(plugin): 修两处持锁自死锁(CI 双红真因)+ tools_ready 事件声明
## 死锁(这是 CI 连续两次红的真因) `StopAll` 与 `StopPlugin` 都在持有 `r.mu` **写锁**时调用 `RemoveBuiltinSkill` → `skillDirOf` → `r.mu.RLock()` ⇒ **自死锁**。 后果不是报错而是**卡死**: - internal/plugins 卡 300s 超时(TestCapability_RestrictedPluginStillLoads) - internal/plugin 卡 840s 超时(TestStopPluginDoesNotTouchPersistentState) - CI 的 `go test ./...` 与 `go test -race` 双双红 修法:拆出**不加锁**的内部版本,持锁路径只走它: - `skillDirFrom(dataDir, name)`(纯函数) - `removeBuiltinSkillLocked(pluginName, p, dataDir)`(显式传 dataDir, 避免"看起来没取锁、一改就取锁"的隐患) - `announceSkillRemovedLocked`(持锁下直读 r.evBus) 修后:internal/plugins 300s → 12s,internal/plugin 840s → 0.7s。 ## 防回归判据 `TestStopAllAndStopPluginDoNotDeadlock`:死锁**不返回**,普通断言抓不到, 所以用**超时**当判据(5 秒未返回即失败),并指明应改用 Locked 版本。 ## tools_ready 事件声明 新增的 tools_ready 触发 `TestEveryPublishedEventHasAnOutwardPath` (该判据防"事件发了、客户端收不到"的静默失效)。 tools_ready 是**内核内部**就绪信号(plugin-host → seq 等订阅者), 不是对话流程事件,前端无需感知 ⇒ 在 allowed 里显式声明理由。 ## 验证 go build ./... / go vet ./... / go test -p 1 ./... / -race 全通过。
This commit is contained in:
90
internal/plugins/seq/boot.go
Normal file
90
internal/plugins/seq/boot.go
Normal file
@ -0,0 +1,90 @@
|
||||
package seq
|
||||
|
||||
// 开机序列:系统就绪时自动执行标记了 boot 的序列。
|
||||
//
|
||||
// ## 为什么这是 seq 最核心的用途(用户原话)
|
||||
//
|
||||
// 「seq 最核心的是定时复用,比如开机执行一段序列」
|
||||
//
|
||||
// 单次任务里编序列往往**不划算**:先 seq_create 写 JSON 再 seq_run,
|
||||
// 比直接调工具多两步(e2e 实测:多文件读取场景模型直接并行 files_read
|
||||
// 就是最优解,包成序列纯亏)。真正省事的是**同一套流程反复跑** ——
|
||||
// 而开机是最典型的周期性触发点。
|
||||
//
|
||||
// ## 时序(为什么订阅 tools_ready 而不是在 Start 里跑)
|
||||
//
|
||||
// 插件 Start 的先后由依赖图决定,此刻:
|
||||
// · 序列要调的工具可能属于还没拉起的插件
|
||||
// · agent 循环还没开始,投递的输入无人接收
|
||||
//
|
||||
// plugin-host 在 `ks.PluginsReady = true`(全部插件含内置都已加载)
|
||||
// 之后发布 tools_ready,那一刻**工具才全部可调**。
|
||||
// 本文件只订阅那一个事件 —— 不需要等 agent 起来。
|
||||
|
||||
import (
|
||||
"log"
|
||||
|
||||
"github.com/JianFeeeee/HologramAgent/internal/events"
|
||||
)
|
||||
|
||||
// subscribeBoot 订阅 tools_ready,在该时刻执行所有 boot 序列。
|
||||
//
|
||||
// 订阅 tools_ready 而不是"agent 就绪":seq 自动运行**不依赖 agent** ——
|
||||
// 它只是一段工具调用序列,返回值进队列等 agent 起来后处理即可。
|
||||
// 绑在 agent 启动上会把时机人为推后,且依赖编排层的判断。
|
||||
//
|
||||
// 返回退订函数(Stop 时调用)。订阅失败不阻断插件启动:
|
||||
// 工具本身仍可用,只是开机序列不会跑 —— 但必须留日志,
|
||||
// 否则"为什么开机没执行"完全无从查起。
|
||||
func (p *Plugin) subscribeBoot() func() {
|
||||
if p.sdk == nil {
|
||||
return func() {}
|
||||
}
|
||||
unsub := p.sdk.Subscribe(events.EventToolsReady, func(evt *events.Event) {
|
||||
// 事件在总线 goroutine 上分发,不能阻塞它:开机序列可能跑很久。
|
||||
go p.runBootSequences()
|
||||
})
|
||||
return unsub
|
||||
}
|
||||
|
||||
// runBootSequences 依次执行所有 boot 序列。
|
||||
//
|
||||
// 串行执行(不是并发):开机序列通常彼此独立,但并发跑会让日志交错、
|
||||
// 且若两条序列争用同一资源,失败原因极难定位。开机不是性能敏感路径。
|
||||
//
|
||||
// 单条失败不影响后续序列 —— 一条坏了不该让其余开机动作全部失效。
|
||||
func (p *Plugin) runBootSequences() {
|
||||
names := p.store.List()
|
||||
var boots []string
|
||||
for _, n := range names {
|
||||
sq, err := p.store.Load(n)
|
||||
if err != nil {
|
||||
// 坏序列不阻断:store 里可能有历史遗留文件
|
||||
log.Printf("[seq] boot scan skip %q: %v", n, err)
|
||||
continue
|
||||
}
|
||||
if sq.Boot {
|
||||
boots = append(boots, n)
|
||||
}
|
||||
}
|
||||
if len(boots) == 0 {
|
||||
return
|
||||
}
|
||||
|
||||
log.Printf("[seq] tools_ready: running %d boot sequence(s): %v", len(boots), boots)
|
||||
for _, name := range boots {
|
||||
sq, err := p.store.Load(name)
|
||||
if err != nil {
|
||||
log.Printf("[seq] boot %q load failed: %v", name, err)
|
||||
continue
|
||||
}
|
||||
// 开机序列无外部入参(nil):组的 when/args 按既有语义取零值。
|
||||
res, err := p.runSequence(sq, nil, nil)
|
||||
if err != nil {
|
||||
log.Printf("[seq] boot %q failed: %v", name, err)
|
||||
continue
|
||||
}
|
||||
// runSequence 返回渲染后的摘要(interface{}),含失败标记时如实转述。
|
||||
log.Printf("[seq] boot %q done (%d groups): %v", name, len(sq.Groups), res)
|
||||
}
|
||||
}
|
||||
70
internal/plugins/seq/boot_test.go
Normal file
70
internal/plugins/seq/boot_test.go
Normal file
@ -0,0 +1,70 @@
|
||||
package seq
|
||||
|
||||
// 开机序列的判据。
|
||||
//
|
||||
// 背景(用户裁定):seq 的核心价值不是"单次任务省几次调用",而是
|
||||
// **定时/开机复用**。e2e 实测证明前者不成立(一次性多文件读直接并行
|
||||
// files_read 就是最优解),所以 Boot 才是这个插件存在的理由。
|
||||
|
||||
import (
|
||||
"strings"
|
||||
"testing"
|
||||
)
|
||||
|
||||
// boot 必须能从 JSON 里反序列化出来(序列文件手写 "boot": true 的路径)。
|
||||
func TestBootMarkerParsesFromJSON(t *testing.T) {
|
||||
text := `{
|
||||
"name": "boot-check",
|
||||
"boot": true,
|
||||
"groups": [
|
||||
{"name": "g1", "parallel": true, "out": {"r": "string"},
|
||||
"tools": "{\"tool\":\"cmd_run\",\"args\":{},\"as\":\"r\"};"}
|
||||
]
|
||||
}`
|
||||
sq, err := Parse([]byte(text))
|
||||
if err != nil {
|
||||
t.Fatalf("parse: %v", err)
|
||||
}
|
||||
if !sq.Boot {
|
||||
t.Fatal("boot:true 未被解析(开机序列会静默不执行)")
|
||||
}
|
||||
}
|
||||
|
||||
// 未标 boot 的序列默认不参与开机执行 —— 否则每条序列都会在启动时乱跑。
|
||||
func TestBootDefaultsToFalse(t *testing.T) {
|
||||
text := `{"name":"normal","groups":[{"name":"g","parallel":true,"out":{"r":"string"},
|
||||
"tools":"{\"tool\":\"cmd_run\",\"args\":{},\"as\":\"r\"};"}]}`
|
||||
sq, err := Parse([]byte(text))
|
||||
if err != nil {
|
||||
t.Fatalf("parse: %v", err)
|
||||
}
|
||||
if sq.Boot {
|
||||
t.Fatal("未声明 boot 的序列不应标记为开机序列")
|
||||
}
|
||||
}
|
||||
|
||||
// runBootSequences 必须能在没有任何 boot 序列时安静返回(不能 panic、不能乱跑)。
|
||||
func TestRunBootSequencesNoopWhenNone(t *testing.T) {
|
||||
p := New("seq")
|
||||
p.store = NewStore(t.TempDir())
|
||||
// 不 panic、不报错即为通过
|
||||
p.runBootSequences()
|
||||
}
|
||||
|
||||
// 工具描述必须告知模型 boot 的语义(否则没人会想到标它)。
|
||||
func TestCreateToolAdvertisesBoot(t *testing.T) {
|
||||
defs := New("seq").toolDefs()
|
||||
def, ok := defs["seq_create"]
|
||||
if !ok {
|
||||
t.Fatal("缺 seq_create 定义")
|
||||
}
|
||||
props, _ := def.Parameters["properties"].(map[string]interface{})
|
||||
boot, ok := props["boot"].(map[string]interface{})
|
||||
if !ok {
|
||||
t.Fatal("seq_create 必须暴露 boot 参数(否则模型无从知道可标记开机序列)")
|
||||
}
|
||||
desc, _ := boot["description"].(string)
|
||||
if !strings.Contains(desc, "启动") {
|
||||
t.Fatalf("boot 描述应说明「系统就绪/启动时执行」,实际:%q", desc)
|
||||
}
|
||||
}
|
||||
@ -63,6 +63,13 @@ func (p *Plugin) seqCreate(args map[string]interface{}) (interface{}, error) {
|
||||
if d := argString(args, "description"); d != "" {
|
||||
seq.Description = d
|
||||
}
|
||||
// boot:系统就绪(agent_ready)时自动执行一次。
|
||||
// 显式传才生效;未传则保留序列文件里可能已写的 boot 标记。
|
||||
if v, ok := args["boot"]; ok {
|
||||
if b, isBool := v.(bool); isBool {
|
||||
seq.Boot = b
|
||||
}
|
||||
}
|
||||
if err := p.store.Save(seq); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
@ -71,6 +78,9 @@ func (p *Plugin) seqCreate(args map[string]interface{}) (interface{}, error) {
|
||||
|
||||
var sb strings.Builder
|
||||
fmt.Fprintf(&sb, "序列 %q 已保存(%s,%d 个 group)", seq.Name, from, len(seq.Groups))
|
||||
if seq.Boot {
|
||||
fmt.Fprintf(&sb, "\n- ⚙ 已标记 boot:**下次系统就绪时自动执行一次**")
|
||||
}
|
||||
for _, g := range seq.Groups {
|
||||
fmt.Fprintf(&sb, "\n- %s", g.Name)
|
||||
if len(g.In) > 0 {
|
||||
|
||||
@ -25,6 +25,20 @@ type Sequence struct {
|
||||
Name string `json:"name"`
|
||||
Description string `json:"description,omitempty"`
|
||||
Groups []Group `json:"groups"`
|
||||
|
||||
// Boot 标记该序列在**系统就绪时自动执行一次**(每次进程启动一次)。
|
||||
//
|
||||
// 这是序列最核心的用途之一(用户原话:"seq 最核心的是定时复用,
|
||||
// 比如开机执行一段序列")。单次任务里编序列往往不划算 —— 先 create
|
||||
// 再 run 比直接调工具多两步;真正省事的是**同一套流程反复跑**,
|
||||
// 而开机是最典型的周期性触发点。
|
||||
//
|
||||
// 执行时机:订阅 agent_ready 事件(内核在 kernel assembly → 插件调度
|
||||
// → agent.Start() 之后发布),因此执行时工具已全部注册、agent 已能
|
||||
// 接受输入。
|
||||
//
|
||||
// 声明方式:seq_create 传 boot:true,或序列文件里写 "boot": true。
|
||||
Boot bool `json:"boot,omitempty"`
|
||||
}
|
||||
|
||||
// Group 是一组工具:**组内并行、组间串行**,且拥有独立签名。
|
||||
@ -93,6 +107,11 @@ type rawSeq struct {
|
||||
Name string `json:"name"`
|
||||
Description string `json:"description"`
|
||||
Groups []rawGroup `json:"groups"`
|
||||
// Boot 必须在这里声明,否则 Parse 的 DisallowUnknownFields 会把
|
||||
// "boot": true 当成**拼写错误**拒绝 —— 开机序列根本存不进去。
|
||||
// (这正是 DisallowUnknownFields 的价值:漏声明字段会立刻报错,
|
||||
// 而不是静默忽略后让人纳闷"为什么开机不执行"。)
|
||||
Boot bool `json:"boot"`
|
||||
}
|
||||
|
||||
// Parse 把序列文本解析为 AST 并完成**全部静态校验**。
|
||||
@ -117,7 +136,7 @@ func Parse(data []byte) (*Sequence, error) {
|
||||
return nil, fmt.Errorf("序列 %q 没有任何 group", raw.Name)
|
||||
}
|
||||
|
||||
seq := &Sequence{Name: raw.Name, Description: raw.Description}
|
||||
seq := &Sequence{Name: raw.Name, Description: raw.Description, Boot: raw.Boot}
|
||||
seenGroup := map[string]bool{}
|
||||
for gi, rg := range raw.Groups {
|
||||
g, err := buildGroup(rg)
|
||||
|
||||
@ -58,6 +58,8 @@ type Plugin struct {
|
||||
sdk *sdk.PluginSDK
|
||||
// runner 指向内核(由 Start 注入)
|
||||
runner seqRunner
|
||||
// unsubBoot 是 agent_ready 的退订函数(Stop 时调用)
|
||||
unsubBoot func()
|
||||
// 调用栈深度:跨序列/跨 group 嵌套的**结构上界**(maxCallDepth)
|
||||
callDepth int
|
||||
}
|
||||
@ -82,11 +84,21 @@ func (p *Plugin) Start(s *sdk.PluginSDK) error {
|
||||
p.store = NewStore(dir)
|
||||
p.runner = &kernelRunner{tool: s.Tool()}
|
||||
p.registerTools()
|
||||
|
||||
// 订阅 agent_ready:用它的 boot 序列在**系统真正可用**时自动执行。
|
||||
// 不能在这里直接跑 —— 此刻工具所属插件可能还没拉起、agent 循环还没开始。
|
||||
p.unsubBoot = p.subscribeBoot()
|
||||
return nil
|
||||
}
|
||||
|
||||
// Stop 清理(无订阅需注销)。
|
||||
func (p *Plugin) Stop() error { return nil }
|
||||
// Stop 清理订阅。
|
||||
func (p *Plugin) Stop() error {
|
||||
if p.unsubBoot != nil {
|
||||
p.unsubBoot()
|
||||
p.unsubBoot = nil
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// kernelRunner 把 sdk.ToolAPI 适配成 seqRunner。
|
||||
type kernelRunner struct{ tool sdk.ToolAPI }
|
||||
|
||||
@ -39,21 +39,35 @@ func (p *Plugin) BuiltinSkills() map[string]string {
|
||||
// 它才是注入系统提示词的那一行。
|
||||
const seqSkillMarkdown = `---
|
||||
name: seq-tools
|
||||
description: 把反复出现的同类工具操作编成序列存起来,以后一次执行。
|
||||
description: 把反复出现的同类工具操作编成序列存起来,以后一次执行。仅适合要反复跑同一套流程的批处理。
|
||||
---
|
||||
|
||||
把反复出现的同类工具操作编成「序列」存起来,以后一次执行:组内并行、组间串行,可带条件判断与变量槽引用。
|
||||
把反复出现的同类工具操作编成「序列」存起来,以后一次执行。只适合**要反复跑同一套流程**的批处理场景;一次性的事直接调工具更快。
|
||||
|
||||
## 什么时候用
|
||||
## 什么时候用(**多数时候不该用**)
|
||||
|
||||
**看到下面任一情形,先想一下要不要用序列:**
|
||||
序列是**批量脚本**工具,不是常规调用方式。常见做法是直接调工具,那是对的。
|
||||
|
||||
- 你要连续做几件**顺序已知**的事(例:拉日志 → 过滤 → 汇总 → 发通知)。
|
||||
- 同样一套操作要在**多个对象**上重复(例:对 5 个仓库各跑一次同样的三步检查)。
|
||||
- 某套流程**以前做过**,这次还要再做一遍 —— 先用 ` + "`seq_list`" + ` 看有没有现成的,直接 ` + "`seq_run`" + `。
|
||||
- 一轮里要调的工具很多、参数很长 —— 编成序列可以写成文件(` + "`seq_create`" + ` 的 file 参数),避免参数被长度上限截断。
|
||||
**真的划算只有一种情形:同一套流程要被反复执行。**
|
||||
|
||||
**不适合**:只有一两个工具、或后一步依赖前一步的**实际返回值**(那种要分多轮)。
|
||||
判据是「这套东西**以后还要跑第二遍、第三遍**吗」:
|
||||
|
||||
- ✅ 每天/每周都要做的固定检查(拉日志 → 过滤 → 汇总 → 发通知)
|
||||
- ✅ 同一套操作要在**很多对象**上跑(对 20 个仓库各跑同样四步);对象越多越划算
|
||||
- ✅ 参数很长、容易超长度上限 —— 写成文件(` + "`seq_create`" + ` 的 file 参数)
|
||||
- ✅ 上次已经编过 —— 用 ` + "`seq_list`" + ` 找现成的,` + "`seq_run`" + ` 直接跑
|
||||
|
||||
**不划算(直接调工具更快):**
|
||||
|
||||
- ❌ **只做一次**的事。先 ` + "`seq_create`" + ` 写序列再 ` + "`seq_run`" + `,比直接调多两步
|
||||
- ❌ 只有两三个工具。省不下时间
|
||||
- ❌ 几个工具可以**同批并行**发出(如同时读 5 个文件)—— 那已经是最优路径,
|
||||
包进序列只是多一层壳
|
||||
- ❌ 后一步要依赖前一步的**实际返回值** —— 组内并行拿不到彼此结果,得拆组,不如分多轮
|
||||
|
||||
> 曾经有过「应该多用序列」的说法,但实测不成立:对一次性的多文件读,
|
||||
> 模型直接并行 ` + "`files_read`" + ` 就是最优解,包成序列纯亏。
|
||||
> 所以这份文档的第一句话就是"多数时候不该用"。
|
||||
|
||||
## 核心概念
|
||||
|
||||
|
||||
@ -1,6 +1,7 @@
|
||||
package seq
|
||||
|
||||
import (
|
||||
"strings"
|
||||
"testing"
|
||||
|
||||
"github.com/JianFeeeee/HologramAgent/internal/plugin"
|
||||
@ -24,3 +25,62 @@ func TestSkillProviderContract(t *testing.T) {
|
||||
t.Fatalf("SKILL 正文过短(%d 字节),不足以让模型知道怎么用", len(body))
|
||||
}
|
||||
}
|
||||
|
||||
// ── 使用边界必须写清楚(B 方案的判据)──
|
||||
//
|
||||
// 决策背景:曾试图"让模型多用 seq"(加系统提示词块、加推销式措辞),
|
||||
// e2e 实测证明那是错的方向 —— 对一次性的多文件读,模型并行调
|
||||
// files_read 就是最优解,包成序列纯亏。模型不用它是对的。
|
||||
//
|
||||
// 所以文档改为**直说边界**:多数时候不该用,只在"要反复跑同一套流程"时才划算。
|
||||
// 这条判据守的就是这个立场不被后续"优化"悄悄推回推销腔。
|
||||
func TestSkillStatesNonUseCases(t *testing.T) {
|
||||
body := New("seq").BuiltinSkills()[builtinSkillName]
|
||||
|
||||
// 必须明确写"多数时候不该用"这个立场
|
||||
for _, want := range []string{"多数时候不该用", "不划算", "只做一次"} {
|
||||
if !strings.Contains(body, want) {
|
||||
t.Fatalf("SKILL 必须写清不适用情形(缺 %q)——否则会重新变成推销", want)
|
||||
}
|
||||
}
|
||||
// 必须点出"并行调工具已经最优"这个真实反例(e2e 实测结论)
|
||||
if !strings.Contains(body, "并行") {
|
||||
t.Fatal("SKILL 应点明「几个工具可同批并行时,包进序列只是多一层壳」这个反例")
|
||||
}
|
||||
// 适用判据必须是「要反复跑」而不是「操作多」
|
||||
if !strings.Contains(body, "反复") {
|
||||
t.Fatal("SKILL 的核心判据应是「这套流程要反复跑」,不是「操作数量多」")
|
||||
}
|
||||
}
|
||||
|
||||
// 索引行(模型每轮唯一能看到的一行)也必须带边界,不能只有卖点。
|
||||
func TestSkillIndexLineStatesBoundary(t *testing.T) {
|
||||
body := New("seq").BuiltinSkills()[builtinSkillName]
|
||||
// 取正文首段(内核 extractDescription 的取值方式)
|
||||
lines := strings.Split(body, "\n")
|
||||
inFM, started := false, false
|
||||
desc := ""
|
||||
for i, l := range lines {
|
||||
t := strings.TrimSpace(l)
|
||||
if i == 0 && t == "---" {
|
||||
inFM = true
|
||||
continue
|
||||
}
|
||||
if inFM {
|
||||
if t == "---" {
|
||||
inFM, started = false, true
|
||||
}
|
||||
continue
|
||||
}
|
||||
if started && t != "" && !strings.HasPrefix(t, "#") {
|
||||
desc = t
|
||||
break
|
||||
}
|
||||
}
|
||||
if desc == "" {
|
||||
t.Fatal("取不到索引行(正文首段为空)")
|
||||
}
|
||||
if !strings.Contains(desc, "反复") || !strings.Contains(desc, "一次性") {
|
||||
t.Fatalf("索引行必须同时说清适用与不适用,实际:%q", desc)
|
||||
}
|
||||
}
|
||||
|
||||
@ -42,6 +42,12 @@ func seqToolDefs() []toolDefInfo {
|
||||
"type": "string",
|
||||
"description": "序列文件路径(与 groups 二选一)。目录内文件优先用 files 工具查看",
|
||||
},
|
||||
"boot": map[string]interface{}{
|
||||
"type": "boolean",
|
||||
"description": "标记为开机序列:下次系统就绪时自动执行一次。" +
|
||||
"适合「每次启动都要做的那套固定动作」(自检、预热、拉取状态)。" +
|
||||
"注意是**每次进程启动一次**,不是定时任务。",
|
||||
},
|
||||
},
|
||||
"required": []string{"name"},
|
||||
},
|
||||
|
||||
Reference in New Issue
Block a user