diff --git a/internal/runtime/loop/parallel_notice_test.go b/internal/runtime/loop/parallel_notice_test.go new file mode 100644 index 0000000..bdb3de9 --- /dev/null +++ b/internal/runtime/loop/parallel_notice_test.go @@ -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。") + } +} diff --git a/internal/runtime/loop/run.go b/internal/runtime/loop/run.go index a061c06..0a7f19d 100644 --- a/internal/runtime/loop/run.go +++ b/internal/runtime/loop/run.go @@ -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...) } diff --git a/internal/runtime/loop/steer.go b/internal/runtime/loop/steer.go index 1a5f0b8..61cc186 100644 --- a/internal/runtime/loop/steer.go +++ b/internal/runtime/loop/steer.go @@ -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))) +}