feat(loop,gateway): 事件批 — tool_execution 三事件 + streamId + turn.start/end(裁决 01.4.1/01.4.2/01.4.3/01.4.5/03.4.4/1.12)
CI / Backend (build / vet / test / cross-compile) (push) Has been cancelled
CI / Frontend (tsc + vite build) (push) Has been cancelled

执行方案阶段 2 · P2-3(`DevDocs/audit/_plan.md`)。契约 `Docs/api-client.md` → **v0.50**。

- **01.4.1 `tool_execution` 拆三事件**:`model/stream.go` 删旧 `StreamKindToolExecution`,新增
  `tool_execution.start/update/end` + `ToolPhaseRunning/Done/Error`(与 design/19 status 同值);
  `StreamEvent` 增 `Tool`/`Phase`;`loop/tools.go` 的 `emitToolExecution` 重写为结构化字段、
  **Delta 恒空**(断言锁死),`ID` 复用 provider tool call id(可与 `tool_call.*` 关联)。
  ⚠️ **诚实边界**:start/end 有真实生产者;`update` 的 wire 映射已实现并单测锁定,但**工具分发器
  无中途进度上报源 ⇒ loop 暂不产出 update 帧**(已在 api-client §3.1 如实登记)。
- **03.4.4 phase 驱动图节点状态**:`invokeOne` 用**单一 phase 决策**同时决定 end 帧 phase、
  span 的 `phase` prop 与终态;`trace/graph.go` 新增 `FlowStatusForPhase`,`graph_build.go` 的
  `AddSpan` 对 tool 节点**优先消费 phase prop**(缺省回落 span 终态 ⇒ 两路不可能矛盾)。
- **01.4.2 `streamId`**:`StreamEvent` 增 `StreamID`;`loop/stream.go` 每次 ChatStream 生成
  `stream-<hex>`(crypto/rand,降级时间戳),该流所有帧共享、多步工具轮取新 id;`ws_bridge`
  把它并进**模型流帧** `data`(`tool_execution.*`/`turn.*` 不带)。语义区分:`turnId` = 哪轮回答,
  `streamId` = 这条流式消息。
- **01.4.3 `turn.start`/`turn.end`**:`loop/run.go` 入口发 start、defer 发 end,**覆盖全部出口**
  (最终答案/安全网终止/预算耗尽/步骤错误);帧**无 data 载荷**;静音的内部 run(backthink/子代理)
  不发。顺带把 91 行的 `Run` 拆成 `driveTurn` + `settleAndReturn`(清掉一处既有 >80 行违规)。
- **01.4.5**:仅 `Docs/api-client.md` §3.5 补 `graph_update` 三范围(`web`/`web:<uuid>`/`sess-backthink`),
  与 `wire_runtime.go` 的 allow 谓词一致,**零代码变更**。

验证:`go build ./...` 绿;本批四包(loop/model/trace/gateway)vet + 测试绿(含 `-count=1`);
linux/amd64 + darwin/arm64 交叉编译绿;本批 16 文件 ≤500 行、函数 ≤80 行。
**真 WS 帧断言 8 条**:帧序 `turn.start → message.delta → tool_execution.start →
tool_execution.end → message.delta → turn.end`(+`stream_end`)、turn 边界无载荷且整轮 turnId 一致、
两条模型流 `streamId` 互异、tool 帧 `{tool,phase,id}` 且无 `delta`/`streamId` 键、
`turn.end` 在步骤出错时也发出、内部静音 run 不发边界帧。
**变异验证**:把 `ev.StreamID = streamID` 改为赋空 → 两条断言 FAIL(`streamId` 为空/两流同空),
还原后复绿(直接删行会先被编译期 `declared and not used` 拦截)。

未验证:前端未消费这些帧(本轮未动 `frontend/**`,「前端能渲染」待前端批);无真实模型端到端。
This commit is contained in:
AskaEth committed 2026-10-04 11:26:35 +08:00
1 parent afef75f640
commit 6e7ec56f9a
16 files changed
+982 -126

No files matched your search

