feat(loop): 落地 design/15 §2.5.6 并行提示(结构化 ParallelNotice + 独立 system 行,杜绝一题多答/跨端串话)

This commit is contained in:
AskaEth committed 2026-10-05 12:12:23 +08:00
1 parent a79fe9684e
commit fa850f9599
3 files changed
+423

No files matched your search

@@ -0,0 +1,283 @@
package loop
import (
"context"
"encoding/json"
"reflect"
"strings"
"testing"
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/model"
)
// 本文件是 design/15 §2.5.6「上下文快照:第 2 路能看到第 1 路的什么」的落地回归。
//
// 它钉住四件事:
// - 单路(未挂 ctx 键)**零成本、逐字不变**:视图原样返回,不注入任何东西;
// - 并行时本路拿到的是**独立 system 消息**里的结构化元数据,且**不含**对方正文
// (结构里根本没有承载正文的字段 —— 这条红线是结构性的);
// - 提示**绝不进用户正文**(用户内容只进 user 角色消息是项目红线);
// - 提示不写回 msgs ⇒ 不进 checkpoint 快照(同一份 msgs 也是记忆提取的输入面)。
// testNoticeMarker 是进行态摘要的可辨认头标;testOtherLegBody 模拟「另一路」的半成品
// 正文,用来证明它绝不会漏进本路(§2.5.6「不做的事」①:不注入对方半成品正文)。
const (
testNoticeMarker = "[并行路提示]"
testOtherLegBody = "海洋覆盖了地球表面的七成,我想先从洋流的成因说起"
)
// TestParallelNoticeContextRoundTrip 钉住 ctx 通道本身:未挂载 / nil ctx 返回 ok=false,
// 挂载后读回同一份摘要,且**零值摘要也算已挂载**(头尾两句本身就是「另有一路在跑」的
// 核心信号,不能与「没挂」混为一谈)。
func TestParallelNoticeContextRoundTrip(t *testing.T) {
t.Parallel()
if _, ok := ParallelNoticeFrom(context.Background()); ok {
t.Fatal("未挂载的 ctx 不应报告有进行态摘要")
}
if _, ok := ParallelNoticeFrom(nil); ok {
t.Fatal("nil ctx 不应报告有进行态摘要")
}
want := ParallelNotice{
OtherMessageID: "msg-7",
Step: 3,
ToolNames: []string{"search_documents", "document-read"},
Queued: 1,
}
got, ok := ParallelNoticeFrom(WithParallelNotice(context.Background(), want))
if !ok {
t.Fatal("挂载后应报告有进行态摘要")
}
if !reflect.DeepEqual(got, want) {
t.Fatalf("读回的摘要 = %+v, want %+v", got, want)
}
if _, ok := ParallelNoticeFrom(WithParallelNotice(context.Background(), ParallelNotice{})); !ok {
t.Fatal("零值摘要必须仍算已挂载(挂载即信号,与字段是否为空无关)")
}
}
// TestParallelNoticeTextVerbatim 把渲染结果与 design/15 §2.5.6 的原文逐字比对 ——
// 措辞是该设计的硬规格,任何「顺手改写」都应在此变红。
func TestParallelNoticeTextVerbatim(t *testing.T) {
t.Parallel()
got := parallelNoticeText(ParallelNotice{
OtherMessageID: "msg-9f3c",
Step: 3,
ToolNames: []string{"search_documents", "document-read"},
Queued: 1,
})
want := "[并行路提示] 该会话当前另有一路正在进行中(不打断它)。\n" +
"上一条消息 id:msg-9f3c;已进行到第 3 步;已调用工具:search_documents、document-read。\n" +
"另有 1 条消息在排队(未开始)。\n" +
"你这一路独立作答;不要复述或改写对方可能给出的内容。"
if got != want {
t.Fatalf("进行态摘要文本与 design/15 §2.5.6 逐字不符:\n got = %q\nwant = %q", got, want)
}
}
// TestParallelNoticeTextEmptyMetadataKeepsCore 钉住字段裁剪口径:空字段不渲染空段
// (不冒出「已进行到第 0 步」这类噪声),但头尾两句永远保留。
func TestParallelNoticeTextEmptyMetadataKeepsCore(t *testing.T) {
t.Parallel()
got := parallelNoticeText(ParallelNotice{})
if !strings.HasPrefix(got, parallelNoticeHeader) || !strings.HasSuffix(got, parallelNoticeTail) {
t.Fatalf("空元数据也必须保留头尾两句(核心信号): %q", got)
}
for _, noise := range []string{"上一条消息 id", "已进行到第", "在排队", "已调用工具"} {
if strings.Contains(got, noise) {
t.Fatalf("空字段不得渲染出空段 %q: %q", noise, got)
}
}
}
// TestAppendParallelNotice_NoKeyIsZeroCost 钉住单路(未挂 ctx 键)**零成本**:
// 视图长度不变,且返回的是**同一个底层数组的切片**(没有额外分配)——
// 「行为与今天逐字相同」的最强形式。
func TestAppendParallelNotice_NoKeyIsZeroCost(t *testing.T) {
t.Parallel()
l := New(Options{})
base := []model.Message{model.TextMessage(model.RoleUser, "hi")}
got := l.appendParallelNotice(context.Background(), "sess", 1, base)
if len(got) != len(base) {
t.Fatalf("未挂提示时视图长度变了: got %d, want %d", len(got), len(base))
}
if &got[0] != &base[0] {
t.Fatal("未挂提示时应原样返回同一底层数组的切片(零分配)")
}
}
// TestRun_ParallelNoticeIndependentSystemMessage 是核心用例:并行路在**每一步**都拿到
// 进行态摘要,形态是**独立的 system 消息**(恰好一条、内容与结构化渲染逐字相等),
// 且**绝不拼进用户正文**。内容等于 parallelNoticeText 这一断言同时钉死「只含元数据」:
// 任何往提示里塞正文的改动都会变红。
func TestRun_ParallelNoticeIndependentSystemMessage(t *testing.T) {
t.Parallel()
call := `{"tool_calls":[{"name":"echo","args":{"q":"x"}}]}`
ad := &mockAdapter{chat: []mgResp{
{resp: model.Response{Content: call}},
{resp: model.Response{Content: "done"}},
}}
a := newTestAgent(t, ad, dispatcherWith(t, &mockTool{name: "echo", result: "hi"}))
l := New(Options{Agent: a, MaxRounds: 8})
notice := ParallelNotice{
OtherMessageID: "msg-ocean",
Step: 1,
ToolNames: []string{"search_documents"},
Queued: 1,
}
ctx := WithParallelNotice(context.Background(), notice)
userText := "写一篇 2000 字关于海洋的作文"
if _, err := l.Run(ctx, "sess", []model.Message{model.TextMessage(model.RoleUser, userText)}); err != nil {
t.Fatalf("Run: %v", err)
}
if len(ad.reqs) != 2 {
t.Fatalf("模型调用数 = %d, want 2(工具循环两步)", len(ad.reqs))
}
for i, req := range ad.reqs {
found := 0
for _, m := range req.Messages {
text := messageText(m)
if m.Role == model.RoleUser && strings.Contains(text, testNoticeMarker) {
t.Fatalf("step %d: 提示被拼进了用户正文(红线): %q", i+1, text)
}
if !strings.Contains(text, testNoticeMarker) {
continue
}
found++
if m.Role != model.RoleSystem {
t.Fatalf("step %d: 提示必须落在独立 system 消息上,实际 role=%q", i+1, m.Role)
}
if want := parallelNoticeText(notice); text != want {
t.Fatalf("step %d: 提示内容必须逐字等于结构化渲染(只含元数据):\n got = %q\nwant = %q", i+1, text, want)
}
}
if found != 1 {
t.Fatalf("step %d: 提示出现 %d 次, want 恰好 1 次", i+1, found)
}
}
}
// TestRun_NoParallelNoticeUnchanged 钉住单路(未挂 ctx 键)的视图与今天逐字相同:
// 模型请求里根本没有进行态摘要,用户消息保持原样。
func TestRun_NoParallelNoticeUnchanged(t *testing.T) {
t.Parallel()
ad := &mockAdapter{chat: []mgResp{{resp: model.Response{Content: "answer"}}}}
a := newTestAgent(t, ad, nil)
l := New(Options{Agent: a, MaxRounds: 8})
userText := "写一篇 2000 字关于海洋的作文"
if _, err := l.Run(context.Background(), "sess", []model.Message{model.TextMessage(model.RoleUser, userText)}); err != nil {
t.Fatalf("Run: %v", err)
}
if len(ad.reqs) != 1 {
t.Fatalf("模型调用数 = %d, want 1", len(ad.reqs))
}
msgs := ad.reqs[0].Messages
if len(msgs) != 2 {
t.Fatalf("单路视图 = %d 条, want 2(agent 指令 + 用户消息,零新增)", len(msgs))
}
for _, m := range msgs {
if strings.Contains(messageText(m), testNoticeMarker) {
t.Fatalf("单路不得出现任何进行态摘要: %+v", m)
}
}
if got := messageText(msgs[1]); got != userText {
t.Fatalf("用户消息被改写: got %q, want %q", got, userText)
}
}
// TestRun_ParallelTwoLegsEachSeeOwnNotice 覆盖真实事故的形状:同会话两条路各自拿到
// **自己那份**摘要,且任一路都看不到**对方那一路**的消息内容(I4:运行中跨路不可见)——
// 正是「三路各自把同一句话当自己的题答」与「跨端串话」的回归。
func TestRun_ParallelTwoLegsEachSeeOwnNotice(t *testing.T) {
t.Parallel()
legAQuestion := "帮我写一篇 2000 字关于海洋的作文"
legBQuestion := "顺便查一下明天的天气"
// 两条路各自的摘要:A 被告知 B 在跑,B 被告知 A 在跑。
noticeA := ParallelNotice{OtherMessageID: "msg-b", Step: 1, ToolNames: []string{"search_documents"}}
noticeB := ParallelNotice{OtherMessageID: "msg-a", Step: 2, ToolNames: []string{"document-read"}}
runLeg := func(ctx context.Context, question string) []model.Message {
ad := &mockAdapter{chat: []mgResp{{resp: model.Response{Content: "answer"}}}}
a := newTestAgent(t, ad, nil)
l := New(Options{Agent: a, MaxRounds: 8})
if _, err := l.Run(ctx, "sess", []model.Message{model.TextMessage(model.RoleUser, question)}); err != nil {
t.Fatalf("Run(%q): %v", question, err)
}
if len(ad.reqs) != 1 {
t.Fatalf("Run(%q) 模型调用数 = %d, want 1", question, len(ad.reqs))
}
return ad.reqs[0].Messages
}
msgsA := runLeg(WithParallelNotice(context.Background(), noticeA), legAQuestion)
msgsB := runLeg(WithParallelNotice(context.Background(), noticeB), legBQuestion)
assertNotice := func(name string, msgs []model.Message, want ParallelNotice, foreignQuestion string) {
t.Helper()
var texts []string
for _, m := range msgs {
texts = append(texts, messageText(m))
}
joined := strings.Join(texts, "\n")
if !strings.Contains(joined, parallelNoticeText(want)) {
t.Fatalf("%s: 未拿到自己那份摘要:\n got = %q\nwant 含 = %q", name, joined, parallelNoticeText(want))
}
// I4:本路的视图里不得出现对方那一路的用户消息内容。
if strings.Contains(joined, foreignQuestion) {
t.Fatalf("%s: 看到了对方那一路的消息内容(跨路串话): %q", name, foreignQuestion)
}
// 红线:提示里绝不出现对方半成品正文(结构上也无处可放)。
if strings.Contains(joined, testOtherLegBody) {
t.Fatalf("%s: 进行态摘要混入了对方正文", name)
}
}
assertNotice("leg-A", msgsA, noticeA, legBQuestion)
assertNotice("leg-B", msgsB, noticeB, legAQuestion)
}
// TestRun_ParallelNoticeNotInCheckpoint 钉住 §2.5.6 的「不写回 msgs」:进行态摘要
// 绝不进 checkpoint 快照。同一份 msgs 也是后台记忆提取的输入面,因此本用例同时覆盖
// 「不进记忆提取输入面」。
func TestRun_ParallelNoticeNotInCheckpoint(t *testing.T) {
t.Parallel()
store := newKeyedCheckpointStore()
ad := &mockAdapter{chat: []mgResp{{resp: model.Response{Content: "answer"}}}}
a := newTestAgent(t, ad, nil)
l := New(Options{Agent: a, MaxRounds: 8, Checkpoint: store})
ctx := WithParallelNotice(context.Background(), ParallelNotice{
OtherMessageID: "msg-1", Step: 2, ToolNames: []string{"document-read"},
})
if _, err := l.Run(ctx, "sess", []model.Message{model.TextMessage(model.RoleUser, "go")}); err != nil {
t.Fatalf("Run: %v", err)
}
if len(store.saved) == 0 {
t.Fatal("回合落定后应至少写一份 checkpoint")
}
for _, cp := range store.saved {
var st checkpointState
if err := json.Unmarshal(cp.State, &st); err != nil {
t.Fatalf("unmarshal checkpoint: %v", err)
}
for _, m := range st.Messages {
if strings.Contains(messageText(m), testNoticeMarker) {
t.Fatalf("进行态摘要不得进 checkpoint 快照: %+v", m)
}
}
}
}
// TestParallelNoticeToolNamesJoinedByFullWidthComma 钉住工具名分隔符与设计一致
// (全角「、」),避免后续改动悄悄换成半角。
func TestParallelNoticeToolNamesJoinedByFullWidthComma(t *testing.T) {
t.Parallel()
got := parallelNoticeFacts(ParallelNotice{ToolNames: []string{"a", "b", "c"}})
if got != "已调用工具:a、b、c。" {
t.Fatalf("事实行 = %q, want %q", got, "已调用工具:a、b、c。")
}
}
+6
View File
@@ -134,6 +134,12 @@ func (l *Loop) driveTurn(ctx context.Context, sessionID string, msgs []model.Mes
//(design/15 §2.4 最新性)。循环绝不因此新开第二个回合(I1)。
steered = l.steerAtBoundary(ctx, sessionID, step, steered)
view := l.assembleContext(ctx, msgs, step)
// 并行路的进行态摘要(design/15 §2.5.6):本会话此刻另有一路在跑时,把
// 「结构化、有界」的进行态作为一条**独立的 system 消息**追加进本轮视图,
// 使本路知道要独立作答、不复述对方。未挂该 ctx 键的运行(单路 / 子代理 /
// 后台思考 / 未接网关)零成本、逐字不变;它只进视图、不写回 msgs ⇒ 不进
// checkpoint、不进记忆提取输入面(与 steer 注入同一口径)。
view = l.appendParallelNotice(ctx, sessionID, step, view)
if len(steered) > 0 {
view = append(view, steered...)
}
+134
View File
@@ -2,6 +2,8 @@ package loop
import (
"context"
"fmt"
"strings"
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/model"
)
@@ -101,3 +103,135 @@ func (l *Loop) steerAtBoundary(ctx context.Context, sessionID string, step int,
l.log.Info("steer injected", "session", sessionID, "step", step, "messages", len(msgs))
return append(steered, msgs...)
}
// ---------------------------------------------------------------------------
// 并行路进行态摘要(design/15 §2.5.6,主人 2026-10-05 裁决 B 的落地)
// ---------------------------------------------------------------------------
//
// 背景(真实事故):主人连发三条消息(两条来自安卓语音助手设备 `voiceassist:cyrene:v1`、
// 一条微信图片),后端真并发跑了三路;三路因为**互相看不见**,各自把同一句
// 「写 2000 字关于海洋」当成自己的题作答(三篇作文),其中一路还把别路的作文答进了
// 另一个客户端(跨端串话)。§2.5.6 的「进行态摘要」就是为了让每一路**结构化地**
// 知道「本会话此刻另有一路在跑」,从而各答各的。
//
// 它只含**元数据**(对方消息 id / 步数 / 已调用工具名 / 排队条数),结构里没有任何承载
// 对方正文的字段 —— 半句话会诱导本路去「补全」或「抢答」,也无法诚实地标注
// who said what(§2.5.6「不做的事」①)。它作为**独立的 system 消息**注入本轮视图,
// **绝不拼进任何 user 正文**(项目红线:用户内容只进 user 角色消息);这段文本也
// **不**承载任何身份 / 权限信息(身份识别靠 channel 层 actor 元数据 + 策略引擎)。
//
// 注入点与口径与 steer 注入**完全相同**(§2.5.6 明文要求):在 driveTurn 每步的
// assembleContext 之后追加进**本轮视图**,不写回 msgs ⇒ 不进 checkpoint 快照、
// 不进后台记忆提取的输入面。
// design/15 §2.5.6 定死的措辞(逐字,含标点)。头尾两句在任何字段为空时也照发 ——
// 「另有一路在跑、你独立作答」本身就是这条提示要传递的核心事实,元数据只是增益。
const (
parallelNoticeHeader = "[并行路提示] 该会话当前另有一路正在进行中(不打断它)。"
parallelNoticeTail = "你这一路独立作答;不要复述或改写对方可能给出的内容。"
)
// ParallelNotice 是本会话「另有一路正在进行」的**结构化**进行态摘要(design/15 §2.5.6)。
//
// 它由**网关**(会话编排器)在起第二路时按该路上游的实时状态构造,经
// WithParallelNotice 挂到该路的 runCtx 上;循环在每一步把它渲染成一条独立的
// system 消息(appendParallelNotice)。**空字段对应的那一段不渲染**,因此哪怕一条
// 元数据都没有,挂载这个值本身仍然是有意义的信号(头尾两句照发)。
//
// 结构里**没有**任何能承载对方半成品正文的字段 —— 这条红线是**结构性**的:
// 想拼也拼不进来。
type ParallelNotice struct {
// OtherMessageID 是另一路(或上一条)用户消息的服务端 id,作为「上一条消息 id」
// 渲染。为空时该段不渲染。
OtherMessageID string
// Step 是另一路**已进行到**的步序号(1 基)。<= 0 时该段不渲染。
Step int
// ToolNames 是另一路已经调用过的工具名,按调用顺序、已去重。为空时该段不渲染。
ToolNames []string
// Queued 是该会话仍在 pending、尚未被任何一路接手的消息条数。<= 0 时该段不渲染。
Queued int
}
// parallelNoticeKey 是并行路进行态摘要的 context key。
type parallelNoticeKey struct{}
// WithParallelNotice 把进行态摘要挂到 ctx 上,使本次 Run 每一步的模型可见视图都带上
// 一条独立的 system 提示(design/15 §2.5.6)。调用方(网关会话编排器)在**起第二路
// 之前**挂一份:摘要是该路起点的一份**快照**,中途不再改写(I4:两路之间不做运行中
// 的实时可见)。nil ctx 是空操作,ctx 原样返回 —— 未挂该键的运行(子代理 / 后台思考 /
// 未接网关的循环 / 单路)**行为逐字不变**。
func WithParallelNotice(ctx context.Context, notice ParallelNotice) context.Context {
if ctx == nil {
return ctx
}
return context.WithValue(ctx, parallelNoticeKey{}, notice)
}
// ParallelNoticeFrom 返回 ctx 上挂的进行态摘要,ok 报告是否挂载。它是
// WithParallelNotice 的读侧,供注入点与测试观察本路的提示能力。
//
// 它刻意返回 ok 而不只返回值:零值摘要也是「另有一路在跑」的有效信号(头尾两句本身
// 就是核心事实),因此「未挂载」不能与「挂了但没元数据」混为一谈。
func ParallelNoticeFrom(ctx context.Context) (ParallelNotice, bool) {
if ctx == nil {
return ParallelNotice{}, false
}
notice, ok := ctx.Value(parallelNoticeKey{}).(ParallelNotice)
return notice, ok
}
// parallelNoticeText 把进行态摘要渲染成 design/15 §2.5.6 定死的那段文本。措辞(含
// 标点与全角分隔)逐字取自设计;只有**有数据的字段**才出现在事实行里,使空字段不
// 产生空段(如「已进行到第 0 步」这种无信息量的噪声)。
func parallelNoticeText(notice ParallelNotice) string {
var b strings.Builder
b.WriteString(parallelNoticeHeader)
if facts := parallelNoticeFacts(notice); facts != "" {
b.WriteString("\n")
b.WriteString(facts)
}
if notice.Queued > 0 {
fmt.Fprintf(&b, "\n另有 %d 条消息在排队(未开始)。", notice.Queued)
}
b.WriteString("\n")
b.WriteString(parallelNoticeTail)
return b.String()
}
// parallelNoticeFacts 渲染事实行(设计示例里的
// 「上一条消息 id:msg-…;已进行到第 3 步;已调用工具:search_documents、document-read。」):
// 只列出有值的元数据段,以全角「;」分隔、以「。」收尾。一段都没有时返回空串,
// 该行整体不出现。
func parallelNoticeFacts(notice ParallelNotice) string {
parts := make([]string, 0, 3)
if notice.OtherMessageID != "" {
parts = append(parts, "上一条消息 id:"+notice.OtherMessageID)
}
if notice.Step > 0 {
parts = append(parts, fmt.Sprintf("已进行到第 %d 步", notice.Step))
}
if len(notice.ToolNames) > 0 {
parts = append(parts, "已调用工具:"+strings.Join(notice.ToolNames, "、"))
}
if len(parts) == 0 {
return ""
}
return strings.Join(parts, ";") + "。"
}
// appendParallelNotice 在**本步视图**末尾追加 §2.5.6 的进行态摘要,作为一条
// **独立的 system 消息**。它必须在 assembleContext **之后**调用(§2.5.6 的落点)。
//
// 它只改本轮视图、**不写回 msgs**:checkpoint 快照与后台记忆提取的输入面因此完全
// 看不到它(与 steer 注入同一口径,见本文件头)。ctx 上未挂摘要时它原样返回 view,
// 不做任何分配 ⇒ 单路 / 子代理 / 后台思考的路径**零成本、逐字不变**。
func (l *Loop) appendParallelNotice(ctx context.Context, sessionID string, step int, view []model.Message) []model.Message {
notice, ok := ParallelNoticeFrom(ctx)
if !ok {
return view
}
l.log.Debug("parallel notice injected", "session", sessionID, "step", step,
"other_message", notice.OtherMessageID, "other_step", notice.Step,
"tools", len(notice.ToolNames), "queued", notice.Queued)
return append(view, model.TextMessage(model.RoleSystem, parallelNoticeText(notice)))
}