diff --git a/internal/agent/core/agent.go b/internal/agent/core/agent.go index 8fd6d883..47e7c76c 100644 --- a/internal/agent/core/agent.go +++ b/internal/agent/core/agent.go @@ -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 diff --git a/internal/agent/core/rtchild.go b/internal/agent/core/rtchild.go index c17f63af..e9ff1a0c 100644 --- a/internal/agent/core/rtchild.go +++ b/internal/agent/core/rtchild.go @@ -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-),本函数曾漏了这一步。 +// +// 漏的后果与驻留子那边记录的是同一个形态:登记表是共享的、按 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-。 + 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 的完整上下文 ── diff --git a/internal/agent/core/rtchild_reuse_test.go b/internal/agent/core/rtchild_reuse_test.go new file mode 100644 index 00000000..8efded3d --- /dev/null +++ b/internal/agent/core/rtchild_reuse_test.go @@ -0,0 +1,224 @@ +package core + +// rtchild 的**同名复用**与**批量停止**的回收纪律(2026-10-30 修)。 +// +// ## 这一组判据锁的是什么 +// +// rtchild 的留档表与通道登记表都以 **id 字符串**为键,而 id 由模型给。 +// 「拨完一通、再拨同名的一通」是完全正常的使用方式,但它会让上一代 +// 留下的东西被当成这一代的 —— 而且**完全不报错**,只是内容错了。 +// +// 三条缺陷、三条判据、每条都能变红(变异方式写在各自注释里): +// +// ① 留档跨代串线 —— rt_transcript 读到上一通的对话 +// ② 批量停止漏注销 —— child/rt- 登记单调累积(只能重启 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-。 +// +// 为什么必须一致:登记表是共享的、按 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) +} diff --git a/internal/plugins/rtchild_reuse_e2e_test.go b/internal/plugins/rtchild_reuse_e2e_test.go new file mode 100644 index 00000000..b7ba2ff8 --- /dev/null +++ b/internal/plugins/rtchild_reuse_e2e_test.go @@ -0,0 +1,271 @@ +//go:build linux || darwin + +package plugins + +// 端到端判据:已发布的 rtchild 修复在**真实外部插件 + 真 RPC** 下成立。 +// +// ## 为什么这一组必须单独存在(与 FullChain 的分工) +// +// `TestExternalRTChild_FullChain` 证明"全链路能跑通"——登记→划拨→注入→ +// 输出回写→回收悬空。但它**一次都不测同名复用**,也不测批量停止: +// 它用的是全新的 id、且只在末尾 Cleanup 里 StopRTChildren。 +// +// 而 2026-10-30 修的三处恰好全在"复用/清理"这条线上: +// +// ① RTTranscript 的读序(留档遮蔽在跑实例 ⇒ 两通电话串线) +// ② StopRTChildren 漏注销入站登记(child/rt- 单调累积) +// ③ 留档跨代覆写语义 +// +// 三者都是**跨通话生命周期**的缺陷 —— 单实例 e2e 天然测不到。这一组用 +// 同一个插件产物、同一个通道对,反复"拨号—挂断",把清理路径跑成主路径。 +// +// ## 前置(与 FullChain 同口径) +// +// RTCHILD_E2E_PLUGIN_BIN=/path/to/plugin.bin.linux.amd64 \ +// go test ./internal/plugins/ -run TestExternalRTChild_Reuse -v +// +// 未提供产物时 **skip** 而非 fail(CI 不强制先构建插件)。 + +import ( + "fmt" + "strings" + "testing" + "time" + + agentCore "github.com/JianFeeeee/HologramAgent/internal/agent/core" +) + +// rtE2EHost 装配一个"插件面 + 内核 Agent"的现场,供复用类判据共用。 +// +// 与 FullChain 内联装配的唯一差别是抽出来复用:三条判据各要一份干净现场, +// 内联复制会让"为什么这三条要各自 spawn 一套"变得不可读。 +func rtE2EHost(t *testing.T, bin string) (*testPluginEnv, *agentCore.Agent, string, string) { + t.Helper() + env := setupIntegration(t) + + plgDir := env.tmpDir + "/plugins" + installRTPlugin(t, plgDir, "rtvoice-e2e", bin) + if err := env.pluginReg.Load(plgDir); err != nil { + t.Fatalf("加载外部插件: %v", err) + } + if env.pluginReg.Get("rtvoice-e2e") == nil { + t.Fatal("rtvoice-e2e 未加载") + } + + host := agentCore.New(agentCore.AgentConfig{ + ID: "reuse-host", + IO: env.iom, + Provider: &e2eEchoProvider{}, + }) + return env, host, "rtvoice-e2e-in", "rtvoice-e2e" +} + +// TestExternalRTChild_ReuseDoesNotLeakTranscript 钉住①在**真实链路**下成立: +// 同名 rtchild 回收后重建,新建实例的 transcript 不得含上一通的内容; +// 同时上一通的留档(在覆写之前)必须仍然可取 —— 那是交接契约。 +// +// 变异方式:RTTranscript 改回 archive-first ⇒ 第二条断言必红(进程内判据 +// 已实测;本判据保证同样的错误在跨进程装配下也拦得住)。 +func TestExternalRTChild_ReuseDoesNotLeakTranscript(t *testing.T) { + bin := rtE2EPluginBin(t) + env, host, inCh, outCh := rtE2EHost(t, bin) + defer env.cleanup() + defer host.StopRTChildren() + + const id = "e2e-reuse" + + // ── 第一通:注入 → 跑一轮 → 回收 ── + if _, err := host.SpawnRTChild(agentCore.RTChildOptions{ID: id, Input: inCh, Output: outCh}); err != nil { + t.Fatalf("第一通创建失败: %v", err) + } + if _, err := env.stageHost.ExecuteTool("rtvoice_inject", + map[string]interface{}{"text": "第一通说的话"}); err != nil { + t.Fatalf("第一通注入失败: %v", err) + } + waitHistory(t, host, id, 2, "第一通(user+assistant)") + + firstMsgs, ok := host.RTTranscript(id) + if !ok || len(firstMsgs) == 0 { + t.Fatalf("第一通 transcript 为空(ok=%v len=%d)", ok, len(firstMsgs)) + } + firstText := agentCore.FormatRTTranscript(firstMsgs) + if n, err := host.CloseRTChild(id); err != nil || n == 0 { + t.Fatalf("第一通回收失败(n=%d err=%v)", n, err) + } + + // ★ 回收后留档必须仍可取 —— 这是「推信号 + 拉状态」的另一半。 + // 若这条红,说明有人为了"修串线"把留档删了(我本人曾写错过这一版)。 + archived, ok := host.RTTranscript(id) + if !ok || len(archived) == 0 { + t.Fatalf("★ 回收后取不到上一通原文 —— 交接契约断链(推了信号却拉不到正文)") + } + if !strings.Contains(agentCore.FormatRTTranscript(archived), "第一通说的话") { + t.Fatalf("留档里没有第一通的内容:%s", + truncateE2E(agentCore.FormatRTTranscript(archived))) + } + + // ── 第二通:同名重建,必须"忘掉"上一通 ── + if _, err := host.SpawnRTChild(agentCore.RTChildOptions{ID: id, Input: inCh, Output: outCh}); err != nil { + t.Fatalf("同名重建失败: %v", err) + } + if _, err := env.stageHost.ExecuteTool("rtvoice_inject", + map[string]interface{}{"text": "第二通说的话"}); err != nil { + t.Fatalf("第二通注入失败: %v", err) + } + waitHistory(t, host, id, 2, "第二通(user+assistant)") + + second, ok := host.RTTranscript(id) + if !ok { + t.Fatal("第二通取不到 transcript") + } + secondText := agentCore.FormatRTTranscript(second) + if strings.Contains(secondText, "第一通说的话") { + t.Fatalf("★ 同名重建后 transcript 含**第一通**的内容 —— 两通电话串线:%s", + truncateE2E(secondText)) + } + if !strings.Contains(secondText, "第二通说的话") { + t.Fatalf("第二通 transcript 里没有它自己的输入:%s", truncateE2E(secondText)) + } + _ = firstText +} + +// TestExternalRTChild_BulkStopUnregistersInbound 钉住②在真实链路下成立: +// 批量停止必须把 child/rt- 从共享登记表里摘掉。 +// +// 为什么要在跨进程层测:入站登记是**内核自己建的**,但登记表由插件 +// 装配共用。进程内判据用的是裸 IOManager;这里用的是真实插件装好的那张表 +// ——"在真实装配下也摘得干净"才算数。 +func TestExternalRTChild_BulkStopUnregistersInbound(t *testing.T) { + bin := rtE2EPluginBin(t) + env, host, inCh, outCh := rtE2EHost(t, bin) + defer env.cleanup() + + reg := env.iom.ChannelRegistry() + + const id = "e2e-bulk" + if _, err := host.SpawnRTChild(agentCore.RTChildOptions{ID: id, Input: inCh, Output: outCh}); err != nil { + t.Fatalf("创建失败: %v", err) + } + + // 入站登记的名字由内核约定:child/rt-。 + // 这里不直接调用未导出的 inboundChannelName,而是按登记表现实探测: + // 批量停止前后必须有**恰好一条** child/rt- 前缀的登记消失。 + const inboundPrefix = "child/rt/" + countInbound := func() int { + n := 0 + for _, ch := range reg.List() { + if strings.HasPrefix(ch.Name, inboundPrefix) { + n++ + } + } + return n + } + before := countInbound() + if before == 0 { + t.Fatalf("创建后没有任何 %q 前缀的入站登记 —— 判据前提不成立", inboundPrefix) + } + + if n := host.StopRTChildren(); n != 1 { + t.Fatalf("StopRTChildren 返回 %d,期望 1", n) + } + after := countInbound() + if after != before-1 { + t.Fatalf("★ 批量停止后入站登记 %d → %d(期望减少 1)—— "+ + "登记会随 create/destroy 次数单调累积,而外部无法单独注销", + before, after) + } + + // 插件声明的 rt 通道必须**悬空**而非消失(用户裁定:主 agent 无 rt 能力)。 + for _, n := range []string{inCh, outCh} { + ch, ok := reg.Lookup(n) + if !ok { + t.Fatalf("批量停止把插件声明的 rt 通道 %q 也删了 —— 应保持悬空", n) + } + if ch.Owner == "" { + t.Fatalf("rt 通道 %q 被归还为未分配(应悬空指向已销毁的 %q)", n, id) + } + } +} + +// TestExternalRTChild_RepeatedCyclesDoNotAccumulate 是"长跑"形态的护栏: +// 连续多轮建—停,登记表与 rtChildren 都不得累积。 +// +// 这条对应生产里的真实场景:模型被反复要求"打个电话问一句"。 +// 单次回收正确、但多次之后残留堆积,是典型的"每次少清一点"型缺陷 +// (②正是这种:单数路径对、复数路径漏)。 +func TestExternalRTChild_RepeatedCyclesDoNotAccumulate(t *testing.T) { + bin := rtE2EPluginBin(t) + env, host, inCh, outCh := rtE2EHost(t, bin) + defer env.cleanup() + + reg := env.iom.ChannelRegistry() + countInbound := func() int { + n := 0 + for _, ch := range reg.List() { + if strings.HasPrefix(ch.Name, "child/rt/") { + n++ + } + } + return n + } + + const cycles = 3 + base := countInbound() + for i := 0; i < cycles; i++ { + id := fmt.Sprintf("cycle-%d", i) + if _, err := host.SpawnRTChild(agentCore.RTChildOptions{ + ID: id, Input: inCh, Output: outCh, + }); err != nil { + t.Fatalf("第 %d 轮创建失败: %v", i, err) + } + if n := host.StopRTChildren(); n != 1 { + t.Fatalf("第 %d 轮 StopRTChildren 返回 %d,期望 1", i, n) + } + } + + if got := countInbound(); got != base { + t.Fatalf("★ %d 轮建—停后入站登记从 %d 涨到 %d —— 每轮都有残留", cycles, base, got) + } + if ids := host.RTChildren(); len(ids) != 0 { + t.Fatalf("多轮建—停后 rtChildren 仍有 %v", ids) + } +} + +// ── 共用小工具 ── + +// waitHistory 等到 rtchild 的 history 里出现 want 条消息(超时即失败)。 +// +// ## 为什么等 history 而不等 Rounds(本仓已有专门教训) +// +// `rtRunOnce` 先 `rc.round++`(Rounds 立刻 +1),**之后**才流式生成、 +// 最后才 `rtAppendHistory`。所以"Rounds 到了"不等于"话说完了、历史入账了"—— +// 此时 RTTranscript 可能仍是空的。 +// +// 这正是 dff745c「rt_children stop 判据改为等历史入账,不等 Rounds」的 +// 同一个坑,我第一版又踩了一遍(首轮判据报"transcript 为空(len=0)")。 +// 等 history 才是可观测的正确条件。 +func waitHistory(t *testing.T, host *agentCore.Agent, id string, want int, what string) { + t.Helper() + deadline := time.Now().Add(15 * time.Second) + var last int + for time.Now().Before(deadline) { + if msgs, ok := host.RTTranscript(id); ok { + if len(msgs) >= want { + return + } + last = len(msgs) + } + time.Sleep(20 * time.Millisecond) + } + st, ok := host.RTChildStats(id) + t.Fatalf("%s:history 只到 %d 条(期望 ≥%d,ok=%v stats=%+v)—— "+ + "输入可能没到达 rtSlot,或话说完了但历史没入账", what, last, want, ok, st) +} + +func truncateE2E(s string) string { + r := []rune(s) + if len(r) <= 240 { + return s + } + return string(r[:240]) + "…" +}