AskaEth
3acc81cec8
断点（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（契约归一建议报塔）。
197 lines
8.8 KiB
Go
197 lines
8.8 KiB
Go
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
|
||
}
|