Files
HomeAgent/internal/plugins/timer/plugin.go
JianFeeeee 2232d5483c feat(parallel): 并发安全改为声明式,并审计标注 37 个工具
把"能不能并发"从内核硬编码名单改成**工具自己的声明项**,形态照 SDK 的
NoMemory 走。

## ★ 起因:提示词在跟内核不一致

阶段 2.5 写进提示词的「内核默认并行执行」当时是**假的**:toolParallelSafe
只查 stageHost 与 io 两个来源,而全仓 ParallelSafe:true 的生产代码数量
是 **0**。于是除碰巧只发一个工具外,每一批都整批串行回退,而提示词正教
模型把多个查询放同一轮。**内核行为与提示词不一致 = 对模型说谎。**

并发面:0 → 37 个工具(18 插件 ParallelSafe + 19 插件 Serial + 9 内置只读)。

## 声明形态(照 SDK,不自创)

### 插件:结构体字段
    s.RegisterTool("config_get", sdk.ToolDef{
        Name: ..., Description: ...,
        Parameters: map[string]interface{}{...},
        // 已核实只读:…
        ParallelSafe: true,      ← 插在 Parameters 之后、handler 之前
    }, p.handleGet(s))

位置与 SDK 的 NoMemory/ContextPolicy/RecallPolicy 一致:Name 在首位,
声明项在末尾,不打散 gofmt 对齐。

### 新增 SDK 声明项:ToolDef.Serial
ParallelSafe 的**反向**标记,判据优先级高于 ParallelSafe。
为什么需要:ParallelSafe 零值 false 已表达"安全",插件无法区分"我没想过"
与"我确认过必须串行"。没有这个区分,工具作者只能靠命名约定传递意图。

内核已消费它(io.ToolDef 同步加字段对齐),并有判据守"Serial 胜出"。

### 内置工具:toolDef 的 toolParallel 选项
内置工具以裸 schema map 下发,没有 ToolDef 结构,所以用变参选项:
    toolDef(名字, 描述, 属性)                  // 默认串行
    toolDef(名字, 描述, 属性, "toolParallel")  // 已核实只读,可并发
读工具表的老调用点一行不用动,声明就写在工具定义那一行。

## ★ 走过的弯路(都留了判据)

1. **硬编码白名单**:先在 toolParallelSafe 里查一张
   builtinParallelSafeTools map。那把声明从"工具自己"搬回了内核 ——
   工具改名/新增不会自动跟着变,得靠一条 grep 源码的判据才能发现漂移,
   而判据一改就忘。已删,改为从定义读。

2. **判据前提错(同一个坑踩了两次)**:拿裸 &Agent{} 的 buildToolDefs 输出
   当"实际可见工具",但这 9 个内置工具全在条件分支里(a.knowledge != nil /
   a.social != nil / a.parentID != ""…),裸 Agent 一个都不产出 ⇒ 全部误报
   "声明形同虚设"。第一次叫它"幽灵条目",没认出是同一个坑。

3. **注释模仿真实签名污染判据**:toolParallel 的用法注释写着
   `toolDef("knowledge_search", ...)`,判据按文本匹配先撞上注释。

4. **buildToolDefs 的 nil 不一致**:开头判了 a.io != nil,末尾却无条件
   a.io.ListChannels()。任何无 IO 的 Agent 调它都 panic —— 而 panic 报在
   io 包里,根因在 tooldefs.go。已补。

5. **插入脚本用正则找"最后一个顶层字段"**:被嵌套 map 里的同形文本骗到,
   823 处错误重排把文件改坏。改用括号深度 + 记录进入深度 3 的行号
   (空 properties 会让深度在同一行进出平衡,只判 depth==2 不够)。
   工具在 SDK 仓 tools/annotate_parallel/,复用时用绝对路径。

## 提示词措辞同步修正
「默认并行执行」→「尽量并发执行,但这是**逐工具判断**的」,并教模型
**把查询类放同一轮、写操作单独发一轮**(写和查混在一批,整批都串行)。

## 判据
- TestSerialOverridesParallelSafe          Serial 优先于 ParallelSafe
- TestToolParallelDeclarationsAudit        并发面不许再归零
- TestNoToolDeclaresBothParallelAndSerial  两者同标即谎话
- TestBuiltinParallelDeclaredWhereDefined  声明写在定义处、且内核真读到
- TestStoreListIgnoresForeignJSON          压测抓到的 List() 缺陷
2026-09-27 15:15:57 +08:00

