AskaEth
eb1828ee7a
- 只翻注释/文档字符串：剥离注释后代码行逐字节不变（机械校验只剩允许的行尾注释形态） - 裁决编号与 design 节号（02.4.3 / 11.4.2 / design/05 §9 等）逐字保留 - 复核：go build 绿；orchestration 及其子包（audit/comm）测试绿；gofmt 干净
270 lines
10 KiB
Go
270 lines
10 KiB
Go
package orchestration
|
||
|
||
import (
|
||
"context"
|
||
"encoding/json"
|
||
"fmt"
|
||
"strings"
|
||
"sync"
|
||
"testing"
|
||
"time"
|
||
|
||
"git.yeij.top/AskaEth/Cyrene_v2/internal/capability/registry"
|
||
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/agent"
|
||
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/loop"
|
||
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/model"
|
||
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/tool"
|
||
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/trace"
|
||
)
|
||
|
||
// TestSubagentBranchFoldsIntoParentGraph 是分支可视化的端到端接线测试：由调度器在父会话
|
||
// 之下驱动的真实循环，必须作为子代理泳道根出现在该会话的图里、其内部 llm/tool 步骤挂在
|
||
// 它下面；必须不开启新一轮（否则前端唯一的 flowGraph 会被替换）；也必须不以其自身 id
|
||
// 另开一张图。
|
||
func TestSubagentBranchFoldsIntoParentGraph(t *testing.T) {
|
||
t.Parallel()
|
||
capture := &graphCapture{}
|
||
flow := loop.NewFlowPublisher(capture.publish, nil)
|
||
tracer := trace.New(nil)
|
||
|
||
d, err := tool.New(registry.New())
|
||
if err != nil {
|
||
t.Fatalf("tool.New: %v", err)
|
||
}
|
||
if rerr := d.Register(&echoTool{name: "echo", result: map[string]any{"key": secretValue}}); rerr != nil {
|
||
t.Fatalf("Register: %v", rerr)
|
||
}
|
||
s := branchLoop(t, flow, tracer, d, Options{MaxParallel: 2})
|
||
|
||
flow.StartRound(parentSession, "帮我调研")
|
||
sub, serr := s.Spawn(context.Background(), AgentSpec{Goal: "调研", ParentSession: parentSession})
|
||
if serr != nil {
|
||
t.Fatalf("Spawn: %v", serr)
|
||
}
|
||
res, werr := s.Wait(context.Background(), []string{sub.ID})
|
||
if werr != nil {
|
||
t.Fatalf("Wait: %v", werr)
|
||
}
|
||
if res.Succeeded != 1 || res.Total != 1 {
|
||
t.Fatalf("merged = %d total / %d ok, want 1/1 (%+v)", res.Total, res.Succeeded, res.Agents)
|
||
}
|
||
flow.EndRound(parentSession, "结果如下")
|
||
|
||
if ids := capture.ids(parentSession); len(ids) != 1 || ids[0] != parentSession+"-1" {
|
||
t.Fatalf("published rounds for %q = %v, want exactly one round: the branch must not start its own",
|
||
parentSession, ids)
|
||
}
|
||
if capture.has(sub.ID) {
|
||
t.Fatalf("sub-agent %s opened a flow graph of its own; the branch must fold into %s", sub.ID, parentSession)
|
||
}
|
||
|
||
g := capture.graph(t, parentSession)
|
||
subNode := flowSubagentNode(t, g, sub.ID)
|
||
if subNode.Kind != "subagent" || subNode.Status != trace.FlowStatusDone {
|
||
t.Fatalf("subagent node = %+v, want kind subagent / status done", subNode)
|
||
}
|
||
// 该分支的步骤在自己的泳道内成链：llm（请求工具）、tool、llm（作答）。
|
||
chain := flowLaneChain(t, g, subNode)
|
||
if len(chain) != 3 || chain[0].Kind != "llm" || chain[1].Kind != "tool" || chain[2].Kind != "llm" {
|
||
t.Fatalf("lane chain = %s, want llm>tool>llm", chainKinds(chain))
|
||
}
|
||
llmNode, toolNode, finalNode := chain[0], chain[1], chain[2]
|
||
if in := flowNodeByID(t, g, "input"); in.Label != "帮我调研" {
|
||
t.Fatalf("input node label = %q, want the parent round's seed text", in.Label)
|
||
}
|
||
|
||
// 节点日志：每个已执行的步骤都必须带一条真实事件，且不带别的。
|
||
llmEvent := flowLogEvent(t, llmNode, loop.EventLLMDone)
|
||
if got := fmt.Sprint(llmEvent.Props["model"]); got != "mock-1" {
|
||
t.Fatalf("llm log model = %q, want mock-1 (props: %+v)", got, llmEvent.Props)
|
||
}
|
||
if got := fmt.Sprint(llmEvent.Props["stop_reason"]); got != "tool_calls" {
|
||
t.Fatalf("llm log stop_reason = %q, want tool_calls", got)
|
||
}
|
||
if got, ok := llmEvent.Props["elapsed_ms"].(int64); !ok || got < 0 {
|
||
t.Fatalf("llm log elapsed_ms = %v, want a duration in ms", llmEvent.Props["elapsed_ms"])
|
||
}
|
||
if got := fmt.Sprint(flowLogEvent(t, finalNode, loop.EventLLMDone).Props["stop_reason"]); got != "stop" {
|
||
t.Fatalf("final llm log stop_reason = %q, want stop", got)
|
||
}
|
||
toolEvent := flowLogEvent(t, toolNode, loop.EventToolDone)
|
||
if got := fmt.Sprint(toolEvent.Props["tool"]); got != "echo" {
|
||
t.Fatalf("tool log tool = %q, want echo", got)
|
||
}
|
||
if got := fmt.Sprint(toolEvent.Props["args_keys"]); !strings.Contains(got, "api_key") || !strings.Contains(got, "command") {
|
||
t.Fatalf("tool log args_keys = %q, want the argument names", got)
|
||
}
|
||
|
||
// 红线：工具调用或其结果中的任何凭据都不得被发布。
|
||
blob, merr := json.Marshal(g)
|
||
if merr != nil {
|
||
t.Fatalf("marshal graph: %v", merr)
|
||
}
|
||
if strings.Contains(string(blob), secretArg) {
|
||
t.Fatalf("published graph leaked a tool argument: %s", blob)
|
||
}
|
||
if strings.Contains(string(blob), secretValue) {
|
||
t.Fatalf("published graph leaked a tool result value: %s", blob)
|
||
}
|
||
if !strings.Contains(string(blob), "[redacted]") {
|
||
t.Fatalf("published graph carries no redaction marker; result summary was not scrubbed: %s", blob)
|
||
}
|
||
}
|
||
|
||
// TestSpawnParallelBranchesStaySeparate 断言同时运行的两个子代理落在父图的两条互不相交
|
||
// 的泳道里：每条泳道有自己的 llm→tool 链，且一条泳道里没有任何东西可从另一条到达
|
||
// （design/19 §2 泳道，也就是泳道游标存在的意义——防交叉链接失效）。
|
||
func TestSpawnParallelBranchesStaySeparate(t *testing.T) {
|
||
t.Parallel()
|
||
capture := &graphCapture{}
|
||
flow := loop.NewFlowPublisher(capture.publish, nil)
|
||
tracer := trace.New(nil)
|
||
|
||
d, err := tool.New(registry.New())
|
||
if err != nil {
|
||
t.Fatalf("tool.New: %v", err)
|
||
}
|
||
if rerr := d.Register(&echoTool{name: "echo", result: "ok", delay: 15 * time.Millisecond}); rerr != nil {
|
||
t.Fatalf("Register: %v", rerr)
|
||
}
|
||
s := branchLoop(t, flow, tracer, d, Options{MaxParallel: 2})
|
||
|
||
ctx := context.Background()
|
||
flow.StartRound(parentSession, "并行调研")
|
||
subA, aerr := s.Spawn(ctx, AgentSpec{Goal: "a", ParentSession: parentSession})
|
||
if aerr != nil {
|
||
t.Fatalf("Spawn a: %v", aerr)
|
||
}
|
||
subB, berr := s.Spawn(ctx, AgentSpec{Goal: "b", ParentSession: parentSession})
|
||
if berr != nil {
|
||
t.Fatalf("Spawn b: %v", berr)
|
||
}
|
||
res, werr := s.Wait(ctx, []string{subA.ID, subB.ID})
|
||
if werr != nil {
|
||
t.Fatalf("Wait: %v", werr)
|
||
}
|
||
if res.Succeeded != 2 {
|
||
t.Fatalf("merged = %d ok, want 2 (%+v)", res.Succeeded, res.Agents)
|
||
}
|
||
|
||
g := capture.graph(t, parentSession)
|
||
nodeA := flowSubagentNode(t, g, subA.ID)
|
||
nodeB := flowSubagentNode(t, g, subB.ID)
|
||
if nodeA.ID == nodeB.ID {
|
||
t.Fatal("both sub-agents resolved to one node")
|
||
}
|
||
setA := flowReachable(g, nodeA.ID)
|
||
setB := flowReachable(g, nodeB.ID)
|
||
for _, sub := range []trace.FlowNode{nodeA, nodeB} {
|
||
chain := flowLaneChain(t, g, sub)
|
||
if len(chain) != 3 || chain[0].Kind != "llm" || chain[1].Kind != "tool" || chain[2].Kind != "llm" {
|
||
t.Fatalf("lane %s chain = %s, want llm>tool>llm", sub.Label, chainKinds(chain))
|
||
}
|
||
}
|
||
for id := range setA {
|
||
if setB[id] {
|
||
t.Fatalf("node %s is reachable from both lanes (cross-linked edges: %+v)", id, g.Edges)
|
||
}
|
||
}
|
||
// 主干游标必须仍是父轮的种子，而不是最后那个分支节点。
|
||
if !hasFlowEdge(g, "input", nodeA.ID) || !hasFlowEdge(g, "input", nodeB.ID) {
|
||
t.Fatalf("edges = %+v, want both branch roots hanging off the parent's input node", g.Edges)
|
||
}
|
||
}
|
||
|
||
// TestSubagentRunDoesNotStreamToFrontSession 断言子代理的内部流被静音：循环仍驱动模型
|
||
// （其 llm 步骤出现在父图的泳道里），但没有任何增量抵达调用方的 OnStream，因此后台子代理
|
||
// 的独白绝不会泄漏进前台对话（design/05 §8, design/04 §2 不污染前台）。
|
||
func TestSubagentRunDoesNotStreamToFrontSession(t *testing.T) {
|
||
t.Parallel()
|
||
capture := &graphCapture{}
|
||
flow := loop.NewFlowPublisher(capture.publish, nil)
|
||
tracer := trace.New(nil)
|
||
|
||
d, err := tool.New(registry.New())
|
||
if err != nil {
|
||
t.Fatalf("tool.New: %v", err)
|
||
}
|
||
if rerr := d.Register(&echoTool{name: "echo", result: "ok"}); rerr != nil {
|
||
t.Fatalf("Register: %v", rerr)
|
||
}
|
||
ad := &toolThenFinalAdapter{toolName: "echo", args: map[string]any{"command": "date"}, final: "完成"}
|
||
ag, aerr := agent.New("be brief", d, nil, agent.WithModel(ad, nil),
|
||
agent.WithOptions(agent.Options{Model: "mock-1"}))
|
||
if aerr != nil {
|
||
t.Fatalf("agent.New: %v", aerr)
|
||
}
|
||
|
||
var mu sync.Mutex
|
||
var streamed []model.StreamEvent
|
||
lp := loop.New(loop.Options{
|
||
Agent: ag, Tools: d, Tracer: tracer, Flow: flow, MaxRounds: 8,
|
||
OnStream: func(ev model.StreamEvent) {
|
||
mu.Lock()
|
||
streamed = append(streamed, ev)
|
||
mu.Unlock()
|
||
},
|
||
})
|
||
s := newScheduler(t, Options{MaxParallel: 1, Loop: lp, Tracer: tracer, Flow: flow})
|
||
|
||
ctx := context.Background()
|
||
flow.StartRound(parentSession, "调研")
|
||
sub, serr := s.Spawn(ctx, AgentSpec{Goal: "x", ParentSession: parentSession})
|
||
if serr != nil {
|
||
t.Fatalf("Spawn: %v", serr)
|
||
}
|
||
if _, werr := s.Wait(ctx, []string{sub.ID}); werr != nil {
|
||
t.Fatalf("Wait: %v", werr)
|
||
}
|
||
|
||
mu.Lock()
|
||
got := append([]model.StreamEvent(nil), streamed...)
|
||
mu.Unlock()
|
||
if len(got) != 0 {
|
||
t.Fatalf("sub-agent pushed %d stream events into the front session: %+v", len(got), got)
|
||
}
|
||
// 该工作仍在它该在的地方可见：作为父图的一条泳道。
|
||
g := capture.graph(t, parentSession)
|
||
subNode := flowSubagentNode(t, g, sub.ID)
|
||
if chain := flowLaneChain(t, g, subNode); len(chain) == 0 {
|
||
t.Fatal("the muted sub-agent left no steps in the parent's flow lane")
|
||
}
|
||
}
|
||
|
||
// TestSpawnWithoutParentSessionStaysOutOfEveryGraph 断言流图归属是选择性的：未带
|
||
// ParentSession 的派发根本不发布任何图，因此未归属的后台 agent 绝不会作为游离泳道浮现。
|
||
func TestSpawnWithoutParentSessionStaysOutOfEveryGraph(t *testing.T) {
|
||
t.Parallel()
|
||
capture := &graphCapture{}
|
||
flow := loop.NewFlowPublisher(capture.publish, nil)
|
||
tracer := trace.New(nil)
|
||
|
||
d, err := tool.New(registry.New())
|
||
if err != nil {
|
||
t.Fatalf("tool.New: %v", err)
|
||
}
|
||
if rerr := d.Register(&echoTool{name: "echo", result: "ok"}); rerr != nil {
|
||
t.Fatalf("Register: %v", rerr)
|
||
}
|
||
s := branchLoop(t, flow, tracer, d, Options{MaxParallel: 1})
|
||
|
||
flow.StartRound(parentSession, "调研")
|
||
sub, serr := s.Spawn(context.Background(), AgentSpec{Goal: "unattributed"})
|
||
if serr != nil {
|
||
t.Fatalf("Spawn: %v", serr)
|
||
}
|
||
if _, werr := s.Wait(context.Background(), []string{sub.ID}); werr != nil {
|
||
t.Fatalf("Wait: %v", werr)
|
||
}
|
||
|
||
if capture.has(sub.ID) {
|
||
t.Fatal("an unattributed sub-agent opened a flow graph of its own")
|
||
}
|
||
g := capture.graph(t, parentSession)
|
||
for _, n := range g.Nodes {
|
||
if n.Kind == "subagent" {
|
||
t.Fatalf("node %s appeared in %q although the spawn declared no parent session", n.ID, parentSession)
|
||
}
|
||
}
|
||
}
|