Files
HomeAgent/internal/agent/core/residual_freeze_test.go
JianFeeeee a02feaaff7 refactor(agent): 子 agent 类型抽象 + rt 输出异步 + 工具钩子统一
三件事,都是同一个根因的症状:子 agent 的类型逻辑复制成多份,
于是标准路径上的能力没跟着传过来。

## 1. 抽出 child 包(类型抽象 + 单一登记表)

新增 internal/agent/core/child/:
  child.go      Kind / Child(嵌入用)/ State / 残余与通道处置
  registry.go   单一 Registry + Entry 接口 + Cleanup 骨架
  kind.go       ChildKind —— 按两条主轴(上下文 / 工具与输出异步)
  kinds.go      四种子 agent 的清单与迁移状态

此前 Agent 上有两张互不知情的表(residents / rtChildren),各带一把锁。
实测新增一种子 agent 要改 18 个生产文件,并已因复制而漂移过一次
(StopRTChildren 漏注销入站登记,见 engineering-lessons.md §2.3)。

现在单数与复数停止**共用** Cleanup,那个形态在结构上不可能复现。

## 2. rt 的输出改为异步(用户的原始诉求)

rtEmit 曾直接 dev.Execute(...),插件侧延迟(TTS/网络编码/字幕渲染)
直接节流模型的 token 消费 —— 「不攒整段」的收益全被吃掉。

新增 outputpump.go(输出通道的异步壳):
  · Push 立即返回,交付由独立消费 goroutine 完成
  · 有界队列(64),满则**丢最旧**(符合 rt 的"最新者胜")
  · 丢弃与设备缺失都留痕,否则"用户听不到"没有任何线索
  · Close 排空后退出,不吞最后半句
  · 设备每片重新解析(插件崩溃会 Unregister,缓存指针会指向死设备)

## 3. 工具钩子统一(本轮最大发现)

实测:StageBeforeToolcall / StageAfterToolcall 全仓只在
task.go:1022/1079/1194,三处都在 runOneTool / stepToolAfter 里;
而 executeToolCallOutcome 内部**没有任何 runStage**。

⇒ rtchild 与工作式子两条路径**完全跳过插件钩子**:不能审计、
不能拒绝、不能改参、不能脱敏。README 承诺的 before_toolcall
在这两条路径上不成立。

新增 toolbatch.go(唯一批执行实现:批并发 + 双钩子 + 顺序落结果),
两种类型切过去。工作式子的"禁输出工具"与"保持串行"用 batchOpts 注入,
不各写一遍执行逻辑。

## 判据(均经变异验证)

  TestRTChild_OutputIsAsync              改回同步 ⇒ 红(实测确认)
  TestOutputPump_DropsOldestOnFull       改阻塞/无界 ⇒ 红
  TestOutputPump_CloseDrains             丢弃队列 ⇒ 红
  TestOutputPump_ResolvesDeviceEachTime  缓存指针 ⇒ 红
  TestRTChild_ToolsGoThroughPluginStages 绕开钩子 ⇒ 红
  TestRTChild_StageCanDenyTool           同上
  TestWorkerToolsGoThroughPluginStages   同上(变异实测确认)
  TestWorkerToolsStaySerial              改成并发 ⇒ 红
  TestNoChildTypeTablesInCore            源码扫描,禁类型专属表回潮
  TestRTChild_HasGraphMemory 等 5 条     rt 接入图记忆

## 顺带修复

  · rt 接入图记忆(参照驻留子的轻量内核形态):此前 graphMem() == nil,
    memory_recall / memory_commit 在 rt 里全部不可达,每通电话从零开始。
    默认只读(通话高频写入会灌满图谱噪声),4 个失败点合并成一个 rollback
    闭包(逐点挂必然漏一处 ⇒ 泄 SQLite 句柄)。
  · rt 的 PreferredProvider 走候选链(此前裸调 a.provider,优先源无效且零回退)。

## 保留的语义(逐项确认仍在)

  驻留子三层压缩:checkContextFull / handleChildContextFull / CompressResident
  三种通道处置:ChannelRelease / Suspend / Takeover
  残余 keep/drop + harvestResidentMemory
  child.Start() 的 7 个 goroutine

验证:build / vet / gofmt / go test -race 全绿;
且在干净检出(git worktree)上单独 build + test 通过。
2026-11-04 08:21:15 +08:00

