Merge branch 'feat/m32-subagentbackfill-deliver-subagent-re' into dev
This commit is contained in:
AskaEth committed
commit
20cf3e0a93
14 files changed
+1592
-5
No files matched your search
+3
-1
@@ -159,7 +159,9 @@
|
||||
| **owner-only 档** | `ownerAuth`（owner） | **40** | 带审计的策略写、每个获取凭据的渠道端点、其余全部变更 |
|
||||
| **合计** | — | **81** | `14 + 27 + 40`（v0.92） |
|
||||
|
||||
**管理路由总数 = 14 + 27 + 40 = 81 条。**（v0.92 起；`14 + 25 + 40 = 79` 是 v0.85–v0.91 的读数）
|
||||
**管理路由总数 = 18 + 43 + 43 = 104 条（2026-10-24 实测，`routes.go` 逐条枚举）。**
|
||||
> 🔴 **本节计数曾长期偏低，已按实测改正**：`14 + 25 + 40 = 79`（v0.85–v0.91）与 `14 + 27 + 40 = 81`（v0.92）**都对不上代码**。2026-10-24 由复审逐条解析 `routes.go` 得到：**base（含 M22 前）`total=102` → 分组 `18 / 41 / 43`**；**合入 M23（coalesce 两端点）后 `total=104` → 分组 `18 / 43 / 43`**（`writeAuth` 41 → **43**，新增 `GET/PUT /api/v1/admin/channels/coalesce`）。三组口径：`viewerAuth`=18 / `writeAuth`=43 / `ownerAuth`=43，另有 `anyAuth`=3、`unwrapped`=1 未计入本表三段。
|
||||
> ⚠️ **v0.94（MCP 管理页 + 按上游分节的 `origin`）与 v0.95（coalesce 两端点 + 窗口语义）的条目尚未逐条登记**——本节计数已按实测更新，但**形状级条目仍待补**，见 §7.1。
|
||||
|
||||
> **v0.78 计数变更**：`viewerAuth` 段新增两条（A13 `GET /admin/panels`、A14 `POST /admin/plugins/{id}/actions/{action}`）⇒ 12 → **14**、总数 66 → **68**。A14 走 `viewerAuth` 中间件，但**写动作在 handler 内按 owner/admin 收紧**（路由中间件只能有一道门）—— 因此它**是**本表的一行，读写面却不同档，详见 §2.1 A14 与 §4.14。
|
||||
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
}
|
||||
@@ -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 是派发工具返回给派发模型的载荷：批次在前台窗口内结算时是它的合并结果，
|
||||
|
||||
@@ -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")
|
||||
}
|
||||
}
|
||||
@@ -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-<id>`，随后按
|
||||
// 既有 `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
|
||||
}
|
||||
@@ -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)
|
||||
}
|
||||
@@ -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()
|
||||
}
|
||||
@@ -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-<id>`**：归属可追溯（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]) + "…"
|
||||
}
|
||||
@@ -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)
|
||||
}
|
||||
}
|
||||
Reference in new issue
Block a user