mirror of
https://gitcode.com/JianFeeeee/HomeAgent.git
synced 2026-11-06 01:07:35 +00:00
fix(rtchild): 同名复用串线 + 批量停止漏注销
三处缺陷，全部在「回收/复用」这条线上 —— 单实例 e2e 天然测不到：
1. RTTranscript 读序（主线）
原实现先查留档、命中即返回。留档以 id 为键而 id 由模型给，于是
回收后同名重建，rt_transcript 返回的永远是上一通电话的原文，
且完全不报错。改为「在跑实例优先，无活实例才落回留档」。
2. StopRTChildren 漏注销入站登记
单数路径会 Unregister(child/rt-<id>)，复数路径没有。登记表共享、
按 name 全局唯一，残留随 create/destroy 单调累积，而外部没有任何
工具能单独注销一条 inputch。
3. 撤回一个错误的修法
一度在 SpawnRTChild 里 delete 留档。变异测试证明删掉它判据仍绿
（治不了串线），且有害：留档的意义正是「回收后父按需取原文」，
提前删掉会造成「推了信号却拉不到正文」。现仅在 CloseRTChild 写
同名键时自然覆写。
判据（均通过变异验证，去掉修复必红）：
- internal/agent/core/rtchild_reuse_test.go（进程内，5 条）
- internal/plugins/rtchild_reuse_e2e_test.go（跨进程真插件，3 条）
★ 写判据时踩到两个坑，一并记在注释里：
- 前两版判据假绿（先塞留档被 Spawn 清理掉 / 新实例不 push 导致
「读到旧内容」与「读到空」都是 len==0），靠变异才发现没测到真东西
- 等 Rounds 不等于历史已入账（rtRunOnce 先 round++ 后 rtAppendHistory），
与 dff745c 同一个坑；已改为等 history 条数
This commit is contained in:
JianFeeeee
@ -77,6 +77,14 @@ type Agent struct {
|
||||
// 为什么必须留档：交接是「推信号 + 拉状态」两步（见 CloseRTChild），
|
||||
// 推完中断后实例已出登记表 —— 若不留档，主 agent 收到"可调用 rt_transcript
|
||||
// 取回原文"的提示时，原文已经取不到了。
|
||||
//
|
||||
// ★ 键是 id，所以**同名复用会撞键**。串线的真因在读侧：RTTranscript 曾
|
||||
// "先查留档、命中即返回"，于是新实例读到的永远是上一通已回收的原文
|
||||
// （2026-10-30 实测复现）。现在改为**在跑实例优先**（见 RTTranscript）。
|
||||
//
|
||||
// ★ 刻意**不**在 Spawn 时 delete 这份留档：留档存在的意义就是"回收后
|
||||
// 主 agent 按需取回原文"，提前删掉等于让推了信号却拉不到正文。
|
||||
// 同名键只在下一次 CloseRTChild 写入时被自然覆写。
|
||||
rtClosedTranscripts map[string][]agentAPI.Message
|
||||
// rtSlot 是**本 agent 作为 rtchild 时**的输入槽（作为父时无用）。
|
||||
rtSlot *rtSlot
|
||||
|
||||
@ -174,6 +174,21 @@ func newRTSlot() *rtSlot {
|
||||
return &rtSlot{wake: make(chan struct{}, 1)}
|
||||
}
|
||||
|
||||
// ── 同名复用的清理纪律 ──
|
||||
//
|
||||
// rtchild 的两张表都以 **id 字符串**为键，而 id 由调用方（模型）给。
|
||||
// 于是同一个 id 被回收后重建（反复"拨同一通电话"）时，上一代留下的
|
||||
// 东西会被当成这一代的。两个具体的串线点（都有判据钉住）：
|
||||
//
|
||||
// ① rtClosedTranscripts[id] —— CloseRTChild 写入，读侧原本"先查留档、
|
||||
// 命中即返回" ⇒ 同名重建后新实例的 rt_transcript 返回上一代已回收的原文。
|
||||
// 修法：RTTranscript 改为**在跑实例优先**（留档只在没有活实例时才用）。
|
||||
// ② inboundChannelName(id) —— StopRTChildren（复数）漏了注销，
|
||||
// 与 StopRTChild（单数）不对称。
|
||||
// 修法：复数路径补上同一步注销。
|
||||
//
|
||||
// 两处都修在真正出错的那一侧（一个在读、一个在停），**不引入代际号**
|
||||
// 之类的额外抽象：问题本身只需两句定位准确的改动。
|
||||
// Push 覆盖写入一条输入，返回本次的 seq。
|
||||
//
|
||||
// 旧值**直接丢弃**（不是排队、不是补偿）—— 这正是契约 1/3 的实现。
|
||||
@ -564,6 +579,22 @@ func (a *Agent) SpawnRTChild(opts RTChildOptions) (string, error) {
|
||||
|
||||
a.rtMu.Lock()
|
||||
a.rtChildren[opts.ID] = rc
|
||||
// ★ 这里**不**清留档（与一个曾被写错一版的修复相反，记录以示后人）。
|
||||
//
|
||||
// 我一度在这里加过 `delete(a.rtClosedTranscripts, opts.ID)`，理由是
|
||||
// "同名重开就该让旧留档失效"。实测后撤回，两个原因：
|
||||
//
|
||||
// ① **它治的不是那个病**。串线的真因在**读取顺序**（RTTranscript
|
||||
// 曾先查留档、命中即返回），已在那边修好。把这里改回保留留档，
|
||||
// 判据 TestRTChild_ReuseAfterCloseKeepsTranscriptsApart 仍然绿
|
||||
// —— 反证它不解决问题。
|
||||
// ② **它本身有害**。留档的全部意义就是"回收后、主 agent 按需取回
|
||||
// 原文"（交接 = 推信号 + 拉状态两步）。在 Spawn 时删掉，等于在
|
||||
// 父还没拉之前就把上一通的正文抹了 —— 正是 CloseRTChild 注释里
|
||||
// 反复强调要避免的"推了信号但拉不到正文"。
|
||||
//
|
||||
// 现在的语义：留档只在**下一次 CloseRTChild 写同名键**时被覆写
|
||||
// （自然淘汰），在那之前一直可取。
|
||||
a.rtMu.Unlock()
|
||||
|
||||
go carrier.rtLoop(rc)
|
||||
@ -618,6 +649,13 @@ func (a *Agent) StopRTChild(id string) error {
|
||||
}
|
||||
|
||||
// StopRTChildren 停止全部 rtchild（父退出时必须；不留孤儿）。
|
||||
//
|
||||
// ★ 必须与 StopRTChild（单数）**行为对称**：那个会注销入站登记
|
||||
// （child/rt-<id>），本函数曾漏了这一步。
|
||||
//
|
||||
// 漏的后果与驻留子那边记录的是同一个形态：登记表是共享的、按 name 全局唯一，
|
||||
// 残留随 create/destroy 次数**单调累积**，而外部没有任何工具能单独注销
|
||||
// 一条 inputch —— 只能重启 homed 清。
|
||||
func (a *Agent) StopRTChildren() int {
|
||||
a.rtMu.Lock()
|
||||
all := make([]*rtChild, 0, len(a.rtChildren))
|
||||
@ -632,6 +670,15 @@ func (a *Agent) StopRTChildren() int {
|
||||
rc.mu.Unlock()
|
||||
rc.agent.cancel()
|
||||
_, _, _ = rc.slot.Take()
|
||||
// 注销入站登记（与 StopRTChild 的单数路径逐字一致）。
|
||||
//
|
||||
// 不碰插件声明的 rt 通道本身（那两条销毁后**悬空**，用户裁定）：
|
||||
// 这里只摘本内核自己建的 child/rt-<id>。
|
||||
if a.io != nil {
|
||||
if reg := a.io.ChannelRegistry(); reg != nil {
|
||||
reg.Unregister(inboundChannelName("rt/" + rc.id))
|
||||
}
|
||||
}
|
||||
}
|
||||
if len(all) > 0 {
|
||||
log.Printf("[rtchild] stopped all %d instance(s)", len(all))
|
||||
@ -1238,6 +1285,9 @@ func (a *Agent) CloseRTChild(id string) (int, error) {
|
||||
}
|
||||
|
||||
// ① 先取全文规模（此时实例还活着，history 完整）。
|
||||
//
|
||||
// ★ 必须在 Stop 之前取：Stop 会把实例从登记表摘掉，之后
|
||||
// RTTranscript 就只能走留档分支了。
|
||||
transcript, _ := a.RTTranscript(id)
|
||||
text := FormatRTTranscript(transcript)
|
||||
n, chars := len(transcript), len(text)
|
||||
@ -1252,8 +1302,11 @@ func (a *Agent) CloseRTChild(id string) (int, error) {
|
||||
|
||||
// ③ 推 L4 信号（contextfull 同款），正文留在 RTTranscript 里等按需拉取。
|
||||
//
|
||||
// ★ 实例已出登记表，但原文必须仍可取 —— 所以把 transcript 存进
|
||||
// a.rtTranscripts 留档（见 RTTranscript 的读取顺序）。
|
||||
// ★ 实例已出登记表，但原文必须仍可取 —— 所以留档（见 RTTranscript 的读取顺序）。
|
||||
//
|
||||
// ★★ 同名键在这里**自然覆写**（2026-10-30）：留档以 id 为键，同名回收
|
||||
// 会替换掉上一次的留档。串线问题不在写侧而在**读侧**（见 RTTranscript
|
||||
// 的"在跑实例优先"），所以这里不需要额外清理动作。
|
||||
a.rtMu.Lock()
|
||||
if a.rtClosedTranscripts == nil {
|
||||
a.rtClosedTranscripts = map[string][]agentAPI.Message{}
|
||||
@ -1268,29 +1321,37 @@ func (a *Agent) CloseRTChild(id string) (int, error) {
|
||||
|
||||
// RTTranscript 返回一个 rtchild 的**完整**上下文原文（不过窗口裁剪）。
|
||||
//
|
||||
// 读顺序：先查**已回收留档**（CloseRTChild 保存的），再查在用实例的 history。
|
||||
// 后者保证"通话进行中也能看"，前者保证"回收之后仍能取" —— 两者缺一，
|
||||
// 「回收时向主 agent 展示全部上下文」就会在某些时序下拿不到原文。
|
||||
// ## 读顺序（2026-10-30 修：留档不得遮蔽**在跑的新一代**）
|
||||
//
|
||||
// 原实现是「先查留档，有就返回」，于是同名重建后**新实例的对话永远读不到**：
|
||||
// 留档一直命中，`rt_transcript` 返回的是上一代已回收的原文（实测复现：
|
||||
// 新实例读出的第一条是上一通的对话）。现在按**在跑实例优先**：
|
||||
//
|
||||
// 在跑实例存在 ⇒ 返回它的 history（通话进行中也能看）
|
||||
// 否则查留档 ⇒ 返回该代留档（回收之后仍能取）
|
||||
//
|
||||
// 两条都保留，缺一条就会在某个时序下拿不到原文。二者的先后由
|
||||
// "实例还在不在" 决定，而不是由"留档里有没有" 决定。
|
||||
func (a *Agent) RTTranscript(id string) ([]agentAPI.Message, bool) {
|
||||
a.rtMu.Lock()
|
||||
archived, hasArchived := a.rtClosedTranscripts[id]
|
||||
rc := a.rtChildren[id]
|
||||
archived := a.rtClosedTranscripts[id]
|
||||
a.rtMu.Unlock()
|
||||
|
||||
if hasArchived {
|
||||
if rc != nil {
|
||||
rc.mu.Lock()
|
||||
// 深拷贝：调用方可能长期持有，不能被后续轮次 append 影响。
|
||||
out := make([]agentAPI.Message, len(rc.history))
|
||||
copy(out, rc.history)
|
||||
rc.mu.Unlock()
|
||||
return out, true
|
||||
}
|
||||
if archived != nil {
|
||||
out := make([]agentAPI.Message, len(archived))
|
||||
copy(out, archived)
|
||||
return out, true
|
||||
}
|
||||
if rc == nil {
|
||||
return nil, false
|
||||
}
|
||||
rc.mu.Lock()
|
||||
defer rc.mu.Unlock()
|
||||
// 深拷贝：调用方可能长期持有，不能被后续轮次 append 影响。
|
||||
out := make([]agentAPI.Message, len(rc.history))
|
||||
copy(out, rc.history)
|
||||
return out, true
|
||||
return nil, false
|
||||
}
|
||||
|
||||
// ── rt_transcript：主 agent 按需拉取 rtchild 的完整上下文 ──
|
||||
|
||||
224
internal/agent/core/rtchild_reuse_test.go
Normal file
224
internal/agent/core/rtchild_reuse_test.go
Normal file
@ -0,0 +1,224 @@
|
||||
package core
|
||||
|
||||
// rtchild 的**同名复用**与**批量停止**的回收纪律（2026-10-30 修）。
|
||||
//
|
||||
// ## 这一组判据锁的是什么
|
||||
//
|
||||
// rtchild 的留档表与通道登记表都以 **id 字符串**为键，而 id 由模型给。
|
||||
// 「拨完一通、再拨同名的一通」是完全正常的使用方式，但它会让上一代
|
||||
// 留下的东西被当成这一代的 —— 而且**完全不报错**，只是内容错了。
|
||||
//
|
||||
// 三条缺陷、三条判据、每条都能变红（变异方式写在各自注释里）：
|
||||
//
|
||||
// ① 留档跨代串线 —— rt_transcript 读到上一通的对话
|
||||
// ② 批量停止漏注销 —— child/rt-<id> 登记单调累积（只能重启 homed 清）
|
||||
// ③ 留档遮蔽在跑实例 —— 通话进行中 rt_transcript 读不到自己的历史
|
||||
//
|
||||
// ③ 与 ① 是同一个读路径的两端：修复前 RTTranscript 先查留档，只要留档
|
||||
// 命中就返回。于是不仅"重建后读到老的"，连"老的在跑、留档还在"时也会
|
||||
// 读到旧的那份。
|
||||
|
||||
import (
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
agentAPI "github.com/JianFeeeee/HologramAgent/internal/agent/api"
|
||||
)
|
||||
|
||||
// TestRTChild_ReuseAfterCloseKeepsTranscriptsApart 钉住①：
|
||||
// 同名 rtchild 回收后重建，新实例的 rt_transcript **不得**返回上一代原文。
|
||||
//
|
||||
// 变异方式：把 RTTranscript 改回"先查留档、命中即返回"（archive-first）
|
||||
// ⇒ 本判据必红（新实例读出的第一条是"上一通的原话"）。已实测。
|
||||
func TestRTChild_ReuseAfterCloseKeepsTranscriptsApart(t *testing.T) {
|
||||
p := &rtLongHistoryProvider{}
|
||||
host, _ := newRTHost(t, p)
|
||||
|
||||
const id = "reuse"
|
||||
first := rtSpawn(t, host, id)
|
||||
host.PushRTInput(first, "上一通的原话")
|
||||
waitFor(t, "上一通第一轮完成", func() bool { return p.rounds() >= 1 })
|
||||
|
||||
if n, err := host.CloseRTChild(id); err != nil {
|
||||
t.Fatalf("回收上一通失败：%v", err)
|
||||
} else if n == 0 {
|
||||
t.Fatal("上一通没有交接任何消息 —— 判据前提不成立")
|
||||
}
|
||||
// 回收后留档必须仍在（这是「回收后原文仍可取」的契约）。
|
||||
if got, ok := host.RTTranscript(id); !ok || len(got) == 0 {
|
||||
t.Fatalf("回收后取不到原文，留档机制本身坏了（ok=%v len=%d）", ok, len(got))
|
||||
}
|
||||
|
||||
// 同一个 id 再开一通 —— 这是真实用法（模型常常复用 id）。
|
||||
second := rtSpawn(t, host, id)
|
||||
|
||||
got, ok := host.RTTranscript(id)
|
||||
if !ok {
|
||||
t.Fatal("重建后的实例取不到 transcript")
|
||||
}
|
||||
for _, m := range got {
|
||||
if m.Content == "上一通的原话" {
|
||||
t.Fatalf("★ 同名重建后 rt_transcript 读到了**上一通**的原话 —— 两通电话串线")
|
||||
}
|
||||
}
|
||||
if len(got) != 0 {
|
||||
t.Fatalf("新一代刚创建、还没说话，transcript 应为空；实际 %d 条：%+v",
|
||||
len(got), truncateStr(FormatRTTranscript(got), 200))
|
||||
}
|
||||
|
||||
// 新一代说自己的话，必须只看到自己的。
|
||||
p.mu.Lock()
|
||||
base := len(p.reqs)
|
||||
p.mu.Unlock()
|
||||
host.PushRTInput(second, "这一通的原话")
|
||||
waitFor(t, "这一通第一轮完成", func() bool {
|
||||
p.mu.Lock()
|
||||
defer p.mu.Unlock()
|
||||
return len(p.reqs) > base
|
||||
})
|
||||
|
||||
got, _ = host.RTTranscript(id)
|
||||
if len(got) == 0 || got[0].Content != "这一通的原话" {
|
||||
t.Fatalf("新一代的历史第一条应是「这一通的原话」，实际：%s",
|
||||
truncateStr(FormatRTTranscript(got), 200))
|
||||
}
|
||||
}
|
||||
|
||||
// TestRTTranscript_PrefersLiveInstanceOverArchive 钉住③：
|
||||
// 通话**进行中**（实例在跑）时，rt_transcript 必须返回在跑实例的历史，
|
||||
// 而不是该 id 上一代的留档。
|
||||
//
|
||||
// ## 为什么要先 Spawn 再注入留档（两版假绿的教训）
|
||||
//
|
||||
// 这个判据前两版都是**假绿**的，教训值得钉在注释里：
|
||||
//
|
||||
// 版本 A：先手塞留档再 Spawn → 活实例建立时留档已存在，但当时判据
|
||||
// 断言过弱（只查"没有旧内容"），变异下仍绿。
|
||||
// 版本 B：真回收再重建，但新实例不 push（history 空）
|
||||
// ⇒ "读到旧留档"与"读到空"表面都是 len==0 ⇒ 仍然假绿。
|
||||
//
|
||||
// 现在：先 Spawn 建活实例 → **再**注入同 id 的陈旧留档 → 再 push 一句。
|
||||
// 这样两种实现才分得清楚（活实例优先 / 留档优先）。
|
||||
func TestRTTranscript_PrefersLiveInstanceOverArchive(t *testing.T) {
|
||||
p := &rtLongHistoryProvider{}
|
||||
host, _ := newRTHost(t, p)
|
||||
|
||||
// ① 先建活实例。
|
||||
const id = "live"
|
||||
rtSpawn(t, host, id)
|
||||
|
||||
// ② 再注入同 id 的陈旧留档 —— "活实例与留档共存"的那一瞬。
|
||||
host.rtMu.Lock()
|
||||
if host.rtClosedTranscripts == nil { // 懒初始化：没人回收过就该是 nil
|
||||
host.rtClosedTranscripts = map[string][]agentAPI.Message{}
|
||||
}
|
||||
host.rtClosedTranscripts[id] = []agentAPI.Message{
|
||||
{Role: "user", Content: "上一通的陈旧原话"},
|
||||
}
|
||||
host.rtMu.Unlock()
|
||||
|
||||
// ③ 让活实例真的说一句（否则 history 为空，判据会假绿）。
|
||||
host.PushRTInput(id, "正在说的这句")
|
||||
waitFor(t, "活实例第一轮完成", func() bool { return p.rounds() >= 1 })
|
||||
|
||||
got, ok := host.RTTranscript(id)
|
||||
if !ok {
|
||||
t.Fatal("在跑实例取不到 transcript")
|
||||
}
|
||||
for _, m := range got {
|
||||
if m.Content == "上一通的陈旧原话" {
|
||||
t.Fatalf("★ 在跑实例的 transcript 返回了留档里的陈旧原文 —— 留档遮蔽了活实例")
|
||||
}
|
||||
}
|
||||
found := false
|
||||
for _, m := range got {
|
||||
if m.Content == "正在说的这句" {
|
||||
found = true
|
||||
}
|
||||
}
|
||||
if !found {
|
||||
t.Fatalf("在跑实例的 transcript 里没有它自己的输入：%s",
|
||||
truncateStr(FormatRTTranscript(got), 200))
|
||||
}
|
||||
}
|
||||
|
||||
// TestStopRTChildren_UnregistersInbound 钉住②：
|
||||
// 批量停止（父退出路径）必须与单实例停止**行为对称**，注销 child/rt-<id>。
|
||||
//
|
||||
// 为什么必须一致：登记表是共享的、按 name 全局唯一，残留会随
|
||||
// create/destroy 次数单调累积，而外部没有任何工具能单独注销一条 inputch
|
||||
// —— 只能重启 homed 清（驻留子那边已有同形态的历史缺陷记录）。
|
||||
//
|
||||
// 变异方式：去掉 StopRTChildren 循环里的 Unregister ⇒ 本判据红。
|
||||
func TestStopRTChildren_UnregistersInbound(t *testing.T) {
|
||||
p := &rtLongHistoryProvider{}
|
||||
host, _ := newRTHost(t, p)
|
||||
reg := host.io.ChannelRegistry()
|
||||
|
||||
const id = "bulk"
|
||||
rtSpawn(t, host, id)
|
||||
|
||||
inbound := inboundChannelName("rt/" + id)
|
||||
if _, ok := reg.Lookup(inbound); !ok {
|
||||
t.Fatalf("创建后没有登记入站通道 %q —— 判据前提不成立", inbound)
|
||||
}
|
||||
|
||||
if n := host.StopRTChildren(); n != 1 {
|
||||
t.Fatalf("StopRTChildren 返回 %d，期望 1", n)
|
||||
}
|
||||
if _, ok := reg.Lookup(inbound); ok {
|
||||
t.Fatalf("★ 批量停止后入站登记 %q 仍留在共享登记表里 —— "+
|
||||
"会随 create/destroy 次数单调累积，且外部无法单独注销", inbound)
|
||||
}
|
||||
}
|
||||
|
||||
// TestStopRTChild_SingleAlsoUnregisters 是②的对照组：
|
||||
// 单实例停止**一直**是对的，这条留着防止"修复数时把单数改坏"。
|
||||
func TestStopRTChild_SingleAlsoUnregisters(t *testing.T) {
|
||||
p := &rtLongHistoryProvider{}
|
||||
host, _ := newRTHost(t, p)
|
||||
reg := host.io.ChannelRegistry()
|
||||
|
||||
const id = "single"
|
||||
rtSpawn(t, host, id)
|
||||
inbound := inboundChannelName("rt/" + id)
|
||||
if _, ok := reg.Lookup(inbound); !ok {
|
||||
t.Fatalf("创建后没有登记入站通道 %q", inbound)
|
||||
}
|
||||
if err := host.StopRTChild(id); err != nil {
|
||||
t.Fatalf("停止失败：%v", err)
|
||||
}
|
||||
if _, ok := reg.Lookup(inbound); ok {
|
||||
t.Fatalf("单实例停止后入站登记 %q 仍残留", inbound)
|
||||
}
|
||||
}
|
||||
|
||||
// TestStopRTChildren_RtChannelsStayDangling 守住用户裁定：
|
||||
// 批量停止**不得**动插件声明的 rt 通道（那两条销毁后悬空）。
|
||||
//
|
||||
// 这条是②的护栏：注销入站登记很容易顺手写成"把 rt 通道也清掉"，
|
||||
// 而那样会让主 agent（不持有 rt 能力）通过 Owner=="" 的兜底分支
|
||||
// 静默接管一条 rt 通道 —— 正是裁定明确禁止的。
|
||||
func TestStopRTChildren_RtChannelsStayDangling(t *testing.T) {
|
||||
p := &rtLongHistoryProvider{}
|
||||
host, _ := newRTHost(t, p)
|
||||
reg := host.io.ChannelRegistry()
|
||||
|
||||
const id = "dangling"
|
||||
rtSpawn(t, host, id)
|
||||
host.StopRTChildren()
|
||||
|
||||
for _, name := range []string{"voice-in", "voice"} {
|
||||
ch, ok := reg.Lookup(name)
|
||||
if !ok {
|
||||
t.Fatalf("批量停止把插件声明的 rt 通道 %q 也删了 —— 通道应保持悬空而非消失", name)
|
||||
}
|
||||
if ch.Owner != id {
|
||||
t.Fatalf("rt 通道 %q 的归属被改成了 %q（期望仍悬空指向已销毁的 %q）",
|
||||
name, ch.Owner, id)
|
||||
}
|
||||
}
|
||||
// 悬空是可诊断的：路由查不到该 owner 时应留日志并兜底给父。
|
||||
// 这里只断言"不会凭空变成可复用通道"。
|
||||
time.Sleep(10 * time.Millisecond)
|
||||
}
|
||||
Reference in New Issue
Block a user