Files
HomeAgent/internal/agent/core/resident_tools.go
JianFeeeee d9c2e51f61 fix(agent): cancel 停不下调度循环 —— 队列非空时永远不看 ctx.Done
## 缺陷(真缺陷,不是测试问题)

`schedulerLoop` 只在"队列空"的分支里 select `ctx.Done()`:

	if kind == nextNone {
		select { ... case <-a.ctx.Done(): return }
	}

于是**只要队列里有任务,循环就永远不看 ctx.Done** —— 一直取、一直执行,
直到队列被取空才可能退出。

生产后果(`reclaim` / `destroy`):销毁流程先回收残余任务、再销毁子。
而回收到销毁之间子仍在运行,继续把队列里的任务取走执行 —— 那些任务的
结果**永远无人认领**(子随即被销毁),且日志无痕迹:既不在"残余清单"里,
也没跑到出结果。带 ResponseCh 的同步调用方(cli/a2a)更糟:Stop 的
drainPendingInterrupts 给它补一个 "skipped",用户看到"已跳过",
而任务其实执行过。

## 修法

1. `schedulerLoop` 循环顶部(**取任务之前**)检查 ctx.Done。
2. 新增 `waitSchedulerStopped`:cancel 之后等 running 归零(上限 5s,
   超时如实记日志,不让控制面永久挂住)。
3. 新增 `FinishResident`:停止调度 → 回收残余 → 处置 → 收割记忆 → 销毁,
   **一步完成**,消除"取完到销毁"之间的窗口。`reclaim` / `destroy` 改走它。
4. 记忆收割抽成 `harvestResidentMemory`,供 FinishResident 的 onDestroy 复用
   (必须在 Destroy 之前 —— Destroy 会关掉 temp)。

## 判据(2 条,都做了"撤销修复必须变红"的验证)

 · TestSchedulerStopsTakingTasksAfterCancelWithNonEmptyQueue
   用阻塞 provider 把循环钉在 executeNewTask 里,此时塞 5 条进队列、
   cancel、放行。修复在位 ⇒ 5 条一条都不执行;撤销修复 ⇒ 取走 5 条。
   **实测撤销后报「仍取走了 5 条(剩 0/5)」,恢复后 10/10 绿。**
 · TestFinishResidentIsAtomic
   断言返回后子已从登记表移除(销毁确实在同一事务内完成)。

原 flaky 的 TestResidualKeepReturnsTasksToParent 与
TestResidualDropNotifiesSyncCaller 也修了:先 cancel + 等静止再入队,
得到确定的初始状态(它们验的是转交逻辑,不是子的生命周期)。30/30 绿。

## 我踩过的两条弯路(都写进判据注释了)

1. 加 `scheduler.freeze` 绕过。错在:调度器本来就有停止语义(ctx),
   再引入一个状态位会让"谁负责停"含糊,且忘了解除就永久僵住。已删。
2. 判据第一版写成「先 Start、再 cancel、然后入队」—— 那时队列是**空的**,
   循环正好走 nextNone 看到 ctx.Done 退出,于是**撤掉产品修复判据依然
   30/30 全绿**。是用户追问"凭什么说修好了"才发现这是假绿。
   必须造出"队列非空时 cancel"这个确切场景才测得到。
2026-10-18 13:07:40 +08:00