+2 -2
View File
@@ -319,12 +319,12 @@ func TestRun_StreamToolExecution_EmitsProgress(t *testing.T) {
defer mu.Unlock()
saw := false
for _, ev := range evs {
if ev.Kind == model.StreamKindToolExecution {
if ev.Kind == model.StreamKindToolExecutionStart || ev.Kind == model.StreamKindToolExecutionEnd {
saw = true
break
}
}
if !saw {
t.Fatalf("expected a tool_execution_update event on the streaming path, got %+v", evs)
t.Fatalf("expected a tool_execution start/end event on the streaming path, got %+v", evs)
}
}
+9 -2
View File
@@ -283,8 +283,15 @@ func TestRun_StreamDeltas(t *testing.T) {
var mu sync.Mutex
var deltas []string
l := New(Options{
Agent: a,
OnStream: func(ev model.StreamEvent) { mu.Lock(); deltas = append(deltas, ev.Delta); mu.Unlock() },
Agent: a,
OnStream: func(ev model.StreamEvent) {
if ev.Kind != model.StreamKindTextDelta {
return
}
mu.Lock()
deltas = append(deltas, ev.Delta)
mu.Unlock()
},
MaxRounds: 8,
})
+26 -11
View File
@@ -23,6 +23,10 @@ import (
// goroutine 上,因此提取绝不延迟回复),错误记录日志
// 而非致命。
//
// 回合边界发出 turn.start / turn.end 流式帧(design/04 §1.1):
// start 在任何工作之前,end 经 defer 覆盖 Run 的每个出口
// (最终答案 / 安全网终止 / 预算耗尽 / 步骤错误)。
//
// 当 turn 带着 Result 落定时,OnTurnEnd 钩子对该
// turn 调用一次(绝不按步调用),其情绪标签记录在 Result.Mood 上,
// 使循环外围的分层能推进人格状态并标记持久化的回复。
@@ -46,6 +50,9 @@ func (l *Loop) Run(ctx context.Context, sessionID string, messages []model.Messa
ctx = withCapState(ctx, &capStateAccum{state: capacity.StateQuiet})
ctx = withLoopSession(ctx, sessionID)
l.emitTurnBoundary(ctx, model.StreamKindTurnStart)
defer l.emitTurnBoundary(ctx, model.StreamKindTurnEnd)
// 恢复被中断的 run(design §1 durable execution):该会话有
// 可恢复的 checkpoint 时,重建其上下文并从下一步继续,
// 而不是从全新对话重启。不同的
@@ -53,7 +60,6 @@ func (l *Loop) Run(ctx context.Context, sessionID string, messages []model.Messa
// 全新开始(LoadLatest 未命中,或 stop reason 不可恢复)。
msgs, startStep, _ := l.loadCheckpoint(ctx, sessionID, messages)
msgs = l.resolveImages(ctx, msgs)
toolState := &toolLoopState{prevFP: make(map[string]bool)}
res := &Result{}
// 分支 Run(子代理)把其 span 并入父会话的实时
// 图;只有所属的 Run 开始与结束轮次。没有它,子代理
@@ -62,6 +68,14 @@ func (l *Loop) Run(ctx context.Context, sessionID string, messages []model.Messa
l.flow.StartRound(sessionID, lastUserText(msgs))
defer func() { l.flow.EndRound(sessionID, res.Text) }()
}
return l.driveTurn(ctx, sessionID, msgs, startStep, res)
}
// driveTurn 运行至多 maxRounds 个 step 的主循环:每步组装上下文、
// 调用模型、执行被请求的工具并回灌。它在任何一种落定路径上返回
// 已 settle 的 Result;step 或工具批次的错误以 %w 包装返回。
func (l *Loop) driveTurn(ctx context.Context, sessionID string, msgs []model.Message, startStep int, res *Result) (*Result, error) {
toolState := &toolLoopState{prevFP: make(map[string]bool)}
for step := startStep; step <= l.maxRounds; step++ {
if err := ctx.Err(); err != nil {
return nil, fmt.Errorf("loop: run: step %d: %w", step, err)
@@ -88,10 +102,7 @@ func (l *Loop) Run(ctx context.Context, sessionID string, messages []model.Messa
if outcome.final {
res.Text = outcome.text
res.StopReason = outcome.stopReason
l.saveCheckpoint(ctx, sessionID, step, msgs, outcome.stopReason)
l.extractMemoryAsync(ctx, msgs, step)
l.settleTurn(ctx, sessionID, msgs, res)
return res, nil
return l.settleAndReturn(ctx, sessionID, msgs, step, outcome.stopReason, res)
}
batch, err := l.runToolCalls(rc, msgs, outcome.toolCalls, toolState)
@@ -107,18 +118,22 @@ func (l *Loop) Run(ctx context.Context, sessionID string, messages []model.Messa
if batch.terminate {
res.Text = criticalToolLoopTerminalReason
res.StopReason = StopReasonToolLoopTerminated
l.saveCheckpoint(ctx, sessionID, step, msgs, StopReasonToolLoopTerminated)
l.extractMemoryAsync(ctx, msgs, step)
l.settleTurn(ctx, sessionID, msgs, res)
return res, nil
return l.settleAndReturn(ctx, sessionID, msgs, step, StopReasonToolLoopTerminated, res)
}
}
// 预算耗尽而无最终答案。
res.Steps = l.maxRounds
res.StopReason = StopReasonMaxRounds
l.saveCheckpoint(ctx, sessionID, l.maxRounds, msgs, StopReasonMaxRounds)
l.extractMemoryAsync(ctx, msgs, l.maxRounds)
return l.settleAndReturn(ctx, sessionID, msgs, l.maxRounds, StopReasonMaxRounds, res)
}
// settleAndReturn 落定一个已得出 Result 的回合:以该路径的
// stop reason 持久化终态 checkpoint、触发后台记忆提取、
// 上报 turn 钩子(情绪标签写回 Result.Mood),然后把结果交还调用方。
func (l *Loop) settleAndReturn(ctx context.Context, sessionID string, msgs []model.Message, step int, stopReason string, res *Result) (*Result, error) {
l.saveCheckpoint(ctx, sessionID, step, msgs, stopReason)
l.extractMemoryAsync(ctx, msgs, step)
l.settleTurn(ctx, sessionID, msgs, res)
return res, nil
}
+37
View File
@@ -2,7 +2,10 @@ package loop
import (
"context"
"crypto/rand"
"encoding/hex"
"fmt"
"time"
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/agent"
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/model"
@@ -36,6 +39,11 @@ func (l *Loop) streamStep(ctx context.Context, messages []model.Message) (*stepO
}
}
// 一次模型流(一次 ChatStream 调用)取一个 streamId:该流的
// 每一帧共享它,而多步工具轮的后续模型调用取新的 id
// (design/04 §1.1)。它与 TurnID 分工:turnId 标记整个回合,
// streamId 标记回合内的一条流式消息。
streamID := newStreamID()
resp, err := adapter.ChatStream(ctx, req, func(ev model.StreamEvent) {
// 在循环边界打上 session/turn 身份:适配器运行在
// 循环之下,无法知道该流属于哪个会话,
@@ -43,6 +51,7 @@ func (l *Loop) streamStep(ctx context.Context, messages []model.Message) (*stepO
// 前端便能把增量路由到活跃会话。
ev.SessionID = loopSession(ctx)
ev.TurnID = loopTurn(ctx)
ev.StreamID = streamID
// 为内部自主流程(backthink)抑制 UI 广播,
// 其中间推理不得泄露进聊天流;当该 turn 的角色
// 关闭思考帧时(showToUser: false)也抑制。两处门
@@ -95,3 +104,31 @@ func isThinkingKind(kind model.StreamKind) bool {
return false
}
}
// newStreamID 返回新的随机十六进制标识,用于把一串流帧关联到
// 同一次模型流(design/04 §1.1 streamId)。它消耗 crypto/rand 源,
// 源不可用时回退到纳秒时间戳,因此总能产出 id(空的 streamId 会
// 静默破坏客户端关联)。
func newStreamID() string {
var b [8]byte
if _, err := rand.Read(b[:]); err == nil {
return "stream-" + hex.EncodeToString(b[:])
}
return "stream-" + time.Now().UTC().Format("20060102150405.000000000")
}
// emitTurnBoundary 在循环的回合边界发出 turn.start / turn.end 帧
// (design/04 §1.1)。它复用流式路径的广播门:未配置 OnStream 或
// 被静音的内部流程(backthink / 子代理)不发出。两帧不带载荷 ——
// 帧壳上的 session/turn 身份就是边界信号,回合的终态信息仍由
// hub 在 Run 返回后发出的 stream_end 携带。
func (l *Loop) emitTurnBoundary(ctx context.Context, kind model.StreamKind) {
if l.onStream == nil || isMutedBroadcast(ctx) {
return
}
l.onStream(model.StreamEvent{
Kind: kind,
SessionID: loopSession(ctx),
TurnID: loopTurn(ctx),
})
}
+275
View File
@@ -0,0 +1,275 @@
package loop
import (
"context"
"errors"
"strings"
"sync"
"testing"
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/model"
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/trace"
)
// captureFlow 返回一个 FlowPublisher 与读取其已发布快照的 reader,
// 使测试能断言工具节点的 status 如何随 phase 演进(design/19 §1)。
func captureFlow() (*FlowPublisher, func() []*trace.FlowGraph) {
var mu sync.Mutex
var seen []*trace.FlowGraph
flow := NewFlowPublisher(func(subject string, payload any) {
if subject != graphUpdateSubject {
return
}
if g, ok := payload.(*trace.FlowGraph); ok {
mu.Lock()
seen = append(seen, g)
mu.Unlock()
}
}, func(string) bool { return true })
return flow, func() []*trace.FlowGraph {
mu.Lock()
defer mu.Unlock()
return append([]*trace.FlowGraph(nil), seen...)
}
}
// toolExecEvents 过滤出 tool_execution 三事件,保持发出顺序。
func toolExecEvents(evs []model.StreamEvent) []model.StreamEvent {
var out []model.StreamEvent
for _, ev := range evs {
switch ev.Kind {
case model.StreamKindToolExecutionStart, model.StreamKindToolExecutionUpdate, model.StreamKindToolExecutionEnd:
out = append(out, ev)
}
}
return out
}
// toolNodeStatus 返回图中所有 tool 节点的当前 status。
func toolNodeStatus(g *trace.FlowGraph) []string {
var out []string
for _, n := range g.Nodes {
if n.Kind == "tool" {
out = append(out, n.Status)
}
}
return out
}
// runToolTurn 用「模型请求 datetime → 工具执行 → 模型收尾」的两步流
// 驱动一个带流式回调与 flow 图的 Run,返回收集到的事件与图快照。
func runToolTurn(t *testing.T, toolResult any, toolErr error) ([]model.StreamEvent, []*trace.FlowGraph) {
t.Helper()
ad := &mockAdapter{stream: []mgResp{
{resp: nativeDatetimeCall()},
{resp: model.Response{Model: "m", Content: "done"}},
}}
td := dispatcherWith(t, &mockTool{name: "datetime", result: toolResult, err: toolErr})
a := newTestAgent(t, ad, td)
onStream, streams := collectStream()
flow, graphs := captureFlow()
l := New(Options{Agent: a, OnStream: onStream, MaxRounds: 8, Flow: flow, Tracer: trace.New(nil)})
if _, err := l.Run(context.Background(), "web", []model.Message{model.TextMessage(model.RoleUser, "time?")}); err != nil {
t.Fatalf("Run: %v", err)
}
return streams(), graphs()
}
// TestRun_ToolExecutionThreeEvents 锁住裁决 01.4.1:一次工具执行发出
// start 与 end 两帧(顺序固定),phase 是 running/done 的结构化字段,
// 绝不再塞进 Delta;工具名走 Tool 字段,ID 复用 provider 的
// tool call id,使执行帧可与 tool_call.* 帧关联。
func TestRun_ToolExecutionThreeEvents(t *testing.T) {
t.Parallel()
evs, _ := runToolTurn(t, map[string]any{"now": "2026-09-08 22:34:00"}, nil)
ex := toolExecEvents(evs)
if len(ex) != 2 {
t.Fatalf("tool_execution events = %d, want 2 (start+end): %+v", len(ex), ex)
}
start, end := ex[0], ex[1]
if start.Kind != model.StreamKindToolExecutionStart || start.Phase != model.ToolPhaseRunning {
t.Fatalf("start event = %+v, want kind start phase running", start)
}
if end.Kind != model.StreamKindToolExecutionEnd || end.Phase != model.ToolPhaseDone {
t.Fatalf("end event = %+v, want kind end phase done", end)
}
for _, ev := range ex {
if ev.Tool != "datetime" {
t.Fatalf("event tool = %q, want datetime (%+v)", ev.Tool, ev)
}
if ev.ID != "call_native_1" {
t.Fatalf("event id = %q, want the provider tool call id (%+v)", ev.ID, ev)
}
if ev.Delta != "" {
t.Fatalf("event delta = %q, want empty (phase must not ride in Delta)", ev.Delta)
}
if ev.StreamID != "" {
t.Fatalf("event streamId = %q, want empty (tool execution is not a model stream)", ev.StreamID)
}
}
}
// TestRun_ToolExecutionErrorPhase 覆盖失败路径:end 帧的 phase 为
// error,且 flow 图中该 tool 节点最终也是 error —— 同一 phase 驱动
// 两处(裁决 03.4.4)。
func TestRun_ToolExecutionErrorPhase(t *testing.T) {
t.Parallel()
evs, graphs := runToolTurn(t, nil, errors.New("boom"))
ex := toolExecEvents(evs)
if len(ex) != 2 {
t.Fatalf("tool_execution events = %d, want 2: %+v", len(ex), ex)
}
if ex[1].Kind != model.StreamKindToolExecutionEnd || ex[1].Phase != model.ToolPhaseError {
t.Fatalf("end event = %+v, want kind end phase error", ex[1])
}
last := graphs[len(graphs)-1]
statuses := toolNodeStatus(last)
if len(statuses) == 0 || statuses[0] != trace.FlowStatusError {
t.Fatalf("final tool node status = %v, want [error] (phase-driven)", statuses)
}
}
// TestRun_ToolPhaseDrivesFlowNodeStatus 锁住裁决 03.4.4:工具执行期间
// flow 图出现 running 的 tool 节点,执行结束后翻转为 done —— 节点状态
// 由细粒度 phase 驱动,取值与事件 phase 完全一致。
func TestRun_ToolPhaseDrivesFlowNodeStatus(t *testing.T) {
t.Parallel()
evs, graphs := runToolTurn(t, map[string]any{"now": "x"}, nil)
sawRunning, sawDone := false, false
for _, g := range graphs {
for _, s := range toolNodeStatus(g) {
switch s {
case trace.FlowStatusRunning:
sawRunning = true
case trace.FlowStatusDone:
sawDone = true
}
}
}
if !sawRunning || !sawDone {
t.Fatalf("tool node statuses over the round: running=%v done=%v, want both (phase-driven)", sawRunning, sawDone)
}
ex := toolExecEvents(evs)
if len(ex) != 2 || ex[0].Phase != model.ToolPhaseRunning || ex[1].Phase != model.ToolPhaseDone {
t.Fatalf("phases = %+v, want [running done] matching the node statuses", ex)
}
}
// TestRun_StreamIDPerModelStream 锁住裁决 01.4.2:同一次模型流的多帧
// 共享同一 streamId;多步工具轮的第二次模型调用取新的 streamId;
// turn 与 tool_execution 帧不属于任何模型流,不带 streamId。
func TestRun_StreamIDPerModelStream(t *testing.T) {
t.Parallel()
ad := &mockAdapter{stream: []mgResp{
{resp: nativeDatetimeCall()},
{resp: model.Response{Model: "m", Content: "done"}},
}, chunks: [][]string{{"Hel"}, {"lo"}}}
td := dispatcherWith(t, &mockTool{name: "datetime", result: "now"})
a := newTestAgent(t, ad, td)
onStream, streams := collectStream()
l := New(Options{Agent: a, OnStream: onStream, MaxRounds: 8})
if _, err := l.Run(context.Background(), "web", []model.Message{model.TextMessage(model.RoleUser, "time?")}); err != nil {
t.Fatalf("Run: %v", err)
}
var firstStream, secondStream string
for _, ev := range streams() {
if ev.Kind != model.StreamKindTextDelta {
continue
}
if !strings.HasPrefix(ev.StreamID, "stream-") {
t.Fatalf("delta streamId = %q, want a stream-* id", ev.StreamID)
}
switch ev.Delta {
case "Hel":
firstStream = ev.StreamID
case "lo":
secondStream = ev.StreamID
}
}
if firstStream == "" || secondStream == "" {
t.Fatalf("missing deltas for the two model streams (first=%q second=%q)", firstStream, secondStream)
}
if firstStream == secondStream {
t.Fatalf("the two model streams share streamId %q; each ChatStream call must mint a fresh one", firstStream)
}
for _, ev := range streams() {
switch ev.Kind {
case model.StreamKindTurnStart, model.StreamKindTurnEnd,
model.StreamKindToolExecutionStart, model.StreamKindToolExecutionEnd:
if ev.StreamID != "" {
t.Fatalf("%s carries streamId %q; only model-stream frames do", ev.Kind, ev.StreamID)
}
}
}
}
// TestRun_TurnBoundaryFrames 锁住裁决 01.4.3:回合以 turn.start 开始、
// turn.end 结束(无论正常落定还是预算耗尽),两帧携带会话与 turn
// 作用域、无载荷。
func TestRun_TurnBoundaryFrames(t *testing.T) {
t.Parallel()
ad := &mockAdapter{stream: []mgResp{{resp: model.Response{Content: "ok"}}}, chunks: [][]string{{"ok"}}}
a := newTestAgent(t, ad, nil)
onStream, streams := collectStream()
l := New(Options{Agent: a, OnStream: onStream, MaxRounds: 8})
ctx := WithTurnID(context.Background(), "turn-b")
if _, err := l.Run(ctx, "web", []model.Message{model.TextMessage(model.RoleUser, "hi")}); err != nil {
t.Fatalf("Run: %v", err)
}
evs := streams()
if len(evs) < 3 {
t.Fatalf("events = %+v, want at least turn.start + one delta + turn.end", evs)
}
if evs[0].Kind != model.StreamKindTurnStart || evs[len(evs)-1].Kind != model.StreamKindTurnEnd {
t.Fatalf("boundary kinds = %q…%q, want turn_start…turn_end", evs[0].Kind, evs[len(evs)-1].Kind)
}
for _, boundary := range []model.StreamEvent{evs[0], evs[len(evs)-1]} {
if boundary.SessionID != "web" || boundary.TurnID != "turn-b" {
t.Fatalf("boundary scope = session %q turn %q, want web turn-b", boundary.SessionID, boundary.TurnID)
}
if boundary.Delta != "" || boundary.Phase != "" || boundary.Tool != "" {
t.Fatalf("boundary carries payload: %+v", boundary)
}
}
}
// TestRun_TurnEndEmittedOnStepError 覆盖错误出口:模型步失败时 Run
// 返回错误,但回合已经以 turn.start 开过,因此 turn.end 必须照样
// 发出,客户端不会留下一个永远开着的回合。
func TestRun_TurnEndEmittedOnStepError(t *testing.T) {
t.Parallel()
ad := &mockAdapter{stream: []mgResp{{err: errors.New("model down")}}}
a := newTestAgent(t, ad, nil)
onStream, streams := collectStream()
l := New(Options{Agent: a, OnStream: onStream, MaxRounds: 8})
if _, err := l.Run(context.Background(), "web", []model.Message{model.TextMessage(model.RoleUser, "hi")}); err == nil {
t.Fatal("Run: expected the step error to surface")
}
evs := streams()
if len(evs) != 2 || evs[0].Kind != model.StreamKindTurnStart || evs[1].Kind != model.StreamKindTurnEnd {
t.Fatalf("events = %+v, want exactly turn.start + turn.end around the failed run", evs)
}
}
// TestRun_TurnBoundaryMutedForInternalRuns 内部自主流程(backthink /
// 子代理)经 WithMutedBroadcast 驱动同一个循环:回合边界帧与流帧
// 一样被静音,中间过程绝不泄露进面向用户的聊天流。
func TestRun_TurnBoundaryMutedForInternalRuns(t *testing.T) {
t.Parallel()
ad := &mockAdapter{stream: []mgResp{{resp: model.Response{Content: "ok"}}}, chunks: [][]string{{"o"}}}
a := newTestAgent(t, ad, nil)
onStream, streams := collectStream()
l := New(Options{Agent: a, OnStream: onStream, MaxRounds: 8})
ctx := WithMutedBroadcast(context.Background())
if _, err := l.Run(ctx, "sess-backthink", []model.Message{model.TextMessage(model.RoleUser, "reflect")}); err != nil {
t.Fatalf("Run: %v", err)
}
if evs := streams(); len(evs) != 0 {
t.Fatalf("muted run emitted %d events, want none: %+v", len(evs), evs)
}
}
+12 -8
View File
@@ -29,7 +29,8 @@ func collectStream() (func(model.StreamEvent), func() []model.StreamEvent) {
// TestRun_StreamCarriesSessionAndTurn 钉住 A1 契约:循环给每个
// 流式事件打上它被驱动的会话,以及经 WithTurnID 挂载的 turn id,
// 使 WS hub 能把 message.delta/thinking.delta/tool_call.*
// 路由到正确的会话与 turn(design/16 §4.2)。
// 路由到正确的会话与 turn(design/16 §4.2)。回合边界帧
// (turn.start/end,v0.50)同样带作用域,因此计入总数。
func TestRun_StreamCarriesSessionAndTurn(t *testing.T) {
t.Parallel()
ad := &mockAdapter{
@@ -45,16 +46,19 @@ func TestRun_StreamCarriesSessionAndTurn(t *testing.T) {
t.Fatalf("Run: %v", err)
}
events := streams()
if len(events) != 2 {
t.Fatalf("stream events = %d, want 2", len(events))
if len(events) != 4 {
t.Fatalf("stream events = %d, want 4 (turn.start + 2 deltas + turn.end)", len(events))
}
for i, ev := range events {
if ev.SessionID != "web" || ev.TurnID != "turn-xyz" {
t.Fatalf("event[%d] scope = session %q turn %q, want session web turn turn-xyz", i, ev.SessionID, ev.TurnID)
}
}
if events[0].Delta != "Hel" || events[1].Delta != "lo" {
t.Fatalf("deltas = [%q %q], want [Hel lo]", events[0].Delta, events[1].Delta)
if events[0].Kind != model.StreamKindTurnStart || events[3].Kind != model.StreamKindTurnEnd {
t.Fatalf("boundary kinds = %q…%q, want turn_start…turn_end", events[0].Kind, events[3].Kind)
}
if events[1].Delta != "Hel" || events[2].Delta != "lo" {
t.Fatalf("deltas = [%q %q], want [Hel lo]", events[1].Delta, events[2].Delta)
}
}
@@ -86,7 +90,7 @@ func TestRun_StreamWithoutTurnIDLeavesItEmpty(t *testing.T) {
}
// TestRun_ToolExecutionCarriesScope 覆盖循环的另一个流式发出点:
// tool_execution_update 进度帧也必须带作用域,否则它就会成为
// tool_execution.start/end 进度帧也必须带作用域,否则它就会成为
// 多会话客户端唯一无法路由的帧。
func TestRun_ToolExecutionCarriesScope(t *testing.T) {
t.Parallel()
@@ -105,7 +109,7 @@ func TestRun_ToolExecutionCarriesScope(t *testing.T) {
}
found := 0
for _, ev := range streams() {
if ev.Kind != model.StreamKindToolExecution {
if ev.Kind != model.StreamKindToolExecutionStart && ev.Kind != model.StreamKindToolExecutionEnd {
continue
}
found++
@@ -114,6 +118,6 @@ func TestRun_ToolExecutionCarriesScope(t *testing.T) {
}
}
if found == 0 {
t.Fatal("no tool_execution_update event was emitted")
t.Fatal("no tool_execution event was emitted")
}
}
+4 -4
View File
@@ -85,8 +85,8 @@ func TestRun_ThinkingShowToUserFalseWithholdsThinkingFrames(t *testing.T) {
t.Fatalf("event kind %q emitted %d time(s), want 1 (body must be unaffected)", k, kinds[k])
}
}
if len(events) != 3 {
t.Fatalf("events = %d, want only the 3 text frames", len(events))
if len(events) != 5 {
t.Fatalf("events = %d, want the 3 text frames plus the 2 turn boundaries", len(events))
}
}
@@ -100,8 +100,8 @@ func TestRun_ThinkingShowToUserDefaultBroadcastsThinking(t *testing.T) {
if kinds[model.StreamKindThinkingDelta] != 1 || kinds[model.StreamKindThinkingStart] != 1 || kinds[model.StreamKindThinkingEnd] != 1 {
t.Fatalf("thinking frames = %v, want start/delta/end each once with the option unset", kinds)
}
if len(events) != 6 {
t.Fatalf("events = %d, want all 6 frames", len(events))
if len(events) != 8 {
t.Fatalf("events = %d, want all 8 frames (6 segment frames + 2 turn boundaries)", len(events))
}
}
+94 -69
View File
@@ -114,42 +114,8 @@ func (l *Loop) runToolCalls(ctx context.Context, messages []model.Message, calls
res.terminate = true
}
// 安全门(design/11 §3):群聊/陌生人会话只能调用
// 其可编辑标记允许该环境的工具(默认关闭)。不被允许的
// 调用绝不执行;它以 not_allowed 软结果回灌
// 并记录到审计 sink。它是该批的最终权威 ——
// 后续的审计钩子可以收紧,但绝不重新放行被拒的工具。
if d != nil {
l.applyToolSecurityGate(ctx, d, calls, actions)
}
// BatchAdmit 钩子(审计 / 政治审查):拦截整批。
if l.hooks.BatchAdmit != nil {
if ok, reason := l.hooks.BatchAdmit(ctx, calls); !ok {
if reason == "" {
reason = "tool batch blocked by policy"
}
for i := range calls {
actions[i] = toolCallAction{run: false, reason: reason}
}
}
}
// CallAdmit 钩子:对我们打算运行的调用逐个否决。
if l.hooks.CallAdmit != nil {
for i := range calls {
if !actions[i].run {
continue
}
if veto := l.hooks.CallAdmit(ctx, calls[i]); veto != nil && veto.Block {
reason := veto.Reason
if reason == "" {
reason = "tool call blocked by policy"
}
actions[i] = toolCallAction{run: false, reason: reason}
}
}
}
// 执行前的三道门(安全门 + 批次/逐调用钩子)见 applyToolAdmissions。
l.applyToolAdmissions(ctx, d, calls, actions)
// 决定串行还是并行:显式串行模式,或任一调用的工具
// 被声明为串行(单个串行标记即让整组
@@ -187,26 +153,8 @@ func (l *Loop) runToolCalls(ctx context.Context, messages []model.Message, calls
res.executed += len(idxs)
}
// AfterCall 钩子(仅对实际执行过的调用)。
if l.hooks.AfterCall != nil {
for i := range calls {
if outcomes[i].Skipped {
continue
}
if after := l.hooks.AfterCall(ctx, calls[i], outcomes[i]); after != nil {
outcomes[i] = *after
}
}
}
// AfterOutcome 钩子(对每个最终结果运行,无论是否跳过)。
if l.hooks.AfterOutcome != nil {
for i := range calls {
if after := l.hooks.AfterOutcome(ctx, calls[i], outcomes[i]); after != nil {
outcomes[i] = *after
}
}
}
// 结果落定后的两类观察/改写钩子见 applyToolOutcomeHooks。
l.applyToolOutcomeHooks(ctx, calls, outcomes)
// 钩子可以把每个调用都标记为 terminate 来要求终止本轮
// (OpenClaw 的 shouldTerminateToolBatch)。
@@ -277,48 +225,125 @@ func classifyToolBatch(st *toolLoopState, calls []agent.ToolCall) ([]toolCallAct
return actions, terminate
}
// applyToolAdmissions 对一批调用应用执行前的三道门,把不放行的
// 调用改写为携带原因的跳过决策:① 安全门(design/11 §3)——
// 群聊/陌生人会话只能调用其标记允许该环境的工具(默认关闭),
// 不被允许的调用绝不执行,以 not_allowed 软结果回灌并记录审计;
// 它是该批的最终权威,后续钩子可以收紧但绝不重新放行。②
// BatchAdmit 批次钩子(审计 / 政治审查)拦截整批。③ CallAdmit
// 逐调用否决。dispatcher 缺席时安全门跳过(runToolCalls 已保证
// 此时整批以错误回灌)。
func (l *Loop) applyToolAdmissions(ctx context.Context, d *tool.Dispatcher, calls []agent.ToolCall, actions []toolCallAction) {
if d != nil {
l.applyToolSecurityGate(ctx, d, calls, actions)
}
if l.hooks.BatchAdmit != nil {
if ok, reason := l.hooks.BatchAdmit(ctx, calls); !ok {
if reason == "" {
reason = "tool batch blocked by policy"
}
for i := range calls {
actions[i] = toolCallAction{run: false, reason: reason}
}
}
}
if l.hooks.CallAdmit != nil {
for i := range calls {
if !actions[i].run {
continue
}
if veto := l.hooks.CallAdmit(ctx, calls[i]); veto != nil && veto.Block {
reason := veto.Reason
if reason == "" {
reason = "tool call blocked by policy"
}
actions[i] = toolCallAction{run: false, reason: reason}
}
}
}
}
// applyToolOutcomeHooks 运行结果落定后的两类观察/改写钩子,
// 就地改写 outcomes:AfterCall 仅对实际执行过的调用(在
// AfterOutcome 之前),AfterOutcome 对每个最终结果(无论是否
// 跳过),都在结果回灌给模型之前运行。
func (l *Loop) applyToolOutcomeHooks(ctx context.Context, calls []agent.ToolCall, outcomes []ToolOutcome) {
if l.hooks.AfterCall != nil {
for i := range calls {
if outcomes[i].Skipped {
continue
}
if after := l.hooks.AfterCall(ctx, calls[i], outcomes[i]); after != nil {
outcomes[i] = *after
}
}
}
if l.hooks.AfterOutcome != nil {
for i := range calls {
if after := l.hooks.AfterOutcome(ctx, calls[i], outcomes[i]); after != nil {
outcomes[i] = *after
}
}
}
}
// invokeOne 在其每工具截止时间下执行单个工具调用,发出一个
// tool span、流式输出 tool_execution_update 进度并归一化
// 结果。
// tool span、以 tool_execution.start/end 事件流式输出执行进度
// 并归一化结果。执行全程由单一 phase 决策驱动:start 帧带
// running,Invoke 返回后按成败取 done/error,同一取值既决定
// end 帧的 phase,也决定 span 终态(design/19 的 tool 节点
// running/done/error 状态即由此驱动)。
func (l *Loop) invokeOne(ctx context.Context, d *tool.Dispatcher, tc agent.ToolCall) ToolOutcome {
if d == nil {
return ToolOutcome{Err: fmt.Errorf("no tool dispatcher configured for %q", tc.Name), ErrKind: tool.KindExecution}
}
l.emitToolExecution(ctx, tc.Name, "start")
_, toolSpan := l.span(ctx, "tool."+tc.Name, trace.SpanKindTool)
l.emitToolExecution(ctx, tc, model.StreamKindToolExecutionStart, model.ToolPhaseRunning)
result, err := d.Invoke(ctx, tc.Name, tc.Args)
phase := "end"
phase := model.ToolPhaseDone
if err != nil {
phase = "end:error"
phase = model.ToolPhaseError
}
l.emitToolExecution(ctx, tc.Name, phase)
l.emitToolExecution(ctx, tc, model.StreamKindToolExecutionEnd, phase)
out := ToolOutcome{Result: result, Err: err, ErrKind: errorKind(err)}
l.spanProp(toolSpan, "phase", phase)
l.recordToolDone(toolSpan, tc, out)
if err != nil {
l.spanProp(toolSpan, "error", err.Error())
l.spanProp(toolSpan, "error_kind", out.ErrKind)
l.finishSpan(toolSpan, trace.SpanStatusError)
} else {
l.spanProp(toolSpan, "result", result)
l.finishSpan(toolSpan, trace.SpanStatusOK)
}
l.finishSpan(toolSpan, spanStatusForToolPhase(phase))
return out
}
// emitToolExecution 在循环处于流式路径(配置了 OnStream)时把
// tool_execution_update 进度事件转发给前端;否则为空操作
// (被静音的内部流程也会跳过它)。phase 是一个短标签,
// 如 "start" 或 "end:error"。
func (l *Loop) emitToolExecution(ctx context.Context, name, phase string) {
// tool_execution 三事件(start / update / end)转发给前端;否则为
// 空操作(被静音的内部流程也会跳过它)。phase 是结构化字段
// (running/done/error),绝不塞进 Delta;工具名走 Tool 字段,
// ID 复用 provider 的 tool call id,使执行帧能与其 tool_call.* 帧
// 关联。
func (l *Loop) emitToolExecution(ctx context.Context, tc agent.ToolCall, kind model.StreamKind, phase string) {
if l.onStream == nil || isMutedBroadcast(ctx) {
return
}
ev := model.StreamEvent{Kind: model.StreamKindToolExecution, Delta: phase + ":" + name}
ev := model.StreamEvent{Kind: kind, Phase: phase, Tool: tc.Name, ID: tc.ID}
ev.SessionID = loopSession(ctx)
ev.TurnID = loopTurn(ctx)
l.onStream(ev)
}
// spanStatusForToolPhase 把工具执行的终态 phase 映射为 span 终态,
// 使 trace/流程图与 WS 帧从同一个 phase 取值得出一致结论。
// 非终态 phase(running)不会走到这里 —— span 只在执行结束时收尾。
func spanStatusForToolPhase(phase string) trace.SpanStatus {
if phase == model.ToolPhaseError {
return trace.SpanStatusError
}
return trace.SpanStatusOK
}
// invokeParallel 并发运行 idxs 处的调用,上限为循环的
// 并行度,并把每个结果写回 out 的对应调用索引处,使调用方
// 能按原始顺序渲染结果。信号量限制并发,在途