From 3acc81cec8e005194c9d09d89f40a630ec77e98b Mon Sep 17 00:00:00 2001 From: AskaEth Date: Sat, 24 Oct 2026 22:02:11 +0800 Subject: [PATCH] =?UTF-8?q?feat(gateway):=20=E5=AD=90=E4=BB=A3=E7=90=86?= =?UTF-8?q?=E7=BB=93=E6=9E=9C=E5=9B=9E=E5=A1=AB=E7=88=B6=E4=BC=9A=E8=AF=9D?= =?UTF-8?q?=EF=BC=88=E7=B3=BB=E7=BB=9F=E6=B3=A8=E8=AE=B0=EF=BC=89+=20?= =?UTF-8?q?=E8=AE=B0=E5=BF=86=E6=8A=BD=E5=8F=96=E6=8E=92=E9=99=A4=EF=BC=88?= =?UTF-8?q?M32=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 断点（M29 勘测定论）：后台批次结算后 `subagent.result` 发了、hub 也把它中继成 WS 帧，但**没有任何代码把结果回填父会话** —— `chat_messages` 无行、模型下一轮 看不到，于是「她拍了个子代理出去，但没有下文」。 回填拆成互不冒充的两条面（人格=身份红线：子代理结论绝不写成她的人格发言）： - 可观测/历史面（新文件 subagent_backfill.go）：独立订阅者（不塞进 hub）订阅 `subagent.result`，写父会话一条 `role=system` 注记（内容以 `[子代理结果]` 开头 并点名 `sub-`），随后走既有 `message_persisted` 扇出 ⇒ 刷新/重连后仍在； - 模型面：同一行经转录播种成为下一回合的**系统注记**（`model.RoleSystem`）， 不是「她说过的话」。 落库口径：角色恒为 system（不冒充 user/assistant）、PersonaID 与 TurnID 留空 （注记不属于任何角色、也不属于任何一路回合）。 记忆抽取排除（新文件 subagent_backfill_extract.go）：`WithoutSubagentNotes` 装饰 `memory.Extractor`，判据是元数据（role=system + 注记标记，与 loop 的 summaryMarker 同一手法），组合根经 `loopMemoryExtractor` 套上 ⇒ 子代理结论绝不进记忆库；LLM 提取器的异步能力原样保留，压缩摘要（同样是 system）不被误剔。 可观测补齐（M29 §7）：`SpawnTool.await` 发布成功补带 source 的 INFO，`b==nil` 与 「无结果」的静默 return 改为 WARN，发布失败 WARN 带上 count/session；回填订阅者 顺手读 `bus.Dropped()`，涨了就记一条 WARN（总线丢事件不再是黑洞）。 测试（均可变异打红）：父会话恰一条 system 行且零伪造助手发言；注记/摘要/批次正文 有界且逐代理点名；不可用载荷与落库失败各留 WARN 且不广播；Start 真订阅总线； 结果不进记忆抽取（同步/异步/端到端接缝）；组合根真装配（真 bus 发布 → 父会话多 一行）。回归：go build ./... + go vet ./... + go test -count=1 ./... 全绿。 不做：不改 `subagent_result` 帧形状、不加进 frameReachesAll、不新开路由/帧族、 不注入人设、不改 Docs（契约归一建议报塔）。 --- cmd/cyrene/wire_frontend.go | 6 + cmd/cyrene/wire_memory_extract.go | 28 ++ cmd/cyrene/wire_memory_extract_test.go | 126 +++++++ cmd/cyrene/wire_runtime.go | 2 +- cmd/cyrene/wire_subagent_backfill.go | 80 ++++ cmd/cyrene/wire_subagent_backfill_test.go | 82 +++++ internal/runtime/orchestration/spawntool.go | 22 +- .../spawntool_publish_log_test.go | 187 ++++++++++ .../services/gateway/subagent_backfill.go | 196 ++++++++++ .../gateway/subagent_backfill_extract.go | 116 ++++++ .../gateway/subagent_backfill_extract_test.go | 208 +++++++++++ .../gateway/subagent_backfill_note.go | 196 ++++++++++ .../gateway/subagent_backfill_test.go | 344 ++++++++++++++++++ 13 files changed, 1589 insertions(+), 4 deletions(-) create mode 100644 cmd/cyrene/wire_memory_extract.go create mode 100644 cmd/cyrene/wire_memory_extract_test.go create mode 100644 cmd/cyrene/wire_subagent_backfill.go create mode 100644 cmd/cyrene/wire_subagent_backfill_test.go create mode 100644 internal/runtime/orchestration/spawntool_publish_log_test.go create mode 100644 internal/services/gateway/subagent_backfill.go create mode 100644 internal/services/gateway/subagent_backfill_extract.go create mode 100644 internal/services/gateway/subagent_backfill_extract_test.go create mode 100644 internal/services/gateway/subagent_backfill_note.go create mode 100644 internal/services/gateway/subagent_backfill_test.go diff --git a/cmd/cyrene/wire_frontend.go b/cmd/cyrene/wire_frontend.go index c3538bf..ed1da9a 100644 --- a/cmd/cyrene/wire_frontend.go +++ b/cmd/cyrene/wire_frontend.go @@ -165,6 +165,12 @@ func (a *app) wireFrontend() error { // backthink.New 时网关还没组装。 a.backthink.SetAuditSink(gw) } + // 子代理结果回填（M32；design/05 §2/§5/§9）：网关 **已存在**（回填的实时出口就是 + // 它的 `message_persisted` 扇出）之后才能接线。它订阅 `subagent.result`，把后台 + // 批次的结论写成父会话的一条 **system 注记**（不冒充 user/assistant、不带 turn_id、 + // 绝不进记忆抽取），因此「她拍了个子代理出去，然后没有下文」这件事在历史与下一轮 + // 上下文里都有落点。装配细节与降级口径见 wire_subagent_backfill.go。 + wireSubagentBackfill(a, gw) return nil } diff --git a/cmd/cyrene/wire_memory_extract.go b/cmd/cyrene/wire_memory_extract.go new file mode 100644 index 0000000..8d7fc29 --- /dev/null +++ b/cmd/cyrene/wire_memory_extract.go @@ -0,0 +1,28 @@ +package main + +// 本文件只有一件事：给循环的记忆提取器套上「剔除子代理结果注记」这一层，并把这条 +// 口径写在一个可测的入口上（M32）。 +// +// 为什么要单独一个函数，而不是在 `wire_runtime.go` 的 Options 字面量里就地调一次： +// +// - 它是**红线级**口径（子代理结论绝不进记忆库），值得有一个名字与一个测试， +// 而不是夹在一长串 Options 里的一行调用 —— 那一行被删掉时没有任何东西会红； +// - 组合根里所有「必须套上」的装饰都走这种小函数（同 `a.buildMemoryExtractor()` +// 之于提取器本身），语义在函数名下，接线在唯一一处。 +// +// 判据与实现见 `internal/services/gateway/subagent_backfill_extract.go`：元数据 +// （role=system + 注记标记），不是提示词。 + +import ( + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/memory" + "git.yeij.top/AskaEth/Cyrene_v2/internal/services/gateway" +) + +// loopMemoryExtractor 返回循环实际使用的提取器（`wire_runtime.go` 的 +// `loop.Options.Extractor` 就是它）：内层是组合根按配置构建的那个（LLM 或确定性）， +// 外层保证子代理结果注记**永远不进**抽取输入。 +// +// base 为 nil 时返回 nil（未配置提取器的部署逐字不变：循环那边同样是 nil 即跳过）。 +func loopMemoryExtractor(base memory.Extractor) memory.Extractor { + return gateway.WithoutSubagentNotes(base) +} diff --git a/cmd/cyrene/wire_memory_extract_test.go b/cmd/cyrene/wire_memory_extract_test.go new file mode 100644 index 0000000..2dfa9d3 --- /dev/null +++ b/cmd/cyrene/wire_memory_extract_test.go @@ -0,0 +1,126 @@ +package main + +import ( + "context" + "strings" + "sync" + "testing" + + "git.yeij.top/AskaEth/Cyrene_v2/internal/bus" + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/history" + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/memory" + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/model" + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/orchestration" + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/transcript" + "git.yeij.top/AskaEth/Cyrene_v2/internal/services/gateway" +) + +// 本文件锁住组合根那一条红线：循环用的提取器**必须**套上「剔除子代理注记」这一层 +// （loopMemoryExtractor）。删掉那层装饰，子代理的结论就会被当成对话写进记忆库 —— +// 那条污染线刚查过（84% 记忆来自内部会话），不能再添一条。 +// +// 断言的是**组合根这个入口**的语义，因此即使 gateway 那边的装饰器自己写对，把这里 +// 换成 `base` 直通也会红。 + +// noteCaptureStore 是回填落库面的替身：只留下注记正文。 +type noteCaptureStore struct { + mu sync.Mutex + rows []history.Message +} + +func (s *noteCaptureStore) AppendFull(_ context.Context, msg history.Message) (history.Message, error) { + s.mu.Lock() + defer s.mu.Unlock() + if msg.ID == "" { + msg.ID = "msg-note" + } + s.rows = append(s.rows, msg) + return msg, nil +} + +// seenCaptureExtractor 记录内层提取器实际收到的对话。 +type seenCaptureExtractor struct { + mu sync.Mutex + seen []model.Message +} + +func (e *seenCaptureExtractor) Extract(_ context.Context, conversation []model.Message) ([]memory.MemoryItem, error) { + e.mu.Lock() + defer e.mu.Unlock() + e.seen = append([]model.Message(nil), conversation...) + return nil, nil +} + +// TestLoopMemoryExtractorStripsBackfilledNotes 走真链路取一条注记（回填 → 落库 → +// 转录播种），再把它连同一条主人输入交给组合根返回的提取器：内层必须只看得到主人 +// 那一句。 +func TestLoopMemoryExtractorStripsBackfilledNotes(t *testing.T) { + b := bus.New() + t.Cleanup(func() { _ = b.Close() }) + store := ¬eCaptureStore{} + bf, err := gateway.NewSubagentBackfill(gateway.SubagentBackfillDeps{Bus: b, History: store}) + if err != nil { + t.Fatalf("NewSubagentBackfill: %v", err) + } + if err := bf.Handle(context.Background(), bus.Event{ + Subject: orchestration.SubagentResultSubject, + Payload: &orchestration.SubagentResultEvent{ + SessionID: history.MainSessionID, + SpawnResult: &orchestration.SpawnResult{ + Status: "done", + Summary: "子代理汇总: 共 1 个, 1 成功", + Agents: []orchestration.SpawnAgentResult{{ + ID: "sub-a1", Goal: "调研 X", Status: "done", Summary: "结论：X 可行", + }}, + }, + }, + }); err != nil { + t.Fatalf("Handle: %v", err) + } + if len(store.rows) != 1 { + t.Fatalf("backfill wrote %d rows, want one system note", len(store.rows)) + } + note := transcript.Message(store.rows[0]) + if note.Role != model.RoleSystem { + t.Fatalf("seeded note role = %q, want system", note.Role) + } + + inner := &seenCaptureExtractor{} + extractor := loopMemoryExtractor(inner) + if extractor == nil { + t.Fatal("loopMemoryExtractor(non-nil) = nil") + } + if _, err := extractor.Extract(context.Background(), []model.Message{ + note, + model.TextMessage(model.RoleUser, "接着说"), + }); err != nil { + t.Fatalf("Extract: %v", err) + } + inner.mu.Lock() + seen := inner.seen + inner.mu.Unlock() + if len(seen) != 1 { + t.Fatalf("inner extractor saw %d messages, want only the user turn: the composition root dropped the note filter", len(seen)) + } + if !strings.Contains(messageTextOf(seen[0]), "接着说") { + t.Fatalf("inner extractor lost the real dialogue: %+v", seen) + } +} + +// TestLoopMemoryExtractorKeepsNil 锁住未配置提取器的部署逐字不变。 +func TestLoopMemoryExtractorKeepsNil(t *testing.T) { + if got := loopMemoryExtractor(nil); got != nil { + t.Fatalf("loopMemoryExtractor(nil) = %v, want nil", got) + } +} + +// messageTextOf 返回一条模型消息的正文（测试窄视图）。 +func messageTextOf(m model.Message) string { + var b strings.Builder + for _, c := range m.Content { + if c.Type == model.ContentTypeText { + b.WriteString(c.Text) + } + } + return b.String() +} diff --git a/cmd/cyrene/wire_runtime.go b/cmd/cyrene/wire_runtime.go index 9ecdc9f..fb4c9ed 100644 --- a/cmd/cyrene/wire_runtime.go +++ b/cmd/cyrene/wire_runtime.go @@ -324,7 +324,7 @@ func (a *app) wireRuntime() error { OnTurnEnd: a.onTurnEnd, Memory: store, Retriever: a.retriever, - Extractor: a.buildMemoryExtractor(), + Extractor: loopMemoryExtractor(a.buildMemoryExtractor()), Checkpoint: checkpoint, Tracer: tracer, Flow: flowPub, diff --git a/cmd/cyrene/wire_subagent_backfill.go b/cmd/cyrene/wire_subagent_backfill.go new file mode 100644 index 0000000..d345948 --- /dev/null +++ b/cmd/cyrene/wire_subagent_backfill.go @@ -0,0 +1,80 @@ +package main + +// 本文件把**子代理结果回填**接进组合根（M32；design/05 §2/§5/§9）。 +// +// 为什么必须在组合根装配，而不是在网关 `New` 里顺手订阅： +// +// - 回填要写的是**父会话的聊天历史**，而网关 New 拿到的 History 与组合根 +// `as.history` 是同一个 `*history.Store`，但它此刻还没被 hub 的广播出口包上 —— +// 落库之后必须用**同一个** `message_persisted` 扇出（`Server.BroadcastPersisted`） +// 才能让在线端与刷新/重连的端看到同一行； +// - 与 hub 的分工：hub 只把主题归一成 `subagent_result` 帧（投递），回填订阅者只把 +// 结果写成父会话的一行系统注记（落库）。两者订阅**同一个主题**、互不相识， +// 因此以后改帧不会牵连落库，改落库也不会牵连帧。 +// +// 订阅用的 ctx 是 `context.Background()`：组合根这一步没有任何长生命周期 ctx 可用 +// （`newApp` 的 ctx 只用于构造期的库操作），而进程关停时 `app.close` 会 +// `a.bus.Close()` —— 它取消所有订阅的处理器 ctx，因此这里不需要第二个停止开关。 +// 这与 hub（`Server.Start` 里自建 ctx 并在 `Close` 取消）是同一取舍。 + +import ( + "context" + + "git.yeij.top/AskaEth/Cyrene_v2/internal/logger" + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/history" + "git.yeij.top/AskaEth/Cyrene_v2/internal/services/gateway" +) + +// backfillWireLogSource 是这条接线自己的日志来源：订阅成功/失败要能按来源定位， +// 而不是混在 `main` 或 `gateway` 的裸日志里（AGENTS §10 的日志来源铁律）。 +const backfillWireLogSource = "cmd/cyrene/wire_subagent_backfill" + +// wireSubagentBackfill 把「后台子代理结果 → 父会话系统注记」的订阅者接到组合根的 +// 进程内总线上（与调度器、网关 hub **同一个** bus 实例）。 +// +// 三条降级口径（都**必须可见**，绝不静默）： +// +// - 没有 bus 或没有历史存储：这条回填此刻做不了（未接持久化的构建）； +// - 订阅失败：总线已关闭或主题非法 —— 那意味着后台结果永远不会回填； +// - 广播出口缺失（网关为 nil）：仍然落库（历史是唯一事实来源），只是不推帧。 +func wireSubagentBackfill(a *app, gw *gateway.Server) { + log := logger.WithSource(a.log, backfillWireLogSource) + if a.bus == nil || a.history == nil { + log.Warn("subagent backfill not wired: needs both a bus and a history store", + "bus", a.bus != nil, "history", a.history != nil) + return + } + backfill, err := gateway.NewSubagentBackfill(gateway.SubagentBackfillDeps{ + Bus: a.bus, + History: a.history, + Broadcast: backfillSink(gw), + Log: a.log, + }) + if err != nil { + log.Warn("subagent backfill not wired", "err", err) + return + } + if err := backfill.Start(context.Background()); err != nil { + log.Warn("subagent backfill subscribe failed; background results will not reach the parent session", + "err", err) + return + } + log.Info("subagent backfill wired", + "subject", "subagent.result", "session_role", history.RoleSystem) +} + +// backfillSink 返回回填的下行出口。网关为 nil 时返回**真 nil 接口**（带类型的 nil +// 指针会让订阅者里的 `!= nil` 判断失效，落库之后去调一个空接收者）。 +func backfillSink(gw *gateway.Server) gateway.SubagentNoteSink { + if gw == nil { + return nil + } + return gw +} + +// 编译期锁住这条链路的两端：网关仍是回填的下行出口，历史存储仍是它的落库面。 +// 任何一边改签名，这里先红，而不是等到运行期回填静默缺席。 +var ( + _ gateway.SubagentNoteSink = (*gateway.Server)(nil) + _ gateway.SubagentNoteStore = (*history.Store)(nil) +) diff --git a/cmd/cyrene/wire_subagent_backfill_test.go b/cmd/cyrene/wire_subagent_backfill_test.go new file mode 100644 index 0000000..947fe15 --- /dev/null +++ b/cmd/cyrene/wire_subagent_backfill_test.go @@ -0,0 +1,82 @@ +package main + +import ( + "context" + "testing" + "time" + + "git.yeij.top/AskaEth/Cyrene_v2/internal/bus" + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/history" + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/orchestration" +) + +// 本文件锁住 M32 的接线本身：**组合根真的把回填订阅者挂上了那条总线**。 +// +// 为什么单独测这一步：这次缺陷的形状就是「所有零件都在、没有一根线」。回填订阅者 +// 写对了、总线也发得对，只要 `wireFrontend` 里少掉那一行调用，主人在界面上看到的 +// 仍然是「没有下文」—— 而且没有任何测试会红。因此这里断言的是**装配后的行为**： +// 组合根接线 → 真 bus 上发布编排层载荷 → 父会话真的多出一条系统注记行。 + +// waitForBackfillRow 轮询历史，直到父会话出现一条消息或到期。 +func waitForBackfillRow(t *testing.T, st *history.Store, sessionID string) []history.Message { + t.Helper() + deadline := time.Now().Add(3 * time.Second) + for time.Now().Before(deadline) { + rows, err := st.Recent(context.Background(), sessionID, 10) + if err != nil { + t.Fatalf("read recent: %v", err) + } + if len(rows) > 0 { + return rows + } + time.Sleep(5 * time.Millisecond) + } + t.Fatal("the parent session never received the backfilled note: the composition root did not wire the subscriber") + return nil +} + +// TestWireSubagentBackfillPersistsThroughBus 是 ④ 的正面证据：装配 → 发布 → 落行。 +func TestWireSubagentBackfillPersistsThroughBus(t *testing.T) { + b := bus.New() + t.Cleanup(func() { _ = b.Close() }) + a := &app{bus: b, history: newTestHistory(t), log: testLog()} + + wireSubagentBackfill(a, nil) + + payload := &orchestration.SubagentResultEvent{ + SessionID: history.MainSessionID, + SpawnResult: &orchestration.SpawnResult{ + Status: "done", + Summary: "子代理汇总: 共 1 个, 1 成功", + Agents: []orchestration.SpawnAgentResult{{ + ID: "sub-a1", Goal: "调研 X", Role: "cyrene", Status: "done", Summary: "结论：X 可行", + }}, + }, + } + if err := b.Publish(context.Background(), orchestration.SubagentResultSubject, payload); err != nil { + t.Fatalf("publish: %v", err) + } + + rows := waitForBackfillRow(t, a.history, history.MainSessionID) + if len(rows) != 1 { + t.Fatalf("parent session has %d messages, want exactly the one system note: %+v", len(rows), rows) + } + row := rows[0] + if row.Role != history.RoleSystem { + t.Fatalf("backfilled row role = %q, want %q (a sub-agent conclusion is not her own words)", row.Role, history.RoleSystem) + } + if row.TurnID != "" || row.PersonaID != "" { + t.Fatalf("backfilled row = %+v, want no turn_id and no persona_id", row) + } +} + +// TestWireSubagentBackfillWithoutHistoryWarns 锁住降级口径：没有历史存储时不装配、 +// 不 panic，而是留一条可见的告警（绝不假装回填成立）。 +func TestWireSubagentBackfillWithoutHistoryWarns(t *testing.T) { + a := &app{bus: bus.New(), log: testLog()} + wireSubagentBackfill(a, nil) // 没有 history：只允许告警 + if err := a.bus.Publish(context.Background(), orchestration.SubagentResultSubject, + &orchestration.SubagentResultEvent{SessionID: history.MainSessionID}); err != nil { + t.Fatalf("publish on an unwired root: %v", err) + } +} diff --git a/internal/runtime/orchestration/spawntool.go b/internal/runtime/orchestration/spawntool.go index d46ad17..ddc19bd 100644 --- a/internal/runtime/orchestration/spawntool.go +++ b/internal/runtime/orchestration/spawntool.go @@ -205,25 +205,41 @@ func (t *SpawnTool) foreground(ids []string, specs []AgentSpec) any { } // await 阻塞到批次结算，并把合并结果作为 design/05 §2「结束后写执行总结 → 返回主会话」 -// 的交接广播出去。它运行在脱离的 goroutine 上；总线为 nil 或 id 未知时安静结束。 +// 的交接广播出去。它运行在脱离的 goroutine 上。 +// +// 🔴 **每一条出口都要留下痕迹**（M29 勘测 §7 的可观测性缺口）：修复前「发布成功零 +// 日志、无总线静默 return」，于是「到底发没发」只能靠负证据推断（这次缺陷最费时的 +// 地方）。现在成功发布记一条带来源的 INFO，缺总线与发布失败各记一条 WARN —— +// 「压根没发 / 发了但失败 / 发了」三态都能从日志正面判定。 +// // parentSession 是该批次所属的父会话，随载荷一起广播，使中继能把回填结果 -// 归属到父会话的实时流，而不是发一张无归属的帧。 +// 归属到父会话的实时流，而不是发一张无归属的帧；gateway 的回填订阅者也按它写 +// 父会话的系统注记（M32）。 func (t *SpawnTool) await(ids []string, parentSession string) { res, err := t.sched.Wait(context.Background(), ids) if err != nil { t.log.Warn("sub-agent batch wait failed", "count", len(ids), "error", err) if res == nil { + t.log.Warn("sub-agent results not published: batch produced no result", + "count", len(ids), "session", parentSession) return } } b := t.sched.Bus() if b == nil { + t.log.Warn("sub-agent results not published: no bus", + "count", len(ids), "session", parentSession) return } payload := &SubagentResultEvent{SessionID: parentSession, SpawnResult: doneSpawnResult(res)} if perr := b.Publish(context.Background(), SubagentResultSubject, payload); perr != nil { - t.log.Warn("publish sub-agent results", "error", perr) + t.log.Warn("publish sub-agent results", "error", perr, + "count", len(ids), "session", parentSession) + return } + t.log.Info("sub-agent results published", + "subject", SubagentResultSubject, "session", parentSession, + "count", len(ids), "status", payload.Status, "agents", len(payload.Agents)) } // SpawnResult 是派发工具返回给派发模型的载荷：批次在前台窗口内结算时是它的合并结果， diff --git a/internal/runtime/orchestration/spawntool_publish_log_test.go b/internal/runtime/orchestration/spawntool_publish_log_test.go new file mode 100644 index 0000000..52d3b8f --- /dev/null +++ b/internal/runtime/orchestration/spawntool_publish_log_test.go @@ -0,0 +1,187 @@ +package orchestration + +import ( + "context" + "errors" + "log/slog" + "sync" + "testing" + "time" + + "git.yeij.top/AskaEth/Cyrene_v2/internal/bus" +) + +// 本文件锁住 M29 勘测 §7 的可观测性缺口（M32 顺手补掉的那一半）：`await` 的 +// **每一条出口都必须留痕**。修复前「发布成功零日志、无总线静默 return」，于是 +// 「后台结果到底发没发」只能靠负证据推断 —— 那正是这次缺陷最费时间的地方。 +// +// 三条被测出口，与日志一一对应： +// - 发布成功 ⇒ INFO `sub-agent results published`（带 source、父会话、条数、状态）； +// - 没有总线 ⇒ WARN `... no bus`（以前是静默 return）； +// - 发布失败 ⇒ WARN `publish sub-agent results`（带 err）。 + +// failingPublishBus 是一个 Publish 必定失败的 Bus：用于证明「发布失败不静默」。 +type failingPublishBus struct { + err error +} + +func (b failingPublishBus) Publish(context.Context, string, any) error { return b.err } + +func (b failingPublishBus) Subscribe(context.Context, string, bus.Handler) (func(), error) { + return func() {}, nil +} + +func (b failingPublishBus) Close() error { return nil } + +// awaitLogSink 收集结构化日志记录（spawntool 的 logger 已带 source）。 +type awaitLogSink struct { + mu sync.Mutex + recs []slog.Record +} + +// awaitLogCapture 是一个会把 WithAttrs 累积进记录里的处理器：source 等标签是经 +// `logger.With` 挂上的，只读 Record.Attrs 会看不到它们。 +type awaitLogCapture struct { + sink *awaitLogSink + attrs []slog.Attr +} + +func newAwaitLogCapture() *awaitLogCapture { return &awaitLogCapture{sink: &awaitLogSink{}} } + +func (c *awaitLogCapture) Enabled(context.Context, slog.Level) bool { return true } + +func (c *awaitLogCapture) Handle(_ context.Context, rec slog.Record) error { + rec.AddAttrs(c.attrs...) + c.sink.mu.Lock() + defer c.sink.mu.Unlock() + c.sink.recs = append(c.sink.recs, rec.Clone()) + return nil +} + +func (c *awaitLogCapture) WithAttrs(as []slog.Attr) slog.Handler { + merged := make([]slog.Attr, 0, len(c.attrs)+len(as)) + merged = append(merged, c.attrs...) + merged = append(merged, as...) + return &awaitLogCapture{sink: c.sink, attrs: merged} +} + +func (c *awaitLogCapture) WithGroup(string) slog.Handler { return c } + +// find 返回第一条级别匹配且 msg 等值的记录。 +func (c *awaitLogCapture) find(level slog.Level, want string) (slog.Record, bool) { + c.sink.mu.Lock() + defer c.sink.mu.Unlock() + for _, rec := range c.sink.recs { + if rec.Level == level && rec.Message == want { + return rec, true + } + } + return slog.Record{}, false +} + +// attr 读取一条记录上的字符串属性（含经 WithAttrs 挂上的标签）。 +func (c *awaitLogCapture) attr(rec slog.Record, key string) (string, bool) { + var out string + var ok bool + rec.Attrs(func(a slog.Attr) bool { + if a.Key != key { + return true + } + out, ok = a.Value.String(), true + return false + }) + return out, ok +} + +// settledTool 构建一个派发并结算一个子代理的工具，返回工具、该子代理 id 与日志捕获。 +func settledTool(t *testing.T, opts Options) (*SpawnTool, string, *awaitLogCapture) { + t.Helper() + s := New(opts) + s.runner = &mockRunner{} + sub, err := s.Spawn(context.Background(), AgentSpec{Goal: "查一下 X", ParentSession: "web"}) + if err != nil { + t.Fatalf("Spawn: %v", err) + } + capture := newAwaitLogCapture() + return NewSpawnTool(s, WithSpawnLogger(slog.New(capture)), WithSpawnWait(time.Second)), sub.ID, capture +} + +// TestAwaitLogsSuccessfulPublish 锁住「发成功要能从日志正面看出来」。 +func TestAwaitLogsSuccessfulPublish(t *testing.T) { + t.Parallel() + b := bus.New() + t.Cleanup(func() { _ = b.Close() }) + sp, id, capture := settledTool(t, Options{MaxParallel: 1, Bus: b}) + + published := make(chan *SubagentResultEvent, 1) + unsub, err := b.Subscribe(context.Background(), SubagentResultSubject, + func(_ context.Context, e bus.Event) error { + if ev, ok := e.Payload.(*SubagentResultEvent); ok { + select { + case published <- ev: + default: + } + } + return nil + }) + if err != nil { + t.Fatalf("subscribe: %v", err) + } + defer unsub() + + sp.await([]string{id}, "web") + + select { + case ev := <-published: + if ev.SessionID != "web" || ev.Status != "done" { + t.Fatalf("payload = %+v, want the settled batch for session web", ev) + } + case <-time.After(waitTimeout): + t.Fatal("the settled batch was never published") + } + rec, ok := capture.find(slog.LevelInfo, "sub-agent results published") + if !ok { + t.Fatal("a successful publish left no INFO: 'did it publish?' must be answerable from the log") + } + if src, _ := capture.attr(rec, "source"); src != "orchestration/spawn" { + t.Fatalf("log source = %q, want %q (AGENTS §10: no bare logs)", src, "orchestration/spawn") + } + if session, _ := capture.attr(rec, "session"); session != "web" { + t.Fatalf("log session = %q, want the parent session", session) + } + if status, _ := capture.attr(rec, "status"); status != "done" { + t.Fatalf("log status = %q, want done", status) + } +} + +// TestAwaitWarnsWithoutBus 锁住「没有总线」不再是静默 return。 +func TestAwaitWarnsWithoutBus(t *testing.T) { + t.Parallel() + sp, id, capture := settledTool(t, Options{MaxParallel: 1}) + if sp.sched.Bus() != nil { + t.Skip("this test needs a scheduler without a bus") + } + sp.await([]string{id}, "web") + rec, ok := capture.find(slog.LevelWarn, "sub-agent results not published: no bus") + if !ok { + t.Fatal("a scheduler with no bus dropped the results silently") + } + if session, _ := capture.attr(rec, "session"); session != "web" { + t.Fatalf("log session = %q, want the parent session (traceability)", session) + } +} + +// TestAwaitWarnsOnPublishFailure 锁住「发布失败必须留下 err」。 +func TestAwaitWarnsOnPublishFailure(t *testing.T) { + t.Parallel() + sp, id, capture := settledTool(t, Options{MaxParallel: 1, Bus: failingPublishBus{err: errors.New("bus down")}}) + sp.await([]string{id}, "web") + + rec, ok := capture.find(slog.LevelWarn, "publish sub-agent results") + if !ok { + t.Fatal("a failed publish left no WARN: the result would silently never arrive") + } + if msg, _ := capture.attr(rec, "error"); msg == "" { + t.Fatal("the publish-failure WARN carries no error detail") + } +} diff --git a/internal/services/gateway/subagent_backfill.go b/internal/services/gateway/subagent_backfill.go new file mode 100644 index 0000000..04a38a4 --- /dev/null +++ b/internal/services/gateway/subagent_backfill.go @@ -0,0 +1,196 @@ +package gateway + +import ( + "context" + "fmt" + "log/slog" + + "git.yeij.top/AskaEth/Cyrene_v2/internal/bus" + "git.yeij.top/AskaEth/Cyrene_v2/internal/logger" + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/history" + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/orchestration" +) + +// 本文件实现 design/05 §2/§5/§9 要求的**结果回填主会话**（M32）。修复前的断点不是 +// 传输（帧发了、hub 订阅了、前端 handler 也注册了），而是「**回填这一环根本没有 +// 实现**」：后台批次结算后的载荷只被中继成一个 WS 帧，落到面板的旁路日志面 —— +// `chat_messages` 无行、模型下一轮也看不到，于是「她拍了个子代理出去，但没有下文」。 +// +// 回填被拆成**互不冒充**的两条面（项目红线：人格 = 身份，子代理的结论绝不能读成 +// 她自己说的话）： +// +// - **可观测 / 历史面**：往父会话写一条 `role = "system"` 的注记行（**绝不**是 +// user / assistant），内容以 `subagentNoteMarker` 开头并点名 `sub-`，随后按 +// 既有 `message_persisted` 帧扇出 ⇒ 刷新/重连后仍在（design/05 §9「提示用户」）； +// - **模型面**：同一行被下一回合的转录播种（`transcript.Read`）读进上下文，因此 +// 模型在**下一次**组装时看到的是**系统注记**（`model.RoleSystem`），不是一句 +// 「她说过的话」，也不是她自己的语气。 +// +// 落库口径（防「看着有、库里没有」）： +// +// - **角色恒为 system**：`chat_messages.role` 的合法取值是 user/assistant/system +// （0003 迁移的列注释），子代理结论既不是主人说的、也不是她说的； +// - **PersonaID 留空**：这条注记的作者**不是任何角色**，写上角色 id 等于让客户端 +// 用那个角色的头像渲染「她说过」； +// - **TurnID 留空**：后台批次可能在派发它的那个回合结束**之后**才结算，它不属于 +// 任何一路回合（与主动推送的同一取舍），空值诚实地表示这一点； +// - **绝不被记忆抽取**：抽取输入是回合的 `msgs`（含播种的转录行），因此这条注记 +// 会出现在抽取器面前。排除由 `WithoutSubagentNotes`（subagent_backfill_extract.go） +// 在**组合根**套在提取器上完成 —— 判据是元数据（role=system + 注记标记），不是 +// 提示词；否则子代理结论会被当成对话吞进记忆库（那条污染线刚查过）。 +// +// 刻意**不做**的事：不改 `subagent_result` 帧形状、不把它加进 `frameReachesAll`、 +// 不新开路由/帧族、不往注记里注入任何人设。 +const backfillLogScope = "subagent-backfill" + +// SubagentNoteStore 是回填所需的最小历史写面。*history.Store 满足它；测试替身 +// 只需实现这一个方法。 +type SubagentNoteStore interface { + AppendFull(ctx context.Context, msg history.Message) (history.Message, error) +} + +// SubagentNoteSink 是回填的下行出口（`message_persisted` / `channel_persisted` +// 的既有扇出）。*Server 满足它；为 nil 时只落库不推帧（落库仍是唯一事实来源）。 +type SubagentNoteSink interface { + BroadcastPersisted(msg history.Message) +} + +// SubagentBackfillDeps 是回填订阅者的装配输入（组合根按实际依赖填充）。 +type SubagentBackfillDeps struct { + // Bus 是进程内事件总线（与调度器、网关 hub **同一个实例**）。 + Bus bus.Bus + // History 是父会话注记的落库面。为 nil 时装配即失败（没有落库面就没有回填）。 + History SubagentNoteStore + // Broadcast 是该注记的实时扇出出口。为 nil 时只落库。 + Broadcast SubagentNoteSink + // Log 是日志基底；来源由 NewSubagentBackfill 统一盖成 + // `source=gateway/subagent-backfill`。 + Log *slog.Logger +} + +// SubagentBackfill 是 `orchestration.SubagentResultSubject` 的**独立订阅者**：它把 +// 后台批次的结果写成父会话的一条系统注记并扇出。 +// +// 它刻意**不是** hub 的一部分（hub 只负责投递，写入会话属会话编排职责，混进去会让 +// 「投递」与「落库」耦合），也与 hub 的中继互不相识：两者订阅同一个主题，各干各的 +// 一半（hub 出帧、本订阅者落行）。 +// +// 并发：总线对每条订阅串行派发（bus 包文档），因此 `lastDropped` 不需要加锁。 +type SubagentBackfill struct { + bus bus.Bus + history SubagentNoteStore + broadcast SubagentNoteSink + log *slog.Logger + // dropProbe 是总线的可选背压计数器（*bus.busImpl 的 Dropped()）。总线在缓冲区 + // 满时**丢弃事件且不告诉任何人**，而丢掉的恰恰就是回填事件 —— 因此每次成功 + // 处理一条事件时顺手读一次计数，涨了就记一条 WARN（负证据变成正面可见）。 + dropProbe dropProbe + lastDropped uint64 +} + +// dropProbe 是总线背压计数器的可选窄视图。 +type dropProbe interface { + Dropped() uint64 +} + +// NewSubagentBackfill 构建回填订阅者。History 为 nil 时返回错误：一个只会丢结果的 +// 订阅者比没有订阅者更糟（它会让「已接线」看起来成立）。 +func NewSubagentBackfill(deps SubagentBackfillDeps) (*SubagentBackfill, error) { + if deps.History == nil { + return nil, fmt.Errorf("gateway: subagent backfill requires a history store") + } + if deps.Bus == nil { + return nil, fmt.Errorf("gateway: subagent backfill requires a bus") + } + probe, _ := deps.Bus.(dropProbe) + return &SubagentBackfill{ + bus: deps.Bus, + history: deps.History, + broadcast: deps.Broadcast, + log: logger.Scoped(sourcedLog(deps.Log), source, backfillLogScope), + dropProbe: probe, + }, nil +} + +// Start 把订阅者挂上总线。ctx 取消即自动退订（进程关停时 `bus.Close` 是另一道）。 +// +// 订阅失败**必须**可见：返回包装后的错误，由组合根记 WARN —— 回填静默缺席正是 +// 这次缺陷的形状。 +func (b *SubagentBackfill) Start(ctx context.Context) error { + if b == nil || b.bus == nil { + return fmt.Errorf("gateway: subagent backfill is not wired") + } + if ctx == nil { + ctx = context.Background() + } + if _, err := b.bus.Subscribe(ctx, orchestration.SubagentResultSubject, b.Handle); err != nil { + return fmt.Errorf("gateway: subscribe %s: %w", orchestration.SubagentResultSubject, err) + } + b.log.Info("subagent backfill subscribed", + "subject", orchestration.SubagentResultSubject, "role", history.RoleSystem) + return nil +} + +// Handle 是总线处理器：把一条后台批次结果写成父会话的系统注记并扇出。 +// +// 它绝不返回错误（总线会忽略返回值，只把失败留在没人看的地方）：每一种失败都在 +// 原地记带来源的日志，且**绝不**为了「有话可说」而编造一条注记。 +func (b *SubagentBackfill) Handle(ctx context.Context, e bus.Event) error { + if b == nil || e.Subject != orchestration.SubagentResultSubject { + return nil + } + ev, ok := e.Payload.(*orchestration.SubagentResultEvent) + if !ok || ev == nil { + b.log.Warn("subagent backfill: unexpected payload type; result not persisted", + "subject", e.Subject, "type", fmt.Sprintf("%T", e.Payload)) + return nil + } + // 内嵌的 *SpawnResult 为 nil 时，任何字段访问都会空指针 —— 而总线会吞掉 panic， + // 那等于把这条结果静默丢掉。因此显式拒绝并留痕。 + if ev.SpawnResult == nil { + b.log.Warn("subagent backfill: result payload carries no spawn result; not persisted", + "session", ev.SessionID) + return nil + } + if ev.SessionID == "" { + b.log.Warn("subagent backfill: result has no parent session; not persisted", + "status", ev.Status, "agents", len(ev.Agents)) + return nil + } + note := subagentNote(ev) + full, err := b.history.AppendFull(ctx, history.Message{ + SessionID: ev.SessionID, + Role: history.RoleSystem, + Content: note, + Mode: modeChat, + }) + if err != nil { + b.log.Error("subagent backfill: persist result note", + "err", err, "session", ev.SessionID, "agents", len(ev.Agents)) + return nil + } + if b.broadcast != nil { + b.broadcast.BroadcastPersisted(full) + } + b.log.Info("subagent result backfilled", + "session", ev.SessionID, "message", full.ID, "status", ev.Status, + "agents", len(ev.Agents), "runes", len([]rune(note))) + b.reportDrops() + return nil +} + +// reportDrops 把「总线自上次以来丢过多少条事件」记成一条 WARN。它在成功处理一条 +// 回填事件之后触发，因此这条负面事实至少会在下一次投递时浮出，而不是永远消失。 +func (b *SubagentBackfill) reportDrops() { + if b.dropProbe == nil { + return + } + now := b.dropProbe.Dropped() + if now == b.lastDropped { + return + } + b.log.Warn("bus dropped events (subagent results may be missing)", + "dropped_total", now, "dropped_since_last", now-b.lastDropped, + "subject", orchestration.SubagentResultSubject) + b.lastDropped = now +} diff --git a/internal/services/gateway/subagent_backfill_extract.go b/internal/services/gateway/subagent_backfill_extract.go new file mode 100644 index 0000000..24fa3a6 --- /dev/null +++ b/internal/services/gateway/subagent_backfill_extract.go @@ -0,0 +1,116 @@ +package gateway + +import ( + "context" + "strings" + + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/memory" + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/model" +) + +// 本文件是「子代理结果注记**绝不进记忆抽取**」这道防线（M32）。它是 +// `memory.Extractor` 的装饰器，由**组合根**套在循环的提取器上 +// （`cmd/cyrene/wire_runtime.go` 的 `Extractor:` 一行）。 +// +// 为什么需要它（不是理论问题，是「看着有、库里没有」的反面）： +// +// - 回填注记是一条**已落库的系统消息**，因此它会被下一回合的转录播种 +// （`transcript.Read`）读进该回合的 `msgs`，而循环在回合落定时把这整份 `msgs` +// 交给提取器（`loop.extractMemoryAsync`）。不挡，子代理的结论就会被当成对话， +// 由 LLM 提取器判断「值得记住」后写进记忆库 —— 那正是刚查过的那条污染线 +// （84% 记忆来自内部会话），绝不能再添一条。 +// - **判据必须是元数据**（AGENTS §9「身份/判据靠元数据，不靠提示词」）：这里按 +// `role == system` **且** 正文以 `subagentNoteMarker` 开头来识别，两者都是 +// **结构性事实**（该标记由本包的注记写入方唯一产出，与 loop 的 `summaryMarker` +// 同一手法），绝不靠「提示词里让模型别提取」。 +// +// 它只影响**喂给提取器的输入**：落库行、`message_persisted` 帧、下一回合模型看到的 +// 系统注记、历史接口全部逐字不变 —— 被挡掉的只有「这条注记变成一条记忆」这一件事。 + +// memoryAsyncExtractor 是 memory.LLMExtractor 的异步能力（与 loop 包内部的 +// asyncExtractor 同形）。装饰器必须在**装饰后仍然异步**，否则一次慢提取会从后台 +// goroutine 掉回回合落定的同步路径上，直接拖慢回复。 +type memoryAsyncExtractor interface { + ExtractAsync(ctx context.Context, conversation []model.Message) (<-chan memory.AsyncResult, error) +} + +// WithoutSubagentNotes 用「剔除子代理结果注记」装饰 inner。inner 为 nil 时返回 nil +// （未配置提取器的部署因此逐字不变：循环那边同样是 nil 即跳过）。 +// +// 内层支持异步（LLM 提取器）时返回的装饰器也支持异步；确定性提取器 +// （ManualExtractor）不实现它，因此保持原有的就地执行路径。 +func WithoutSubagentNotes(inner memory.Extractor) memory.Extractor { + if inner == nil { + return nil + } + sync := &subagentNoteFreeExtractor{inner: inner} + if _, ok := inner.(memoryAsyncExtractor); ok { + return &subagentNoteFreeAsyncExtractor{subagentNoteFreeExtractor: sync} + } + return sync +} + +// subagentNoteFreeExtractor 是同步装饰器：抽取前把注记从对话里剔掉。 +type subagentNoteFreeExtractor struct { + inner memory.Extractor +} + +// Extract 在剔除注记后的对话上运行内层提取器。 +func (e *subagentNoteFreeExtractor) Extract(ctx context.Context, conversation []model.Message) ([]memory.MemoryItem, error) { + return e.inner.Extract(ctx, dialogueOnly(conversation)) +} + +// subagentNoteFreeAsyncExtractor 是异步装饰器：异步通道原样透传，只是输入先被剔除。 +type subagentNoteFreeAsyncExtractor struct { + *subagentNoteFreeExtractor +} + +// ExtractAsync 把剔除注记后的对话交给内层的异步提取。内层意外不支持异步时退化为 +// 同步执行并包成一个立即关闭的通道 —— 语义仍正确，只是不再后台跑（该分支不可达： +// 构造器只在它支持异步时才选本类型）。 +func (e *subagentNoteFreeAsyncExtractor) ExtractAsync(ctx context.Context, conversation []model.Message) (<-chan memory.AsyncResult, error) { + if async, ok := e.inner.(memoryAsyncExtractor); ok { + return async.ExtractAsync(ctx, dialogueOnly(conversation)) + } + out := make(chan memory.AsyncResult, 1) + items, err := e.Extract(ctx, conversation) + out <- memory.AsyncResult{Items: items, Err: err} + close(out) + return out, nil +} + +// dialogueOnly 返回剔除全部子代理结果注记后的对话副本。没有注记时原样返回入参 +// （不做任何分配 —— 绝大多数回合因此零成本）。 +func dialogueOnly(conversation []model.Message) []model.Message { + first := -1 + for i, m := range conversation { + if isSubagentNote(m) { + first = i + break + } + } + if first < 0 { + return conversation + } + out := make([]model.Message, 0, len(conversation)-(first/2+1)) + out = append(out, conversation[:first]...) + for _, m := range conversation[first+1:] { + if isSubagentNote(m) { + continue + } + out = append(out, m) + } + return out +} + +// isSubagentNote 报告一条消息是否是子代理结果注记。判据是**元数据 + 结构性标记**： +// `role == system` **且** 正文以 `subagentNoteMarker` 开头。只认标记而不认 role 会 +// 误伤主人自己打的那串字；只认 role 会把压缩摘要（同样是 system 角色，且是对话的 +// 合法压缩，理应继续参与抽取）一起挡掉。 +func isSubagentNote(m model.Message) bool { + if m.Role != model.RoleSystem { + return false + } + text, _ := messageTextImage(m) + return strings.HasPrefix(strings.TrimSpace(text), subagentNoteMarker) +} diff --git a/internal/services/gateway/subagent_backfill_extract_test.go b/internal/services/gateway/subagent_backfill_extract_test.go new file mode 100644 index 0000000..7131742 --- /dev/null +++ b/internal/services/gateway/subagent_backfill_extract_test.go @@ -0,0 +1,208 @@ +package gateway + +import ( + "context" + "strings" + "sync" + "testing" + + "git.yeij.top/AskaEth/Cyrene_v2/internal/bus" + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/history" + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/memory" + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/model" + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/orchestration" + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/transcript" +) + +// 本文件锁住 M32 的第二道红线：**子代理结果绝不进记忆抽取**。 +// +// 它不是「提示词里让模型别提取」，而是一条元数据判据（role=system + 注记标记）： +// 抽取器的输入在到达内层之前就被剔干净。测试因此断言的是**内层实际收到了什么** +// —— 这正是缺陷的判定点（那条线刚查过：84% 记忆来自内部会话，不能再添污染源）。 + +// captureExtractor 记录每次抽取收到的对话，返回预设条目。 +type captureExtractor struct { + mu sync.Mutex + seen [][]model.Message + items []memory.MemoryItem +} + +func (e *captureExtractor) Extract(_ context.Context, conversation []model.Message) ([]memory.MemoryItem, error) { + e.mu.Lock() + defer e.mu.Unlock() + e.seen = append(e.seen, append([]model.Message(nil), conversation...)) + return e.items, nil +} + +func (e *captureExtractor) last() []model.Message { + e.mu.Lock() + defer e.mu.Unlock() + if len(e.seen) == 0 { + return nil + } + return e.seen[len(e.seen)-1] +} + +// captureAsyncExtractor 是带异步能力的捕获提取器（对应 memory.LLMExtractor）。 +type captureAsyncExtractor struct { + *captureExtractor +} + +func (e *captureAsyncExtractor) ExtractAsync(ctx context.Context, conversation []model.Message) (<-chan memory.AsyncResult, error) { + items, err := e.Extract(ctx, conversation) + out := make(chan memory.AsyncResult, 1) + out <- memory.AsyncResult{Items: items, Err: err} + close(out) + return out, nil +} + +// noteMessage 返回一条与回填落库行**同源**的系统注记模型消息：它经过 +// `transcript.Message` 的映射，因此走的就是下一回合真正播种的那条路。 +func noteMessage(t *testing.T) model.Message { + t.Helper() + row := history.Message{ + ID: "msg-note", + SessionID: "web", + Role: history.RoleSystem, + Content: subagentNote(settledResult("web", orchestration.SpawnAgentResult{ID: "sub-a", Status: "done", Summary: "子代理的结论"})), + } + return transcript.Message(row) +} + +// userMessage 构造一条普通主人输入。 +func userMessage(text string) model.Message { + return model.TextMessage(model.RoleUser, text) +} + +// TestWithoutSubagentNotesStripsBackfilledNotes 锁住同步路径：注记在内层提取器看到 +// 之前就被剔掉，其余对话逐字保留，提取结果原样透传。 +func TestWithoutSubagentNotesStripsBackfilledNotes(t *testing.T) { + t.Parallel() + inner := &captureExtractor{items: []memory.MemoryItem{{Content: "记住的事"}}} + wrapped := WithoutSubagentNotes(inner) + conv := []model.Message{ + userMessage("你好"), + noteMessage(t), + model.TextMessage(model.RoleAssistant, "在的"), + } + items, err := wrapped.Extract(context.Background(), conv) + if err != nil { + t.Fatalf("Extract: %v", err) + } + if len(items) != 1 || items[0].Content != "记住的事" { + t.Fatalf("items = %+v, want the inner extractor's result passed through", items) + } + seen := inner.last() + if len(seen) != 2 { + t.Fatalf("inner extractor saw %d messages, want 2 (the note must be removed): %+v", len(seen), seen) + } + for _, m := range seen { + if strings.HasPrefix(modelText(m), subagentNoteMarker) { + t.Fatalf("inner extractor saw the sub-agent note %q: it would be ingested as dialogue", modelText(m)) + } + } + if modelText(seen[1]) != "在的" { + t.Fatalf("conversation order changed: %+v", seen) + } +} + +// TestWithoutSubagentNotesKeepsCompactionSummary 锁住**不过度剔除**：压缩摘要同样是 +// system 角色（loop 的 summaryMarker），它是对话的合法压缩，必须继续参与抽取。 +func TestWithoutSubagentNotesKeepsCompactionSummary(t *testing.T) { + t.Parallel() + inner := &captureExtractor{} + wrapped := WithoutSubagentNotes(inner) + summary := model.TextMessage(model.RoleSystem, "[前文摘要 · history summary]\n\n之前聊过口味") + if _, err := wrapped.Extract(context.Background(), []model.Message{summary, userMessage("我喜欢淡一点")}); err != nil { + t.Fatalf("Extract: %v", err) + } + if got := len(inner.last()); got != 2 { + t.Fatalf("inner extractor saw %d messages, want both the summary and the user turn", got) + } +} + +// TestWithoutSubagentNotesKeepsAsync 锁住「装饰之后仍然异步」：慢提取绝不能从后台 +// goroutine 掉回回合落定的同步路径上（那会直接拖慢回复）。 +func TestWithoutSubagentNotesKeepsAsync(t *testing.T) { + t.Parallel() + inner := &captureAsyncExtractor{&captureExtractor{}} + wrapped := WithoutSubagentNotes(inner) + async, ok := wrapped.(memoryAsyncExtractor) + if !ok { + t.Fatal("the decorator dropped ExtractAsync: a slow LLM extraction would now block turn settlement") + } + ch, err := async.ExtractAsync(context.Background(), []model.Message{noteMessage(t), userMessage("你好")}) + if err != nil { + t.Fatalf("ExtractAsync: %v", err) + } + <-ch + seen := inner.last() + if len(seen) != 1 || modelText(seen[0]) != "你好" { + t.Fatalf("async extractor saw %+v, want only the user turn", seen) + } +} + +// TestWithoutSubagentNotesIsNoopWithoutNotes 锁住零成本路径：没有注记时原样返回入参 +// （不分配、不改顺序），且 nil 提取器仍是 nil（未配置提取器的部署逐字不变）。 +func TestWithoutSubagentNotesIsNoopWithoutNotes(t *testing.T) { + t.Parallel() + if got := WithoutSubagentNotes(nil); got != nil { + t.Fatalf("WithoutSubagentNotes(nil) = %v, want nil", got) + } + conv := []model.Message{userMessage("你好")} + got := dialogueOnly(conv) + if len(got) != len(conv) { + t.Fatalf("dialogueOnly changed a note-free conversation: %+v", got) + } + if &got[0] != &conv[0] { + t.Fatal("a note-free conversation must be returned as-is (no copy on the hot path)") + } +} + +// TestBackfilledNoteStaysOutOfMemoryExtraction 是这条防线的**端到端**接缝断言： +// 真回填落库 → 转录播种 → 抽取。注记在整条链上看得到（历史与模型上下文都有它）， +// 但抽取器**一次也看不到**它。 +func TestBackfilledNoteStaysOutOfMemoryExtraction(t *testing.T) { + t.Parallel() + b := bus.New() + t.Cleanup(func() { _ = b.Close() }) + store := ¬eStore{} + bf, err := NewSubagentBackfill(SubagentBackfillDeps{Bus: b, History: store}) + if err != nil { + t.Fatalf("NewSubagentBackfill: %v", err) + } + if err := bf.Handle(context.Background(), bus.Event{ + Subject: orchestration.SubagentResultSubject, + Payload: settledResult("web", orchestration.SpawnAgentResult{ID: "sub-a", Status: "done", Summary: "子代理的结论"}), + }); err != nil { + t.Fatalf("Handle: %v", err) + } + rows := store.rows() + if len(rows) != 1 { + t.Fatalf("rows = %+v, want one", rows) + } + // 下一回合的模型上下文 = 播种的转录行（这就是 loop 交给提取器的 msgs 的来源）。 + nextTurn := []model.Message{transcript.Message(rows[0]), userMessage("接着说")} + inner := &captureExtractor{} + if _, err := WithoutSubagentNotes(inner).Extract(context.Background(), nextTurn); err != nil { + t.Fatalf("Extract: %v", err) + } + seen := inner.last() + if len(seen) != 1 { + t.Fatalf("extractor input = %+v, want only the user turn", seen) + } + if !strings.Contains(modelText(seen[0]), "接着说") { + t.Fatalf("extractor lost the real dialogue: %+v", seen) + } +} + +// modelText 返回一条模型消息的正文（测试窄视图）。 +func modelText(m model.Message) string { + var b strings.Builder + for _, c := range m.Content { + if c.Type == model.ContentTypeText { + b.WriteString(c.Text) + } + } + return b.String() +} diff --git a/internal/services/gateway/subagent_backfill_note.go b/internal/services/gateway/subagent_backfill_note.go new file mode 100644 index 0000000..551533e --- /dev/null +++ b/internal/services/gateway/subagent_backfill_note.go @@ -0,0 +1,196 @@ +package gateway + +import ( + "fmt" + "strings" + + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/orchestration" +) + +// 本文件拥有「后台批次结果写成什么一行注记」这一件事（subagent_backfill.go 的正文 +// 构造）。它单独成篇的理由与 history_write.go 拆自 history.go 相同：**措辞与上限** +// 是可以在评审里被逐字审的东西，与总线/落库的生命周期无关。 +// +// 三条硬口径： +// +// 1. **必须点名 `sub-`**：归属可追溯（design/05 §3 的可审计访问也依赖它）。 +// 读者与模型都要能分辨「这句是哪个子代理说的」，而不是她说的。 +// 2. **绝不写成人格发言**：没有「我说」「我让 XX 去做」，也没有人设口吻 —— 注记是 +// 一条**系统事实**，作者是 system（project 红线：人格 = 身份）。 +// 3. **有界**：整条注记 ≤ maxSubagentNoteRunes（单代理正文 ≤ maxSubagentNoteBodyRunes）， +// 使一条失控的子代理无法把父会话上下文淹掉（与 maxSpawnTextRunes 同一理由）。 + +const ( + // subagentNoteMarker 标记「这是一条子代理结果注记」。它是**结构性标记**：同一个 + // 包里的 `WithoutSubagentNotes`（subagent_backfill_extract.go）按它把注记挡在 + // 记忆抽取之外，两处共用这一个常量，因此判据不会漂移。取法照抄 loop 的 + // `summaryMarker`（压缩摘要的系统消息同一手法）。 + subagentNoteMarker = "[子代理结果]" + + // maxSubagentNoteRunes 是整条注记的上限。批次上限是 8 个子代理 + // （orchestration 的 maxSpawnBatch），逐个正文 400 rune 时整条仍会被收到这里。 + maxSubagentNoteRunes = 2000 + // maxSubagentNoteBodyRunes 是单个子代理结论的上限：够读出结论，不够塞进整篇。 + maxSubagentNoteBodyRunes = 400 + // maxSubagentNoteGoalRunes 是注记里回显的目标上限（目标只用来认出是哪个任务）。 + maxSubagentNoteGoalRunes = 120 + // maxSubagentNoteSummaryRunes 是批次汇总额的上限。 + maxSubagentNoteSummaryRunes = 400 + // subagentNoteEmptyBody 是「该子代理什么都没写」时的诚实占位，绝不空着让读者 + // 以为注记被截断了。 + subagentNoteEmptyBody = "（无结论正文）" +) + +// subagentNote 把一条后台批次结果渲染成父会话的注记正文。单代理批次写成一行 +// （最常见、也最可读）；多代理批次写汇总行 + 逐代理一行。 +func subagentNote(ev *orchestration.SubagentResultEvent) string { + if ev == nil || ev.SpawnResult == nil { + return subagentNoteMarker + } + var b strings.Builder + b.WriteString(subagentNoteMarker) + b.WriteString(" ") + if len(ev.Agents) == 1 { + b.WriteString(subagentNoteSingleLine(ev.Agents[0])) + return clampNote(b.String(), maxSubagentNoteRunes) + } + b.WriteString(subagentNoteBatchLine(ev)) + for _, a := range ev.Agents { + b.WriteString("\n- ") + b.WriteString(subagentNoteAgentLine(a)) + } + return clampNote(b.String(), maxSubagentNoteRunes) +} + +// subagentNoteSingleLine 渲染单代理批次的正文行： +// `子代理 sub-x（role/mode）已完成（目标：…）：结论`。 +func subagentNoteSingleLine(a orchestration.SpawnAgentResult) string { + var b strings.Builder + b.WriteString("子代理 ") + b.WriteString(subagentAgentName(a.ID)) + if bind := subagentAgentBinding(a); bind != "" { + b.WriteString("（") + b.WriteString(bind) + b.WriteString("）") + } + b.WriteString(" ") + b.WriteString(subagentStatusWord(a.Status)) + if goal := strings.TrimSpace(a.Goal); goal != "" { + fmt.Fprintf(&b, "（目标：%s）", clampNote(goal, maxSubagentNoteGoalRunes)) + } + b.WriteString("：") + b.WriteString(subagentAgentBody(a)) + return b.String() +} + +// subagentNoteBatchLine 渲染批次汇总行：`2 个子代理批次已完成：<汇总>`。 +func subagentNoteBatchLine(ev *orchestration.SubagentResultEvent) string { + var b strings.Builder + fmt.Fprintf(&b, "%d 个子代理批次%s", len(ev.Agents), subagentStatusWord(ev.Status)) + if summary := strings.TrimSpace(ev.Summary); summary != "" { + b.WriteString("：") + b.WriteString(clampNote(summary, maxSubagentNoteSummaryRunes)) + } + return b.String() +} + +// subagentNoteAgentLine 渲染批次里的一行：`sub-x（role/mode）已完成：结论`。 +func subagentNoteAgentLine(a orchestration.SpawnAgentResult) string { + var b strings.Builder + b.WriteString(subagentAgentName(a.ID)) + if bind := subagentAgentBinding(a); bind != "" { + b.WriteString("（") + b.WriteString(bind) + b.WriteString("）") + } + b.WriteString(" ") + b.WriteString(subagentStatusWord(a.Status)) + b.WriteString("：") + b.WriteString(subagentAgentBody(a)) + return b.String() +} + +// subagentAgentName 返回注记里点名的子代理标识。id 为空时退化为 `sub-?`（绝不留空， +// 否则「谁说的」就丢了）。 +func subagentAgentName(id string) string { + if id = strings.TrimSpace(id); id != "" { + return id + } + return "sub-?" +} + +// subagentAgentBinding 返回该子代理绑定的角色/模式标签（只作归属，不冒充发言人）。 +func subagentAgentBinding(a orchestration.SpawnAgentResult) string { + parts := make([]string, 0, 2) + if role := strings.TrimSpace(a.Role); role != "" { + parts = append(parts, role) + } + if mode := strings.TrimSpace(a.Mode); mode != "" { + parts = append(parts, mode) + } + return strings.Join(parts, "/") +} + +// subagentAgentBody 取该子代理可读的结论：优先执行摘要，其次结果正文；两者都没有 +// 时给一个诚实的占位。 +func subagentAgentBody(a orchestration.SpawnAgentResult) string { + for _, s := range []string{a.Summary, a.Text} { + if s = strings.TrimSpace(s); s != "" { + return clampNote(collapseBlankLines(s), maxSubagentNoteBodyRunes) + } + } + return subagentNoteEmptyBody +} + +// subagentStatusWord 把编排层的生命周期状态翻成注记里可读的中文。未知状态原样保留 +// （绝不把它当成「完成」—— 那正是只看状态会误判的地方）。 +func subagentStatusWord(status string) string { + switch status { + case "done": + return "已完成" + case "running": + return "仍在运行" + case "failed": + return "失败" + case "aborted": + return "已中止" + case "": + return "已结束" + default: + return status + } +} + +// collapseBlankLines 把连续空行收成一行空行，使注记在气泡里不会拉出一大片空白 +// （子代理正文常带 markdown 段落间距）。 +func collapseBlankLines(s string) string { + var b strings.Builder + blank := 0 + for _, line := range strings.Split(s, "\n") { + if strings.TrimSpace(line) == "" { + blank++ + if blank > 1 { + continue + } + } else { + blank = 0 + } + if b.Len() > 0 { + b.WriteString("\n") + } + b.WriteString(strings.TrimRight(line, " \t\r")) + } + return strings.TrimSpace(b.String()) +} + +// clampNote 把 s 收到 n 个 rune 以内（超出时以省略号收尾）。n 非正时返回空串。 +func clampNote(s string, n int) string { + if n <= 0 { + return "" + } + r := []rune(s) + if len(r) <= n { + return s + } + return string(r[:n-1]) + "…" +} diff --git a/internal/services/gateway/subagent_backfill_test.go b/internal/services/gateway/subagent_backfill_test.go new file mode 100644 index 0000000..b0678d3 --- /dev/null +++ b/internal/services/gateway/subagent_backfill_test.go @@ -0,0 +1,344 @@ +package gateway + +import ( + "context" + "errors" + "io" + "log/slog" + "strings" + "sync" + "testing" + "time" + + "git.yeij.top/AskaEth/Cyrene_v2/internal/bus" + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/history" + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/model" + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/orchestration" + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/transcript" +) + +// 本文件锁住 M32 的两条面（design/05 §2/§5/§9）与它们**不互相冒充**这条红线： +// +// - 可观测 / 历史面：父会话多出**恰好一条** `role=system` 的注记行（不是 user、 +// 不是 assistant ⇒ 不是「她说的」），并随既有的 `message_persisted` 扇出； +// - 模型面：同一行经转录播种后在下一回合是一条 **system** 消息，且**绝不**进入 +// 记忆抽取的输入面。 +// +// 每条断言都对着一个可能被写坏的实现：把角色写成 assistant（人格冒充）、把 +// `turn_id` 编一个（冒充某一路回合）、把注记漏掉标记（于是它会被当成对话吞进记忆 +// 库）、或者落库失败还照样广播（假的「已回填」）。 + +// noteStore 是回填落库面的测试替身：记录每次写入并按需注入失败。 +type noteStore struct { + mu sync.Mutex + written []history.Message + err error +} + +func (s *noteStore) AppendFull(_ context.Context, msg history.Message) (history.Message, error) { + s.mu.Lock() + defer s.mu.Unlock() + if s.err != nil { + return history.Message{}, s.err + } + if msg.ID == "" { + msg.ID = "msg-backfill-1" + } + if msg.CreatedAt == 0 { + msg.CreatedAt = 1 + } + s.written = append(s.written, msg) + return msg, nil +} + +func (s *noteStore) rows() []history.Message { + s.mu.Lock() + defer s.mu.Unlock() + return append([]history.Message(nil), s.written...) +} + +// noteSink 是下行扇出的测试替身。 +type noteSink struct { + mu sync.Mutex + sent []history.Message +} + +func (s *noteSink) BroadcastPersisted(msg history.Message) { + s.mu.Lock() + defer s.mu.Unlock() + s.sent = append(s.sent, msg) +} + +func (s *noteSink) frames() []history.Message { + s.mu.Lock() + defer s.mu.Unlock() + return append([]history.Message(nil), s.sent...) +} + +// recordSink 收集结构化日志记录，使「失败必须可见」能被断言而不是靠人看输出。 +type recordSink struct { + mu sync.Mutex + recs []slog.Record +} + +func (r *recordSink) Enabled(context.Context, slog.Level) bool { return true } + +func (r *recordSink) Handle(_ context.Context, rec slog.Record) error { + r.mu.Lock() + defer r.mu.Unlock() + r.recs = append(r.recs, rec.Clone()) + return nil +} + +func (r *recordSink) WithAttrs([]slog.Attr) slog.Handler { return r } +func (r *recordSink) WithGroup(string) slog.Handler { return r } + +// has 报告是否有一条指定级别且 msg 含 want 的记录。 +func (r *recordSink) has(level slog.Level, want string) bool { + r.mu.Lock() + defer r.mu.Unlock() + for _, rec := range r.recs { + if rec.Level == level && strings.Contains(rec.Message, want) { + return true + } + } + return false +} + +// backfillFor 构建一个带真 bus（不订阅）的回填订阅者与它的两个替身。 +func backfillFor(t *testing.T, log *slog.Logger) (*SubagentBackfill, *noteStore, *noteSink) { + t.Helper() + b := bus.New() + t.Cleanup(func() { _ = b.Close() }) + store, sink := ¬eStore{}, ¬eSink{} + if log == nil { + log = slog.New(slog.NewTextHandler(io.Discard, nil)) + } + bf, err := NewSubagentBackfill(SubagentBackfillDeps{Bus: b, History: store, Broadcast: sink, Log: log}) + if err != nil { + t.Fatalf("NewSubagentBackfill: %v", err) + } + return bf, store, sink +} + +// settledResult 构造一个与 SpawnTool.await 同形同键集的已结算批次载荷。 +func settledResult(session string, agents ...orchestration.SpawnAgentResult) *orchestration.SubagentResultEvent { + return &orchestration.SubagentResultEvent{ + SessionID: session, + SpawnResult: &orchestration.SpawnResult{ + Status: "done", + Summary: "子代理汇总: 共 1 个, 1 成功", + Agents: agents, + }, + } +} + +// TestSubagentBackfillWritesSystemNoteOnly 是 M32 的第一条面：结果写成父会话的一条 +// **系统注记**，绝不写成她的人格发言，也不冒充任何一路回合。 +func TestSubagentBackfillWritesSystemNoteOnly(t *testing.T) { + t.Parallel() + bf, store, sink := backfillFor(t, nil) + + payload := settledResult("web", orchestration.SpawnAgentResult{ + ID: "sub-a1", Goal: "调研 X", Role: "cyrene", Mode: "engineer", + Status: "done", Summary: "结论：X 可行", Text: "结论：X 可行", Steps: 2, + }) + if err := bf.Handle(context.Background(), bus.Event{Subject: orchestration.SubagentResultSubject, Payload: payload}); err != nil { + t.Fatalf("Handle: %v", err) + } + + rows := store.rows() + if len(rows) != 1 { + t.Fatalf("persisted %d rows, want exactly one system note (no fabricated speech): %+v", len(rows), rows) + } + row := rows[0] + if row.SessionID != "web" { + t.Fatalf("row session = %q, want the parent session %q", row.SessionID, "web") + } + if row.Role != history.RoleSystem { + t.Fatalf("row role = %q, want %q (the note is not the owner's words and not the persona's)", row.Role, history.RoleSystem) + } + // 显式锁住「不冒充」：既不是 user、也不是 assistant —— 这两条是本条红线的全部内容。 + for _, forbidden := range []string{history.RoleUser, history.RoleAssistant} { + if row.Role == forbidden { + t.Fatalf("row role = %q: a sub-agent conclusion must never be written as %s", row.Role, forbidden) + } + } + if !strings.HasPrefix(row.Content, subagentNoteMarker) { + t.Fatalf("content = %q, want the %q marker (it is what keeps the note out of memory extraction)", row.Content, subagentNoteMarker) + } + for _, want := range []string{"sub-a1", "cyrene/engineer", "结论：X 可行"} { + if !strings.Contains(row.Content, want) { + t.Fatalf("content = %q, want it to name %q (attribution must be traceable)", row.Content, want) + } + } + if row.TurnID != "" { + t.Fatalf("turn_id = %q, want empty: a background batch may settle after its turn ended and belongs to no lane", row.TurnID) + } + if row.PersonaID != "" { + t.Fatalf("persona_id = %q, want empty: the note's author is no persona", row.PersonaID) + } + + frames := sink.frames() + if len(frames) != 1 || frames[0].ID != row.ID { + t.Fatalf("broadcast %+v, want exactly the persisted row %q", frames, row.ID) + } +} + +// TestSubagentBackfillNoteIsBoundedAndPerAgent 锁住注记的**形状**：批次逐代理一行、 +// 每行点名 id，且整条有界（失控子代理不能淹掉父会话上下文）。 +func TestSubagentBackfillNoteIsBoundedAndPerAgent(t *testing.T) { + t.Parallel() + long := strings.Repeat("长", maxSubagentNoteBodyRunes*3) + payload := settledResult("web", + orchestration.SpawnAgentResult{ID: "sub-a", Status: "done", Summary: "结论A"}, + orchestration.SpawnAgentResult{ID: "sub-b", Status: "failed", Summary: long}, + orchestration.SpawnAgentResult{ID: "sub-c", Status: "aborted"}, + ) + note := subagentNote(payload) + for _, want := range []string{"[子代理结果] 3 个子代理批次已完成", "- sub-a 已完成：结论A", "- sub-b 失败：", "已中止", subagentNoteEmptyBody} { + if !strings.Contains(note, want) { + t.Fatalf("note = %q, want it to contain %q", note, want) + } + } + if got := len([]rune(note)); got > maxSubagentNoteRunes { + t.Fatalf("note is %d runes, want at most %d", got, maxSubagentNoteRunes) + } + // 单代理批次的正文也不能无限长。 + single := subagentNote(settledResult("web", orchestration.SpawnAgentResult{ID: "sub-x", Status: "done", Text: long})) + if got := len([]rune(single)); got > maxSubagentNoteRunes { + t.Fatalf("single-agent note is %d runes, want at most %d", got, maxSubagentNoteRunes) + } + if !strings.Contains(single, "子代理 sub-x 已完成") { + t.Fatalf("single-agent note = %q, want the one-line shape", single) + } +} + +// TestSubagentBackfillSkipsUnusablePayloads 锁住两种「不能写」的情形：载荷形状不对， +// 或没有父会话可归属。两者都必须留一条 WARN（绝不静默丢结果）。 +func TestSubagentBackfillSkipsUnusablePayloads(t *testing.T) { + t.Parallel() + sinkLog := &recordSink{} + bf, store, sink := backfillFor(t, slog.New(sinkLog)) + + cases := []struct { + name string + subject string + payload any + }{ + {name: "wrong payload type", subject: orchestration.SubagentResultSubject, payload: map[string]any{"sessionId": "web"}}, + {name: "no spawn result", subject: orchestration.SubagentResultSubject, payload: &orchestration.SubagentResultEvent{SessionID: "web"}}, + {name: "no parent session", subject: orchestration.SubagentResultSubject, payload: settledResult("", orchestration.SpawnAgentResult{ID: "sub-a", Status: "done"})}, + {name: "other subject", subject: "task.done", payload: settledResult("web")}, + } + for _, tc := range cases { + if err := bf.Handle(context.Background(), bus.Event{Subject: tc.subject, Payload: tc.payload}); err != nil { + t.Fatalf("%s: Handle: %v", tc.name, err) + } + } + if rows := store.rows(); len(rows) != 0 { + t.Fatalf("persisted %+v, want no rows for unusable payloads", rows) + } + if frames := sink.frames(); len(frames) != 0 { + t.Fatalf("broadcast %+v, want no frames for unusable payloads", frames) + } + if !sinkLog.has(slog.LevelWarn, "unexpected payload") { + t.Fatal("a malformed payload must leave a WARN, not vanish silently") + } + if !sinkLog.has(slog.LevelWarn, "no parent session") { + t.Fatal("a result with no parent session must leave a WARN, not vanish silently") + } + if !sinkLog.has(slog.LevelWarn, "carries no spawn result") { + t.Fatal("a payload without a spawn result must leave a WARN (the bus swallows panics)") + } +} + +// TestSubagentBackfillStoreFailureIsNotSilent 锁住「落库失败 ⇒ 既不广播也不静默」： +// 广播一条并不存在的行会让客户端以为回填成立（假成功）。 +func TestSubagentBackfillStoreFailureIsNotSilent(t *testing.T) { + t.Parallel() + sinkLog := &recordSink{} + bf, store, sink := backfillFor(t, slog.New(sinkLog)) + store.err = errors.New("db down") + + if err := bf.Handle(context.Background(), bus.Event{ + Subject: orchestration.SubagentResultSubject, + Payload: settledResult("web", orchestration.SpawnAgentResult{ID: "sub-a", Status: "done", Summary: "结论"}), + }); err != nil { + t.Fatalf("Handle: %v", err) + } + if frames := sink.frames(); len(frames) != 0 { + t.Fatalf("broadcast %+v after a failed write, want none (never claim a row that does not exist)", frames) + } + if !sinkLog.has(slog.LevelError, "persist result note") { + t.Fatal("a failed write must leave an ERROR with its source, not vanish silently") + } +} + +// TestSubagentBackfillStartSubscribesBus 锁住「Start 真的挂上总线」：真 bus 上发布 +// 一条结果 ⇒ 父会话多出一条注记行。少了这一步，整条回填就只是一个没人调用的方法。 +func TestSubagentBackfillStartSubscribesBus(t *testing.T) { + t.Parallel() + b := bus.New() + t.Cleanup(func() { _ = b.Close() }) + store := ¬eStore{} + bf, err := NewSubagentBackfill(SubagentBackfillDeps{Bus: b, History: store, Log: slog.New(slog.NewTextHandler(io.Discard, nil))}) + if err != nil { + t.Fatalf("NewSubagentBackfill: %v", err) + } + if err := bf.Start(context.Background()); err != nil { + t.Fatalf("Start: %v", err) + } + if err := b.Publish(context.Background(), orchestration.SubagentResultSubject, + settledResult("web", orchestration.SpawnAgentResult{ID: "sub-a", Status: "done", Summary: "结论"})); err != nil { + t.Fatalf("publish: %v", err) + } + deadline := time.Now().Add(2 * time.Second) + for time.Now().Before(deadline) { + if len(store.rows()) > 0 { + break + } + time.Sleep(5 * time.Millisecond) + } + rows := store.rows() + if len(rows) != 1 || rows[0].Role != history.RoleSystem { + t.Fatalf("rows = %+v, want one system note from the bus subscription", rows) + } +} + +// TestSubagentBackfillRequiresStoreAndBus 锁住装配期的诚实失败：没有落库面的订阅者 +// 只会把结果丢掉，却让「已接线」看起来成立。 +func TestSubagentBackfillRequiresStoreAndBus(t *testing.T) { + t.Parallel() + if _, err := NewSubagentBackfill(SubagentBackfillDeps{Bus: bus.New()}); err == nil { + t.Fatal("NewSubagentBackfill without a history store must fail") + } + if _, err := NewSubagentBackfill(SubagentBackfillDeps{History: ¬eStore{}}); err == nil { + t.Fatal("NewSubagentBackfill without a bus must fail") + } +} + +// TestBackfilledNoteNeverReframedAsAssistant 是两条面之间的**接缝**断言：落库行经 +// 转录播种（下一回合真正会走的那条路）之后，是一条 `model.RoleSystem` 消息 —— 不是 +// assistant，因此模型不会把它读成「她自己说过的话」。 +func TestBackfilledNoteNeverReframedAsAssistant(t *testing.T) { + t.Parallel() + bf, store, _ := backfillFor(t, nil) + if err := bf.Handle(context.Background(), bus.Event{ + Subject: orchestration.SubagentResultSubject, + Payload: settledResult("web", orchestration.SpawnAgentResult{ID: "sub-a", Status: "done", Summary: "结论"}), + }); err != nil { + t.Fatalf("Handle: %v", err) + } + rows := store.rows() + if len(rows) != 1 { + t.Fatalf("rows = %+v, want one", rows) + } + seeded := transcript.Message(rows[0]) + if seeded.Role != model.RoleSystem { + t.Fatalf("seeded role = %q, want system: the next turn must read the note as a system note", seeded.Role) + } + if seeded.SenderName != "" { + t.Fatalf("seeded sender = %q, want empty: no persona may be credited with the sub-agent's words", seeded.SenderName) + } +}