215 lines
7.4 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
// 驻留子的**工具面**(对照设计 §7 控制面与 §8 处理表)。
//
// 单工具多动作:父侧一个 `resident_agents`(list/create/send/inspect/compress/reclaim/destroy),
// 子侧两个小工具:`notify_parent`(L3 主动汇报)与 `inputch_note`(主动写处理表)。
import (
"fmt"
"log"
"path/filepath"
"strings"
agentAPI "github.com/JianFeeeee/HomeAgent/internal/agent/api"
agentIO "github.com/JianFeeeee/HomeAgent/internal/agent/io"
"github.com/JianFeeeee/HomeAgent/internal/memory"
)
func strArg(tc agentAPI.ToolCall, key string) string {
s, _ := tc.Arguments[key].(string)
return strings.TrimSpace(s)
}
func splitArg(s string) []string {
if strings.TrimSpace(s) == "" {
return nil
}
parts := strings.Split(s, ",")
out := make([]string, 0, len(parts))
for _, p := range parts {
if p = strings.TrimSpace(p); p != "" {
out = append(out, p)
}
}
return out
}
// executeResidentAgents 是父的驻留子控制面(单工具多动作)。
func (a *Agent) executeResidentAgents(tc agentAPI.ToolCall) string {
switch action := strArg(tc, "action"); action {
case "", "list":
list := a.Residents()
if len(list) == 0 {
return "当前没有驻留子 agent。"
}
var b strings.Builder
fmt.Fprintf(&b, "驻留子 agent(%d 个):", len(list))
for _, r := range list {
fmt.Fprintf(&b, "\n - %s [%s] inputch=%v 轮次=%d 处理表=%d",
r.ID, r.State, r.InputChs, r.Rounds, r.TableSize)
if r.ContextFull {
b.WriteString(" ⚠️ contextfull")
}
}
return b.String()
case "create":
id := strArg(tc, "id")
tempPath := strArg(tc, "temp_path")
if tempPath == "" {
anchor := a.dataDir
if anchor == "" {
// 兜底:从**主图库路径**推导(<data>/memory/graph.db ⇒ <data>)。
// 为什么不静默失败:这条路径只在"配置漏接线"时走到,
// 静默报错会让线上表现为"工具能调但永远建不出来"(实测就是这样)。
if a.memory != nil && a.memory.Path() != "" {
anchor = filepath.Dir(filepath.Dir(a.memory.Path()))
log.Printf("[resident] data_dir 未接线,回退到主图库目录: %s", anchor)
}
}
if anchor == "" {
return "创建驻留子需要 data_dir 或显式 temp_path(内核未接线 DataDir)"
}
tempPath = filepath.Join(anchor, "residents", id, "graph.db")
}
info, err := a.SpawnResident(ResidentOptions{
ID: id,
TaskPrompt: strArg(tc, "task_prompt"),
InputChs: splitArg(strArg(tc, "input_chs")),
AllowedOutputs: splitArg(strArg(tc, "allowed_outputs")),
Capacity: intArg(tc, "capacity"),
TempPath: tempPath,
})
if err != nil {
return fmt.Sprintf("创建驻留子失败: %v", err)
}
return "已创建驻留子: " + MarshalResidentInfo(info)
case "send":
if err := a.SendToResident(strArg(tc, "id"), strArg(tc, "text")); err != nil {
return fmt.Sprintf("发送失败: %v", err)
}
return "已发送(对子而言是 L4 中断)"
case "inspect":
id := strArg(tc, "id")
if id == "" {
return "inspect 需要 id(或先用 action=list)"
}
table, err := a.ResidentTable(id)
if err != nil {
return fmt.Sprintf("查看失败: %v", err)
}
var b strings.Builder
fmt.Fprintf(&b, "驻留子 %s 的 inputch 处理表(%d 条):", id, len(table))
for _, r := range table {
kind := "系统写"
if r.Proactive {
kind = "主动写"
}
fmt.Fprintf(&b, "\n - [%s][%s] %s", r.InputCh, kind, r.Text)
}
if len(table) == 0 {
b.WriteString("\n (尚无记录)")
}
return b.String()
case "compress":
n, err := a.CompressResident(strArg(tc, "id"))
if err != nil {
return fmt.Sprintf("压缩失败: %v", err)
}
return fmt.Sprintf("已压缩子 agent 上下文(丢弃 %d 条旧事件,并发清理其 inputch 处理表);子继续存在", n)
case "reclaim":
id := strArg(tc, "id")
// 残余任务先由父**显式**决定保留还是丢弃(用户 2026-09-19 要求)。
//
// ★ 用 FinishResident 一把做完,不要拆成 ApplyResidual + ReclaimResident:
// 拆开时"解冻到销毁"之间子会继续收新任务,那些任务的结果随子一起丢失
// 且不留痕迹(详见 FinishResident 的说明)。
// onDestroy 里做记忆收割:读子 temp 的活跃三元组并合入主记忆。
// 必须在 Destroy 之前 —— Destroy 会关掉 temp(这正是原先
// ReclaimResident 内部做的顺序)。
var promoted int
residual, rmsg, ferr := a.FinishResident(id, residualPolicyOf(tc),
func(rc *residentChild) error {
p, herr := a.harvestResidentMemory(rc)
if herr != nil {
return fmt.Errorf("合入主记忆失败: %w", herr)
}
promoted = p
return nil
})
if ferr != nil {
return fmt.Sprintf("回收失败: %v", ferr)
}
return fmt.Sprintf("残余任务(%d 条):%s\n已回收(temp 中选中的记录已合入主记忆,promoted=%d,该驻留子已取消)",
residual, rmsg, promoted)
case "destroy":
id := strArg(tc, "id")
// 同上:销毁前先把残余任务交出去,否则它们会随子一起无声消失。
residual, rmsg, ferr := a.FinishResident(id, residualPolicyOf(tc), nil)
if ferr != nil {
return fmt.Sprintf("销毁失败: %v", ferr)
}
return fmt.Sprintf("残余任务(%d 条):%s\n已销毁并移除该驻留子", residual, rmsg)
default:
return fmt.Sprintf("未知 action=%q;可用:list | create | send | inspect | compress | reclaim | destroy", action)
}
}
// reclaimKeepAll 是回收时的默认策略:把子 temp 的活跃记录全部纳入主记忆
// ("哪些纳入"由父的模型决定——这里给的是"全要"这一档)。
func reclaimKeepAll(_ []InputchRecord, triples []memory.Triple) []memory.Triple { return triples }
// residualPolicyOf 从工具参数读残余任务的处置策略(keep/drop)。
//
// 默认 keep:父没明确说丢时,一律转回自己而不是丢弃。
// "宁可多做一件,不可默默丢一条"——吞掉一条输入比多处理一条更糟。
func residualPolicyOf(tc agentAPI.ToolCall) ResidualPolicy {
switch strings.ToLower(strArg(tc, "residual")) {
case "drop", "discard":
return ResidualDrop
default:
return ResidualKeep
}
}
func intArg(tc agentAPI.ToolCall, key string) int {
switch v := tc.Arguments[key].(type) {
case float64:
return int(v)
case int:
return v
}
return 0
}
// executeNotifyParent 是子的"主动向父发消息"(父侧阶梯 = **L3 中断**)。
func (a *Agent) executeNotifyParent(tc agentAPI.ToolCall) string {
return a.notifyParentFrom(strArg(tc, "text"))
}
// executeInputchNote 是子"主动写入本轮 inputch 的处理信息"。
// 主动写过 ⇒ 本轮系统不再自动写(见 autoRecordInputch)。
func (a *Agent) executeInputchNote(tc agentAPI.ToolCall) string {
text := strArg(tc, "text")
if text == "" {
return "text 不能为空"
}
a.recordInputchNote(text)
return "已记录本轮 inputch 处理信息(本轮系统不会再自动写)"
}
// childInboundChannelHint 是给子看的"父会怎么把消息投给你"的提示(不参与调度)。
func childInboundChannelHint(a *Agent) string { return "sub/" + string(a.id) }
// residentTempDir 返回某个驻留子 temp 存储所在目录(销毁时连同目录丢弃)。
func residentTempDir(tempPath string) string { return filepath.Dir(tempPath) }
var _ = agentIO.InputChannel{} // 保持 agentIO 依赖(工具面未来会用通道登记)