diff --git a/internal/runtime/orchestration/spawntool.go b/internal/runtime/orchestration/spawntool.go index 3fd8492..0cac9ba 100644 --- a/internal/runtime/orchestration/spawntool.go +++ b/internal/runtime/orchestration/spawntool.go @@ -42,11 +42,11 @@ const ( minSpawnTimeoutSeconds = 1 maxSpawnTimeoutSeconds = 24 * 60 * 60 - // SubagentResultSubject is the bus subject a finished background batch is - // announced on, carrying the *MergedResult. It is the design/05 §9 结果回填 hook - // for whichever component owns the session message queue (design/15); the - // WS-visible per-agent status already travels on the scheduler's own progress - // events, so this subject has no consumer until that queue lands. + // SubagentResultSubject 是后台批次结算后广播其结果的主题，载荷是 + // *SubagentResultEvent（父会话 + 批次结果）。它是 design/05 §9「结果回填主 + // 会话并提示用户」的载体，消费者是 gateway 的中继：该主题被登记为 WS 下行帧 + // subagent_result（Docs/api-client.md §3.10），帧按载荷里的父会话归属，因此结果 + // 落回父会话的实时流而非悬空 hook；逐代理进行中状态仍走调度器进度事件。 SubagentResultSubject = "subagent.result" ) @@ -191,7 +191,7 @@ func (t *SpawnTool) Execute(ctx context.Context, args any) (any, error) { "persona", specs[0].PersonaID, "mode", specs[0].Mode, "session", specs[0].ParentSession) if boolArg(m, "background") { - go t.await(ids) + go t.await(ids, specs[0].ParentSession) return runningSpawnResult(ids, specs), nil } return t.foreground(ids, specs), nil @@ -206,7 +206,7 @@ func (t *SpawnTool) foreground(ids []string, specs []AgentSpec) any { res, err := t.sched.Wait(waitCtx, ids) if err != nil { t.log.Info("sub-agent batch moved to background", "count", len(ids), "waited", t.wait.String()) - go t.await(ids) + go t.await(ids, specs[0].ParentSession) return runningSpawnResult(ids, specs) } return doneSpawnResult(res) @@ -215,7 +215,9 @@ func (t *SpawnTool) foreground(ids []string, specs []AgentSpec) any { // await blocks until the batch settles and broadcasts the merged outcome as the // design/05 §2 "结束后写执行总结 → 返回主会话" hand-off. It runs on a detached // goroutine; a nil bus or an unknown id ends it quietly. -func (t *SpawnTool) await(ids []string) { +// parentSession 是该批次所属的父会话，随载荷一起广播，使中继能把回填结果 +// 归属到父会话的实时流，而不是发一张无归属的帧。 +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) @@ -227,7 +229,8 @@ func (t *SpawnTool) await(ids []string) { if b == nil { return } - if perr := b.Publish(context.Background(), SubagentResultSubject, res); perr != nil { + 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) } } @@ -265,6 +268,13 @@ type SpawnAgentResult struct { Steps int `json:"steps,omitempty"` } +// SubagentResultEvent 是后台批次结算后发布的回填载荷（design/05 §9；契约见 +// Docs/api-client.md §3.10）：SessionID 为父会话，其余字段即派发工具返回给模型的结果。 +type SubagentResultEvent struct { + SessionID string `json:"sessionId,omitempty"` + *SpawnResult +} + // doneSpawnResult maps a settled batch onto the tool result. func doneSpawnResult(res *MergedResult) *SpawnResult { out := &SpawnResult{Status: "done", Agents: make([]SpawnAgentResult, 0)} diff --git a/internal/runtime/orchestration/spawntool_test.go b/internal/runtime/orchestration/spawntool_test.go index fb56a85..450fb07 100644 --- a/internal/runtime/orchestration/spawntool_test.go +++ b/internal/runtime/orchestration/spawntool_test.go @@ -176,10 +176,10 @@ func TestSpawnToolMovesSlowBatchToBackground(t *testing.T) { s.runner = &mockRunner{gate: gate, result: &loop.Result{Text: "late", Steps: 2}} sp := NewSpawnTool(s, WithSpawnWait(40*time.Millisecond)) - results := make(chan *MergedResult, 1) + results := make(chan *SubagentResultEvent, 1) unsub, err := s.Bus().Subscribe(context.Background(), SubagentResultSubject, func(_ context.Context, e bus.Event) error { - if res, ok := e.Payload.(*MergedResult); ok { + if res, ok := e.Payload.(*SubagentResultEvent); ok { select { case results <- res: default: @@ -192,8 +192,11 @@ func TestSpawnToolMovesSlowBatchToBackground(t *testing.T) { } defer unsub() + // 派发回合的会话（loop 在 Run 顶部把它播进 ctx）：回填载荷必须带上它， + // gateway 的中继才能把结果归属到父会话的实时流（design/05 §9）。 + ctx := loop.WithFlowBranch(context.Background(), parentSession) start := time.Now() - raw, err := sp.Execute(context.Background(), map[string]any{"goal": "slow"}) + raw, err := sp.Execute(ctx, map[string]any{"goal": "slow"}) elapsed := time.Since(start) if err != nil { t.Fatalf("Execute: %v", err) @@ -212,14 +215,32 @@ func TestSpawnToolMovesSlowBatchToBackground(t *testing.T) { close(gate) select { case res := <-results: - if res.Total != 1 || res.Succeeded != 1 { - t.Fatalf("background merge = %d total / %d ok, want 1/1", res.Total, res.Succeeded) + if res.SessionID != parentSession { + t.Fatalf("result sessionId = %q, want the dispatching session %q", res.SessionID, parentSession) } - if res.Agents[0].Text != "late" { - t.Fatalf("background result text = %q, want late", res.Agents[0].Text) + if res.Status != "done" || len(res.Agents) != 1 { + t.Fatalf("result = %+v, want one settled agent", res) + } + if a := res.Agents[0]; a.Text != "late" || a.Steps != 2 || a.Goal != "slow" || a.Status != string(StatusDone) { + t.Fatalf("result agent = %+v, want the late/2-steps/done projection", a) + } + // 载荷形状是 api-client §3.10 登记的 WS 帧 data：键集锁定，避免 + // 生产者悄悄改键而文档与实现分叉。 + wire, merr := json.Marshal(res) + if merr != nil { + t.Fatalf("marshal result payload: %v", merr) + } + var keys map[string]json.RawMessage + if uerr := json.Unmarshal(wire, &keys); uerr != nil { + t.Fatalf("unmarshal result payload: %v", uerr) + } + for _, want := range []string{"sessionId", "status", "summary", "agents"} { + if _, ok := keys[want]; !ok { + t.Fatalf("result payload %s lacks key %q", wire, want) + } } case <-time.After(waitTimeout): - t.Fatal("a background batch never published its merged result") + t.Fatal("a background batch never published its result payload") } } diff --git a/internal/services/gateway/ws_bridge.go b/internal/services/gateway/ws_bridge.go index 711fa02..bafeada 100644 --- a/internal/services/gateway/ws_bridge.go +++ b/internal/services/gateway/ws_bridge.go @@ -9,15 +9,23 @@ import ( "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/orchestration/comm" ) +// wsSubagentResult 是后台子代理批次结算后的下行帧类型（design/05 §9「结果回填 +// 主会话并提示用户」）。编排层在 orchestration.SubagentResultSubject 上发布批次 +// 结果，本文件的中继把它成形为这一帧并按载荷里的父会话归属。它属**内容帧**， +// 因此走 design/11 §7.1 的 fail-closed 默认档：只发 owner + admin（不登记进那 7 个 +// 状态机信号白名单）。 +const wsSubagentResult = "subagent_result" + // eventBridge 把生产者 bus 主题映射为其下行 WS 事件类型。hub // 订阅这些（并重新归一化为 ws.* 类型），使前端既有的 // 消费方亮起来，而生产者无需改变其主题名。 var eventBridge = map[string]string{ - "iot.device_update": "device_update", - "task.output": "task_output", - "task.done": "task_done", - "automation.result": "automation_result", - "reminder.trigger": "reminder", + "iot.device_update": "device_update", + "task.output": "task_output", + "task.done": "task_done", + "automation.result": "automation_result", + "reminder.trigger": "reminder", + orchestration.SubagentResultSubject: wsSubagentResult, } // bridgePatterns 是 hub 在 ws.> 之外额外订阅的 bus 模式，用于 @@ -28,6 +36,9 @@ var bridgePatterns = []string{ "automation.>", "reminder.>", "cyrene.agent.*.event", + // 后台子代理批次结果：它既不匹配 ws.>，也不是 cyrene.agent 的 4 段广播形态， + // 所以必须在这里登记订阅，否则后台结果永远上不了前台（只能看到进度帧）。 + orchestration.SubagentResultSubject, } // normalizeEvent 把 bus 事件映射为下行 wsEvent。ws.* 主题保持 @@ -36,6 +47,14 @@ var bridgePatterns = []string{ // ws.subagent_event。对没有 WS 可见表示的事件返回 ok=false // （例如内部的 cyrene.agent ask/reply 流量）。 func normalizeEvent(e bus.Event) (wsEvent, bool) { + // 后台批次结果先走专用成形：它要按载荷里的父会话把结果归属到该会话的实时流 + // （帧壳 sessionId），这是 eventBridge 的原样透传做不到的。载荷形状不符预期时 + // 退回下面的通用映射（两路都只发 owner + admin）。 + if e.Subject == orchestration.SubagentResultSubject { + if ev, ok := subagentResultEvent(e.Payload); ok { + return ev, true + } + } if typ, ok := eventBridge[e.Subject]; ok { return wsEvent{Type: typ, Data: e.Payload}, true } @@ -48,6 +67,18 @@ func normalizeEvent(e bus.Event) (wsEvent, bool) { return wsEvent{}, false } +// subagentResultEvent 把后台批次的结果成形为下行帧：帧壳 sessionId 取自载荷的 +// 父会话，使结果落回**父会话**的实时消息流（design/05 §9）；data 即载荷本身 +// （批次状态/汇总 + 逐代理结果，与派发工具返回给模型的结果同形）。载荷不是 +// *orchestration.SubagentResultEvent 时返回 ok=false，由调用方退回通用映射。 +func subagentResultEvent(payload any) (wsEvent, bool) { + ev, ok := payload.(*orchestration.SubagentResultEvent) + if !ok || ev == nil { + return wsEvent{}, false + } + return wsEvent{Type: wsSubagentResult, SessionID: ev.SessionID, Data: ev}, true +} + // subagentProgressEvent 把编排进度事件 —— 经 comm.PublishEvent 广播在 // cyrene.agent..event 上 —— 映射为 ws.subagent_event // 帧。只有五个子代理进度名会被浮现；同一主题上广播的 diff --git a/internal/services/gateway/ws_bridge_subagent_result_test.go b/internal/services/gateway/ws_bridge_subagent_result_test.go new file mode 100644 index 0000000..4c6212d --- /dev/null +++ b/internal/services/gateway/ws_bridge_subagent_result_test.go @@ -0,0 +1,211 @@ +package gateway + +import ( + "context" + "encoding/json" + "testing" + "time" + + "github.com/gorilla/websocket" + + "git.yeij.top/AskaEth/Cyrene_v2/internal/auth" + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/orchestration" +) + +// 本文件锁住 02.4.3 的结果回填链路：编排层在 +// orchestration.SubagentResultSubject 上发布的后台批次结果，必须经 +// ws_bridge 的中继变成 `subagent_result` 下行帧、并**按父会话归属** +// （帧壳 sessionId），从而出现在父会话的实时流里（design/05 §9）； +// 它是内容帧，因此只发 owner + admin（design/11 §7.1，fail-closed）。 +// +// 断言走真路径：真 bus Publish → hub 的真订阅与中继 → 真 /ws/chat 连接 +// （真 JWT）。viewer 的负数用例（收不到）与 owner/admin 的正数用例 +// （收得到 + 载荷形状）在同一次发布上一起判定。 + +// subagentResultReadWindow 是排空一条连接时的单帧等待窗口。广播是异步的 +// （bus worker → relay → hub → 连接写协程），窗口内应收到全部相关帧； +// 读超时即视为排空完毕。 +const subagentResultReadWindow = 1500 * time.Millisecond + +// subagentFrame 是测试侧的线上帧视图（等价于客户端解码）：帧壳 type/sessionId 直接 +// 解出，data 保留原始 JSON，使断言既能按形状解码、又能锁住键集。 +type subagentFrame struct { + Type string `json:"type"` + SessionID string `json:"sessionId"` + Data json.RawMessage `json:"data"` +} + +// subagentFrameData 是 api-client §3.10 登记的 data 形状。 +type subagentFrameData struct { + SessionID string `json:"sessionId"` + Status string `json:"status"` + Summary string `json:"summary"` + Agents []struct { + ID string `json:"id"` + Goal string `json:"goal"` + Role string `json:"role"` + Mode string `json:"mode"` + Status string `json:"status"` + Summary string `json:"summary"` + Text string `json:"text"` + Steps int `json:"steps"` + } `json:"agents"` +} + +// TestSubagentResultReachesOwnerAndAdminOnly 是后台结果回填的真 WS 帧断言： +// 真 bus 上发布编排层的真实载荷类型 → owner/admin 收到 `subagent_result` +// 帧（带父会话与逐代理结果），viewer **收不到**（内容帧默认拒）。 +func TestSubagentResultReachesOwnerAndAdminOnly(t *testing.T) { + t.Parallel() + // 新帧属内容帧：不在那 7 个状态机信号白名单里，必须落 fail-closed 的 + // owner+admin 档（design/11 §7.1「要放进白名单必须有明确理由并在本表登记」）。 + if frameReachesAll(wsSubagentResult) { + t.Fatalf("%s reached every connection; a result frame carries content and must stay owner/admin only", wsSubagentResult) + } + + const parentSessionID = "web:subagent-dispatch" + env := newEnv(t, nil, false) + + conns := map[auth.Role]*websocket.Conn{} + for _, role := range []auth.Role{auth.RoleOwner, auth.RoleAdmin, auth.RoleViewer} { + conn := dialWS(t, env.ts.URL, env.token(t, role)) + defer conn.Close() + conns[role] = conn + } + waitConnCount(t, env.srv, len(conns)) + + // 编排层发布的真实载荷（与 SpawnTool.await 同类型同键集）。 + payload := &orchestration.SubagentResultEvent{ + SessionID: parentSessionID, + SpawnResult: &orchestration.SpawnResult{ + Status: "done", + Summary: "子代理汇总: 共 1 个, 1 成功, 0 失败, 0 中止", + Agents: []orchestration.SpawnAgentResult{{ + ID: "sub-a1", Goal: "调研 X", Role: "cyrene", Mode: "engineer", + Status: "done", Summary: "完成 (步骤=2): 结论", + Text: "结论", Steps: 2, + }}, + }, + } + if err := env.bus.Publish(context.Background(), orchestration.SubagentResultSubject, payload); err != nil { + t.Fatalf("publish %s: %v", orchestration.SubagentResultSubject, err) + } + // 哨兵：一个白名单状态信号收尾。它让每条连接的「本批帧已到齐」从猜测的 + // 空闲窗口变成可判定的边界；viewer 也必定收到它，故「viewer 什么都没收到」 + // 不会与「连接根本没工作」混淆。 + if err := env.bus.Publish(context.Background(), wsPrefix+scopeSentinel, map[string]any{"sessionId": parentSessionID}); err != nil { + t.Fatalf("publish sentinel: %v", err) + } + + for _, role := range []auth.Role{auth.RoleOwner, auth.RoleAdmin} { + frames := drainFrames(t, conns[role]) + assertFrameCount(t, string(role), frames, wsSubagentResult, 1) + assertFrameCount(t, string(role), frames, scopeSentinel, 1) + if len(frames) != 2 { + t.Fatalf("%s received %d frames, want exactly the result + the sentinel: %v", role, len(frames), frameTypesOf(frames)) + } + assertSubagentResultFrame(t, string(role), findFrame(t, string(role), frames, wsSubagentResult), parentSessionID, payload) + } + + viewerFrames := drainFrames(t, conns[auth.RoleViewer]) + if len(viewerFrames) != 1 || viewerFrames[0].Type != scopeSentinel { + t.Fatalf("viewer received %v, want only the whitelisted %q: the result frame is owner/admin only (design/11 §7.1)", + frameTypesOf(viewerFrames), scopeSentinel) + } +} + +// assertSubagentResultFrame 按 api-client §3.10 登记的帧形状逐项断言一帧 +// `subagent_result`：帧壳 sessionId 必须是被回填的父会话，data 的**键集**必须 +// 恰好是登记的那四个，其汇总与逐代理字段必须与该批次发布的载荷一致。 +func assertSubagentResultFrame(t *testing.T, role string, frame subagentFrame, parentSessionID string, payload *orchestration.SubagentResultEvent) { + t.Helper() + if frame.SessionID != parentSessionID { + t.Fatalf("%s frame sessionId = %q, want the parent session %q (the result must land in that session's stream)", + role, frame.SessionID, parentSessionID) + } + wantKeys := []string{"sessionId", "status", "summary", "agents"} + var keys map[string]json.RawMessage + if err := json.Unmarshal(frame.Data, &keys); err != nil { + t.Fatalf("%s frame data is not a JSON object: %v (%s)", role, err, frame.Data) + } + if len(keys) != len(wantKeys) { + t.Fatalf("%s frame data keys = %v, want exactly %v (api-client §3.10)", role, keys, wantKeys) + } + for _, k := range wantKeys { + if _, ok := keys[k]; !ok { + t.Fatalf("%s frame data lacks key %q (api-client §3.10): %v", role, k, keys) + } + } + var data subagentFrameData + if err := json.Unmarshal(frame.Data, &data); err != nil { + t.Fatalf("%s frame data = %s: %v", role, frame.Data, err) + } + if data.SessionID != parentSessionID || data.Status != "done" { + t.Fatalf("%s frame data = %+v, want the parent session and a settled batch", role, data) + } + if data.Summary != payload.Summary { + t.Fatalf("%s frame summary = %q, want %q", role, data.Summary, payload.Summary) + } + if len(data.Agents) != 1 { + t.Fatalf("%s frame agents = %+v, want one entry", role, data.Agents) + } + if got := data.Agents[0]; got.ID != "sub-a1" || got.Goal != "调研 X" || got.Role != "cyrene" || + got.Mode != "engineer" || got.Status != "done" || got.Text != "结论" || got.Steps != 2 { + t.Fatalf("%s frame agent = %+v, want the dispatched sub-agent's outcome", role, got) + } +} + +// drainFrames 读取连接直到读超时，返回按到达顺序收到的帧。它只用于 +// 已发布的批次之后：第一批帧到达后剩余窗口用于确认「没有别的帧了」， +// 这正是 viewer 负数用例所需的判定。 +func drainFrames(t *testing.T, conn *websocket.Conn) []subagentFrame { + t.Helper() + var out []subagentFrame + for { + _ = conn.SetReadDeadline(time.Now().Add(subagentResultReadWindow)) + _, msg, err := conn.ReadMessage() + if err != nil { + return out + } + var frame subagentFrame + if uerr := json.Unmarshal(msg, &frame); uerr != nil { + t.Fatalf("decode frame %s: %v", msg, uerr) + } + out = append(out, frame) + } +} + +// assertFrameCount 断言角色收到的某类型帧恰好 want 条。 +func assertFrameCount(t *testing.T, role string, frames []subagentFrame, typ string, want int) { + t.Helper() + got := 0 + for _, f := range frames { + if f.Type == typ { + got++ + } + } + if got != want { + t.Fatalf("%s received %d %q frames, want %d (all frames: %v)", role, got, typ, want, frameTypesOf(frames)) + } +} + +// findFrame 返回角色收到的第一条 typ 帧。 +func findFrame(t *testing.T, role string, frames []subagentFrame, typ string) subagentFrame { + t.Helper() + for _, f := range frames { + if f.Type == typ { + return f + } + } + t.Fatalf("%s received no %q frame (all frames: %v)", role, typ, frameTypesOf(frames)) + return subagentFrame{} +} + +// frameTypesOf 列出收到的帧类型，用于失败信息。 +func frameTypesOf(frames []subagentFrame) []string { + out := make([]string, 0, len(frames)) + for _, f := range frames { + out = append(out, f.Type) + } + return out +}