feat(gateway): 子代理结果回填父会话（系统注记）+ 记忆抽取排除（M32）
断点（M29 勘测定论）：后台批次结算后 `subagent.result` 发了、hub 也把它中继成 WS 帧，但**没有任何代码把结果回填父会话** —— `chat_messages` 无行、模型下一轮 看不到，于是「她拍了个子代理出去，但没有下文」。 回填拆成互不冒充的两条面（人格=身份红线：子代理结论绝不写成她的人格发言）： - 可观测/历史面（新文件 subagent_backfill.go）：独立订阅者（不塞进 hub）订阅 `subagent.result`，写父会话一条 `role=system` 注记（内容以 `[子代理结果]` 开头 并点名 `sub-<id>`），随后走既有 `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（契约归一建议报塔）。
This commit is contained in:
AskaEth committed
1 parent
bd49080bd1
commit
3acc81cec8
13 files changed
+1589
-4
No files matched your search
@@ -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
|
||||
}
|
||||
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
@@ -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()
|
||||
}
|
||||
@@ -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,
|
||||
|
||||
@@ -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)
|
||||
)
|
||||
@@ -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)
|
||||
}
|
||||
}
|
||||
Reference in new issue
Block a user