140 lines
3.9 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 timer
import (
"fmt"
"log"
"sync"
"time"
"gitcode.com/JianFeeeee/HomeAgent/internal/plugin"
sdk "gitcode.com/JianFeeeee/HomeAgent/internal/sdk"
)
func init() {
plugin.RegisterPluginMeta("timer", "定时任务", "Timer")
plugin.RegisterFactory("timer", func(name string, config map[string]interface{}) (sdk.Plugin, error) {
return New(name), nil
})
}
type Plugin struct {
name string
mu sync.Mutex
wg sync.WaitGroup
stopCh chan struct{}
maxDur time.Duration
}
type timerTask struct {
id int
dur time.Duration
message string
doneAt time.Time
s *sdk.PluginSDK
}
func New(name string) *Plugin {
return &Plugin{name: name, stopCh: make(chan struct{})}
}
func (p *Plugin) Name() string { return p.name }
func (p *Plugin) Start(s *sdk.PluginSDK) error {
s.SetAutoRestart(true)
p.maxDur = 24 * time.Hour
// 停止清理(取消倒计时)交由 stop handler:内核在调用 Stop() 之前执行。
s.RegisterStopHandler(func() { close(p.stopCh) })
// 删除清理:移除插件自身配置(删除专用回调,重载不触发)。
s.RegisterOnRemoveHandler(func() {
if err := s.Settings().Remove("max_duration"); err != nil {
log.Printf("[timer] onRemove cleanup: %v", err)
}
})
s.Settings().RegisterDef(sdk.ConfigDef{
Key: "max_duration", Type: "string", DisplayName: "最大定时时长",
Description: "允许设置的最大定时时长,例如 24h, 7d, 1h(默认 24h)",
Default: "24h",
})
if v, _ := s.Settings().Get("max_duration"); v != nil {
if s, ok := v.(string); ok && s != "" {
if d, err := time.ParseDuration(s); err == nil && d > 0 {
p.maxDur = d
}
}
}
// timer 通道:定时器到点经它注入 agent(见本文件 InjectInterruptTextOpts 调用)。
_ = s.RegisterInputChannel("timer", sdk.ChannelDef{})
s.RegisterTool("timer_set", sdk.ToolDef{
Name: "timer_set",
Description: "设置一个定时提醒。倒计时结束后通过中断通道通知 agent。",
Parameters: map[string]interface{}{
"type": "object",
"properties": map[string]interface{}{
"duration": map[string]interface{}{
"type": "string",
"description": "持续时间,例如 5s, 2m, 1h",
},
"message": map[string]interface{}{
"type": "string",
"description": "提醒内容",
},
},
"required": []string{"duration", "message"},
},
// 写定时器:改 p.timers(持 p.mu),按序更可预期
Serial: true,
}, func(args map[string]interface{}) (interface{}, error) {
durStr, _ := args["duration"].(string)
message, _ := args["message"].(string)
if durStr == "" {
return map[string]interface{}{"error": "duration is required"}, nil
}
if message == "" {
return map[string]interface{}{"error": "message is required"}, nil
}
dur, err := time.ParseDuration(durStr)
if err != nil {
return map[string]interface{}{"error": fmt.Sprintf("invalid duration %q: %v", durStr, err)}, nil
}
if dur > p.maxDur {
return map[string]interface{}{"error": fmt.Sprintf("duration %v exceeds max %v", dur, p.maxDur)}, nil
}
p.mu.Lock()
p.wg.Add(1)
p.mu.Unlock()
go func() {
defer p.wg.Done()
select {
case <-time.After(dur):
log.Printf("[timer] firing: %s (%s later)", message, dur)
// NoMemory:定时提醒是系统通知,不是记忆内容。
// PriorityL3:定时器是“时钟那种实时工作”——到点就该处理,
// 比 QQ 那类可无限等待的异步消息高(L3 vs L1)。
s.InjectInterruptTextOpts("timer", "timer", fmt.Sprintf("timer: %s", message),
sdk.InjectOptions{NoMemory: true, Priority: sdk.PriorityL3})
case <-p.stopCh:
log.Printf("[timer] cancelled: %s", message)
}
}()
doneAt := time.Now().Add(dur)
return map[string]interface{}{
"status": "timer_set",
"duration": durStr,
"message": message,
"done_at": doneAt.Format(time.RFC3339),
}, nil
})
return nil
}
func (p *Plugin) Stop() error {
p.wg.Wait()
return nil
}