150 lines
5.7 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
// 判据:`cancel` 之后调度循环**不再取新任务**(2026-10-18 修的真缺陷)。
//
// ## 缺陷
//
// `schedulerLoop` 原来只在"队列空"的分支里 select `ctx.Done()`:
//
// if kind == nextNone {
// select { ... case <-a.ctx.Done(): return }
// }
//
// 于是**只要队列里有任务,循环就永远不看 ctx.Done** —— 一直取、一直执行,
// 直到队列被取空才可能退出。
//
// 后果(生产路径 `reclaim`/`destroy`):销毁流程先取残余任务、再销毁子,
// 而取完到销毁之间子仍在运行,继续把队列里的任务取走执行 —— 那些任务的
// 结果**永远无人认领**(子随即被销毁),且日志无痕迹:既不在"残余清单"里,
// 也没跑到出结果。
//
// ## 判据必须造出"队列非空时 cancel"这个确切场景
//
// 我第一版判据犯过两个错,都记在这里:
//
// 1. 加了 `scheduler.freeze` 绕过 —— 调度器本来就有停止语义(ctx),
// 再引入一个状态位会让"谁负责停"含糊,且忘了解除就永久僵住。已删。
// 2. 判据写成「先 Start、再 cancel、然后入队」。但那时队列是**空的**,
// 循环正好走 nextNone 分支看到 ctx.Done 退出 —— 于是**撤掉产品修复
// 判据依然通过**(实测 30/30 全绿)。那是假绿,测的是"空队列能退出",
// 而不是真正的缺陷("非空队列不会退出")。
//
// 现在的做法:让**第一条任务正在执行且阻塞住**(provider 卡在 Chat 里),
// 于是循环停在 executeNewTask 内;此时再塞入若干任务并 cancel、放行。
// 若修复在位 ⇒ 循环处理完当前任务后看到 ctx.Done 就退出,队列里的新任务
// **一条都不会被执行**。若修复不在 ⇒ 它会继续把队列取空。
import (
"context"
"testing"
"time"
agentAPI "github.com/JianFeeeee/HologramAgent/internal/agent/api"
agentIO "github.com/JianFeeeee/HologramAgent/internal/agent/io"
)
// blockingProvider 的 Chat 会一直阻塞,直到 release 被关闭。
//
// 用途:把调度循环**钉在 executeNewTask 里**,从而在"队列非空 + 任务正在跑"
// 这个确切时刻做 cancel,复现原缺陷的场景。
type blockingProvider struct {
entered chan struct{} // 第一次进入 Chat 时关闭(供判据等待)
release chan struct{} // 关闭后 Chat 返回
}
func (p *blockingProvider) Name() string { return "blocking" }
func (p *blockingProvider) Chat(ctx context.Context, req *agentAPI.CompletionRequest) (*agentAPI.CompletionResponse, error) {
select {
case <-p.entered:
default:
close(p.entered)
}
<-p.release
return &agentAPI.CompletionResponse{Content: "ok"}, nil
}
func (p *blockingProvider) ChatStream(ctx context.Context, req *agentAPI.CompletionRequest) (<-chan agentAPI.StreamChunk, error) {
return nil, context.Canceled
}
func (p *blockingProvider) MaxContextTokens() int { return 8192 }
// ★★★ 判据:队列非空时 cancel ⇒ 剩余任务一条都不被执行。
func TestSchedulerStopsTakingTasksAfterCancelWithNonEmptyQueue(t *testing.T) {
prov := &blockingProvider{
entered: make(chan struct{}),
release: make(chan struct{}),
}
a := New(AgentConfig{
ID: "sched-nonempty",
Provider: prov,
ProviderManager: agentAPI.NewProviderManager(),
IO: agentIO.NewIOManager(),
StageHost: NewStageHost(),
DataDir: t.TempDir(),
})
defer a.Stop()
// ① 第一条:让循环进去执行并卡在 Chat 里。
a.sched.enqueue(newInputTask(&agentIO.InputEvent{
RequestID: "running", Source: "cli",
Payload: map[string]interface{}{"content": "hi"},
}))
a.Start()
select {
case <-prov.entered:
case <-time.After(5 * time.Second):
t.Fatal("provider 未被调用 —— 调度循环没跑起来")
}
// ② 此刻:循环卡在 executeNewTask(running 非空),再塞 5 条进队列。
const pending = 5
for i := 0; i < pending; i++ {
a.sched.enqueue(newInputTask(&agentIO.InputEvent{
RequestID: "pending", Source: "qq",
Payload: map[string]interface{}{"content": "x"},
}))
}
// ③ 在这个时刻 cancel(**队列非空** —— 这正是原缺陷的场景),然后放行。
a.cancel()
close(prov.release)
// ④ 给它足够时间:修复在位时循环处理完当前任务即退出;
// 修复缺失时它会把 5 条全部取走执行。
time.Sleep(1200 * time.Millisecond)
if got := a.sched.queueLen(); got != pending {
t.Fatalf("★ 队列非空时 cancel 后仍取走了 %d 条(剩 %d / %d)——\n"+
" 调度循环没有在取任务前检查 ctx.Done。\n"+
" 生产后果:reclaim/destroy 取完残余到销毁之间,子继续执行那些任务,\n"+
" 结果随销毁永久丢失且日志无痕迹。",
pending-got, got, pending)
}
}
// ★★★ 判据:FinishResident 把「停止调度 → 回收残余 → 销毁」做成一次事务。
//
// 它防的是:调用方拆成 ApplyResidual + Destroy 两步时,中间那段窗口里
// 子仍在收新任务。断言"返回后子已从登记表移除",即销毁确实在同一事务内完成。
func TestFinishResidentIsAtomic(t *testing.T) {
root, main := newRootWithoutSchedulerLoop(t)
defer main.Close()
defer root.Stop()
info, err := root.SpawnResident(ResidentOptions{
ID: "res-fin", TempPath: root.residentTempPath("res-fin"),
})
if err != nil {
t.Fatal(err)
}
if _, _, err := root.FinishResident(info.ID, ResidualKeep, nil); err != nil {
t.Fatalf("FinishResident 失败: %v", err)
}
still := root.residentOf(info.ID) != nil
if still {
t.Fatal("★ FinishResident 之后子仍在登记表 —— 销毁没被纳入同一事务," +
"取残余到销毁之间的窗口仍存在")
}
}