diff --git a/frontend/web/src/admin/modules/flow/NodeDetail.tsx b/frontend/web/src/admin/modules/flow/NodeDetail.tsx index 5bcc42b..65d65b9 100644 --- a/frontend/web/src/admin/modules/flow/NodeDetail.tsx +++ b/frontend/web/src/admin/modules/flow/NodeDetail.tsx @@ -5,11 +5,11 @@ import { Notice } from "../../components"; import { FLOW_TONE, statusOf } from "./tone"; /** NodeDetail shows one selected node: its status, its lane (when it belongs to - * a sub-agent branch), the whitelisted span props and the span's accumulated - * events (design/19 §2 节点实时日志). The panel is driven by the live graph, so - * a newer `graph_update` frame replaces the props/log arrays in place and the - * list keeps growing while the round runs — no local buffering, no invented - * entries. */ + * a sub-agent branch), the tool argument summary (design/19 §1.1), the + * whitelisted span props and the span's accumulated events (design/19 §2 + * 节点实时日志). The panel is driven by the live graph, so a newer + * `graph_update` frame replaces the props/log arrays in place and the list keeps + * growing while the round runs — no local buffering, no invented entries. */ export function NodeDetail({ node, laneLabel, @@ -46,6 +46,8 @@ export function NodeDetail({ {node.id} + {argsSection(props)} + {hasProps ? (

参数 / 结果

@@ -62,6 +64,51 @@ export function NodeDetail({ ); } +/** argsSection 渲染 tool 节点的参数摘要（design/19 §1.1，裁决 11.4.3）：模型传过的 + * 参数名全部列出；放行表内的键显示截断后的值，其余键显式标注为「值未展示」—— + * 一个从未离开后端的值，绝不能在界面上看起来像空值。没有 `args_keys` 的节点 + * （非工具节点）不渲染任何东西。 */ +function argsSection(props: Record): React.ReactNode { + const keys = toStringList(props.args_keys); + if (keys.length === 0) return null; + const summary = toRecord(props.args_summary); + return ( +
+

参数摘要

+
+ {keys.map((key) => { + const value = summary[key]; + return ( +
+ {key} + {typeof value === "string" && value.length > 0 ? ( + {value} + ) : ( + 值未展示（未放行或密钥类） + )} +
+ ); + })} +
+

+ 仅放行键显示截断后的值；密钥类键永不显示值（design/19 §1.1）。 +

+
+ ); +} + +/** toStringList 读取字符串数组属性，忽略任何畸形取值。 */ +function toStringList(value: unknown): string[] { + if (!Array.isArray(value)) return []; + return value.filter((item): item is string => typeof item === "string"); +} + +/** toRecord 读取字符串键的对象属性，忽略任何畸形取值。 */ +function toRecord(value: unknown): Record { + if (value === null || typeof value !== "object" || Array.isArray(value)) return {}; + return value as Record; +} + /** LogList renders the node's accumulated span events, newest last, and is * explicit about why an empty list is empty instead of pretending to have * content. */ diff --git a/internal/runtime/agent/step.go b/internal/runtime/agent/step.go index 5e32955..2cc1b27 100644 --- a/internal/runtime/agent/step.go +++ b/internal/runtime/agent/step.go @@ -5,8 +5,10 @@ import ( "fmt" "time" + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/guardrail" "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" ) // Stop reasons carried by StepResult.StopReason, consumed by the loop (M2) to @@ -53,6 +55,15 @@ type StepResult struct { ModelStopReason string } +// 护栏 span 名与本包记录事件时使用的类型：与循环侧（loop/spanlog.go） +// 同名，使面板无论走流式路径还是 agent 路径都看到同一类 guardrail 节点 +// 与同一条日志事件（裁决 11.4.2，design/04 §9「逐层记到 guardrail」）。 +const ( + guardrailSpanInput = "guardrail.input" + guardrailSpanOutput = "guardrail.output" + guardrailEventDone = "guardrail.done" +) + // RunStep executes one step of the agent loop. It assembles a model.Request from // the agent's instructions prepended as a leading system message plus the // conversation messages, runs input guardrails, calls the model adapter, runs @@ -64,11 +75,20 @@ type StepResult struct { // yields Final=false with the decoded ToolCalls; a guardrail denial yields // Final=true carrying the block reason. A nil adapter or a failed guardrail or // model call returns a wrapped error. +// +// 两侧检查都包在自己的 guardrail span 里（裁决 11.4.2）：本包不持有 Tracer， +// 因此用 trace.ChildSpan 在当前 span（循环的 step span）之下开一个子 span， +// 未开追踪时它是空操作。**本路径保持串行内联**：输入侧在模型调用之前拦下， +// 输出侧紧接模型返回裁决（裁决 13.4.1 的「输出侧异步」由流式主路径 +// loop.streamStep 承担——agent.RunStep 只服务静音/非流式内部流程，不落在 +// 主会话关键路径上，且它的裁决是 StepResult 的一部分）。 func (a *Agent) RunStep(ctx context.Context, sessionID string, messages []model.Message) (*StepResult, error) { req := a.buildRequest(messages) if a.guards != nil { - verdict, err := a.guards.CheckInput(ctx, &req) + gctx, gspan := trace.ChildSpan(ctx, guardrailSpanInput, trace.SpanKindGuardrail) + verdict, err := a.guards.CheckInput(gctx, &req) + finishGuardrailSpan(gspan, guardrailSpanInput, verdict, err) if err != nil { return nil, fmt.Errorf("agent: check input: %w", err) } @@ -96,7 +116,9 @@ func (a *Agent) RunStep(ctx context.Context, sessionID string, messages []model. } if a.guards != nil { - verdict, err := a.guards.CheckOutput(ctx, &resp) + gctx, gspan := trace.ChildSpan(ctx, guardrailSpanOutput, trace.SpanKindGuardrail) + verdict, err := a.guards.CheckOutput(gctx, &resp) + finishGuardrailSpan(gspan, guardrailSpanOutput, verdict, err) if err != nil { return nil, fmt.Errorf("agent: check output: %w", err) } @@ -111,6 +133,58 @@ func (a *Agent) RunStep(ctx context.Context, sessionID string, messages []model. return &StepResult{Final: true, Text: resp.Content, StopReason: StopReasonStop, ModelStopReason: resp.StopReason}, nil } +// finishGuardrailSpan 把一次护栏检查的结论落到其 span 上（裁决 11.4.2）： +// props 只放检查名、裁决与耗时（全部在图的 safePropKeys 白名单内，出图时仍会 +// 再清洗一遍），命中拦截或标记时额外记一条脱敏后的 Reason 事件。拦截的 span +// 终态为 error，面板上显红。span 为 nil（未开追踪）时是空操作。 +func finishGuardrailSpan(span *trace.Span, action string, verdict *guardrail.Verdict, err error) { + if span == nil { + return + } + allow := err == nil && (verdict == nil || verdict.Allow) + reason := "" + if verdict != nil { + reason = verdict.Reason + } + if err != nil { + reason = err.Error() + } + _ = span.SetProp("action", action) + _ = span.SetProp("allow", allow) + _ = span.SetProp("elapsed_ms", trace.ElapsedMS(span)) + if err != nil { + _ = span.SetProp("error", err.Error()) + } + if !allow || reason != "" { + _ = span.Record(trace.Event{Type: guardrailEventDone, Props: map[string]any{ + "action": action, + "allow": allow, + "reason": truncateRunes(reason, maxGuardrailReasonRunes), + }}) + } + if allow { + trace.FinishChild(span, trace.SpanStatusOK, nil) + return + } + trace.FinishChild(span, trace.SpanStatusError, nil) +} + +// maxGuardrailReasonRunes 限制护栏 Reason 进入 span 事件前的长度：Reason 是 +// 面向用户的拦截说明，本就不含密钥（guardrail 包契约），这里只防它变成一大段。 +const maxGuardrailReasonRunes = 200 + +// truncateRunes 把 s 截到最多 max 个 rune，超出时补省略号；max 非正时原样返回。 +func truncateRunes(s string, max int) string { + if max <= 0 { + return s + } + r := []rune(s) + if len(r) <= max { + return s + } + return string(r[:max]) + "…" +} + // ExtractToolCalls decodes tool calls from a model.Response. It prefers the // provider-native ToolCalls the adapters populate (Anthropic tool_use / OpenAI // tool_calls), which carry the provider id needed to correlate the result and diff --git a/internal/runtime/loop/flow_spans_test.go b/internal/runtime/loop/flow_spans_test.go new file mode 100644 index 0000000..fa59a9e --- /dev/null +++ b/internal/runtime/loop/flow_spans_test.go @@ -0,0 +1,378 @@ +package loop + +import ( + "context" + "encoding/json" + "strings" + "sync" + "testing" + "time" + + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/agent" + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/guardrail" + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/memory" + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/model" + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/trace" +) + +// 本文件钉住三条裁决在循环侧的行为： +// - 11.4.2：真实跑一轮后，图里真的出现 memory / guardrail 节点（各有生产者）； +// - 11.4.3：tool 节点的参数摘要出图、密钥值绝不出图； +// - 13.4.1：输出侧检查异步（不阻塞回合）、输入侧检查仍在模型调用前同步拦下。 + +// flowRecorder 收集 FlowPublisher 发布的每一帧图，按会话保留最后一份快照。 +// 后台的输出侧检查会从 goroutine 里发布，因此读取必须加锁。 +type flowRecorder struct { + mu sync.Mutex + graphs map[string]*trace.FlowGraph + frames int +} + +func newFlowRecorder() *flowRecorder { + return &flowRecorder{graphs: make(map[string]*trace.FlowGraph)} +} + +// publisher 返回接在该记录器上的实时图发布者。 +func (r *flowRecorder) publisher() *FlowPublisher { + return NewFlowPublisher(func(_ string, payload any) { + g, ok := payload.(*trace.FlowGraph) + if !ok || g == nil { + return + } + r.mu.Lock() + defer r.mu.Unlock() + r.graphs[g.SessionID] = g + r.frames++ + }, func(string) bool { return true }) +} + +// last 返回某会话的最后一份图快照；一帧都没发过时判失败。 +func (r *flowRecorder) last(t *testing.T, session string) *trace.FlowGraph { + t.Helper() + r.mu.Lock() + defer r.mu.Unlock() + g := r.graphs[session] + if g == nil { + t.Fatalf("会话 %q 一帧图都没发布（frames=%d）", session, r.frames) + } + return g +} + +// fixedExtractor 是确定性的记忆提取器：永远返回脚本化的条目，因此一轮结束后 +// 必然发生一次记忆写入（用来证明写侧有生产者）。它不实现 asyncExtractor，所以 +// 提取留在同步路径上，测试无需等后台。 +type fixedExtractor struct{ items []memory.MemoryItem } + +func (e fixedExtractor) Extract(context.Context, []model.Message) ([]memory.MemoryItem, error) { + return e.items, nil +} + +// nodesOfKind 返回图里某一类 kind 的节点，顺序即出现顺序。 +func nodesOfKind(g *trace.FlowGraph, kind string) []trace.FlowNode { + out := make([]trace.FlowNode, 0, len(g.Nodes)) + for _, n := range g.Nodes { + if n.Kind == kind { + out = append(out, n) + } + } + return out +} + +// nodeByKind 返回图里第一个某类 kind 的节点。 +func nodeByKind(t *testing.T, g *trace.FlowGraph, kind string) trace.FlowNode { + t.Helper() + nodes := nodesOfKind(g, kind) + if len(nodes) == 0 { + t.Fatalf("图中没有 %q 节点（实得 kind：%v）", kind, graphKinds(g)) + } + return nodes[0] +} + +// graphKinds 返回图里出现的 kind 列表，仅用于失败信息。 +func graphKinds(g *trace.FlowGraph) []string { + out := make([]string, 0, len(g.Nodes)) + for _, n := range g.Nodes { + out = append(out, n.Kind) + } + return out +} + +// waitUntil 轮询 cond 直到成立或超时（后台检查落定需要一点时间）。 +func waitUntil(t *testing.T, what string, cond func() bool) { + t.Helper() + deadline := time.Now().Add(2 * time.Second) + for time.Now().Before(deadline) { + if cond() { + return + } + time.Sleep(5 * time.Millisecond) + } + t.Fatalf("等待 %s 超时", what) +} + +// TestRun_RealRound_FlowGraphHasMemoryAndGuardrailNodes 真实跑一轮（流式路径， +// 带真实记忆栈、真实护栏管线与实时 tracer/flow），断言图里出现 memory 与 +// guardrail 两类真节点（裁决 11.4.2），并顺带钉住 tool 节点的参数摘要 +// （裁决 11.4.3）：放行的 path 出图、密钥值绝不出图。 +func TestRun_RealRound_FlowGraphHasMemoryAndGuardrailNodes(t *testing.T) { + t.Parallel() + const secret = "sk-live-0123456789abcdef" + rec := newFlowRecorder() + + call := `{"tool_calls":[{"name":"echo","args":{"path":"/tmp/report.txt","api_key":"` + secret + `"}}]}` + ad := &mockAdapter{stream: []mgResp{ + {resp: model.Response{Model: "mock-1", Content: call}}, + {resp: model.Response{Model: "mock-1", Content: "报告读完了"}}, + }} + td := dispatcherWith(t, &mockTool{name: "echo", result: "报告内容"}) + a := newTestAgent(t, ad, td) + store := newTestMemoryStore(t) + l := New(Options{ + Agent: a, Tools: td, MaxRounds: 8, + Tracer: trace.New(nil), + Flow: rec.publisher(), + Guards: guardrail.New(), + Memory: store, + Retriever: memory.NewRetriever(store), + Extractor: fixedExtractor{items: []memory.MemoryItem{{Content: "主人喜欢甜食"}}}, + PersonaID: "cyrene", + // 流式路径是生产路径（gateway 总会装 OnStream），护栏 span + // 也在这一侧产生，因此这条「真实跑一轮」用例走它。 + OnStream: func(model.StreamEvent) {}, + }) + + res, err := l.Run(context.Background(), "web", []model.Message{model.TextMessage(model.RoleUser, "看看报告")}) + if err != nil { + t.Fatalf("Run: %v", err) + } + if res.Text != "报告读完了" { + t.Fatalf("Text = %q, want 报告读完了", res.Text) + } + + g := rec.last(t, "web") + assertMemoryNodes(t, g) + assertGuardrailNodes(t, rec) + assertToolArgsSummary(t, rec.last(t, "web"), secret) +} + +// assertMemoryNodes 断言读侧（召回）与写侧（提取落库）各留下一个真节点。 +func assertMemoryNodes(t *testing.T, g *trace.FlowGraph) { + t.Helper() + memoryNodes := nodesOfKind(g, "memory") + if len(memoryNodes) < 2 { + t.Fatalf("memory 节点 = %d 个，want ≥2（召回 + 写入）；图中 kind：%v", len(memoryNodes), graphKinds(g)) + } + recall := memoryNodes[0] + if got := recall.Props["action"]; got != "recall" { + t.Fatalf("memory 节点 action = %v，want recall（props：%+v）", got, recall.Props) + } + if got := recall.Props["scope"]; got != "owner" { + t.Fatalf("memory 召回 scope = %v，want owner（props：%+v）", got, recall.Props) + } +} + +// assertGuardrailNodes 断言输入/输出两侧都留下节点且都落下裁决。输出侧检查是 +// 异步的（裁决 13.4.1），因此先等它的裁决落定。 +func assertGuardrailNodes(t *testing.T, rec *flowRecorder) { + t.Helper() + waitUntil(t, "两类 guardrail 节点都落下裁决", func() bool { + guardNodes := nodesOfKind(rec.last(t, "web"), "guardrail") + if len(guardNodes) < 2 { + return false + } + for _, n := range guardNodes { + if _, ok := n.Props["allow"].(bool); !ok { + return false + } + } + return true + }) + g := rec.last(t, "web") + actions := make(map[string]bool) + allowed := 0 + for _, n := range nodesOfKind(g, "guardrail") { + action, _ := n.Props["action"].(string) + actions[action] = true + if allow, _ := n.Props["allow"].(bool); allow { + allowed++ + } + } + if !actions[spanNameGuardrailInput] || !actions[spanNameGuardrailOutput] { + t.Fatalf("guardrail 节点缺输入/输出任一侧（实得 %v；图中 kind：%v）", actions, graphKinds(g)) + } + if allowed < 2 { + t.Fatalf("allow=true 的 guardrail 节点 = %d 个，want ≥2（props：%+v）", allowed, nodesOfKind(g, "guardrail")) + } +} + +// assertToolArgsSummary 钉住参数摘要的红线：放行的结构类键有值、密钥类键连条目 +// 都没有，且发布的图里不含密钥值。 +func assertToolArgsSummary(t *testing.T, g *trace.FlowGraph, secret string) { + t.Helper() + toolNode := nodeByKind(t, g, "tool") + summary, ok := toolNode.Props["args_summary"].(map[string]any) + if !ok { + t.Fatalf("tool 节点缺 args_summary（props：%+v）", toolNode.Props) + } + if summary["path"] != "/tmp/report.txt" { + t.Fatalf("args_summary[path] = %v，want /tmp/report.txt", summary["path"]) + } + if _, leaked := summary["api_key"]; leaked { + t.Fatalf("密钥类键进了摘要：%+v", summary) + } + keys, ok := toolNode.Props["args_keys"].([]any) + if !ok || len(keys) != 2 { + t.Fatalf("args_keys = %v，want 两个参数名", toolNode.Props["args_keys"]) + } + blob, merr := json.Marshal(g) + if merr != nil { + t.Fatalf("marshal graph: %v", merr) + } + if strings.Contains(string(blob), secret) { + t.Fatalf("发布的图泄漏了工具参数里的密钥：%s", blob) + } +} + +// gatedOutputGuard 的输出侧检查会一直等到 release 关闭（或 context 结束）才返回 +// 拦截裁决，用来证明该检查真的落在关键路径之外。 +type gatedOutputGuard struct { + entered chan struct{} + release chan struct{} + once sync.Once +} + +func (g *gatedOutputGuard) CheckInput(context.Context, *model.Request) (*guardrail.Verdict, error) { + return &guardrail.Verdict{Allow: true}, nil +} + +func (g *gatedOutputGuard) CheckOutput(ctx context.Context, _ *model.Response) (*guardrail.Verdict, error) { + g.once.Do(func() { close(g.entered) }) + select { + case <-g.release: + return &guardrail.Verdict{Allow: false, Reason: "输出含敏感内容"}, nil + case <-ctx.Done(): + return nil, ctx.Err() + } +} + +// TestStreamStep_OutputGuardrailDoesNotBlockTurn 是裁决 13.4.1 的输出侧用例： +// 输出侧检查挂住不放时，回合仍必须返回（真并发、不阻塞）；检查放行后它的拦截 +// 结果仍要闭环 —— 落进 guardrail 节点并发出审计事件（异步 ≠ 可绕过）。 +func TestStreamStep_OutputGuardrailDoesNotBlockTurn(t *testing.T) { + t.Parallel() + rec := newFlowRecorder() + guard := &gatedOutputGuard{entered: make(chan struct{}), release: make(chan struct{})} + ad := &mockAdapter{stream: []mgResp{{resp: model.Response{Model: "mock-1", Content: "回答正文"}}}} + td := dispatcherWith(t) + a := newTestAgent(t, ad, td) + + var mu sync.Mutex + var audits []AuditEvent + l := New(Options{ + Agent: a, Tools: td, MaxRounds: 8, + Tracer: trace.New(nil), Flow: rec.publisher(), + Guards: guardrail.New(guardrail.WithGuard(guard)), + OnStream: func(model.StreamEvent) {}, + OnAudit: func(ev AuditEvent) { + mu.Lock() + audits = append(audits, ev) + mu.Unlock() + }, + }) + + type runOut struct { + res *Result + err error + } + done := make(chan runOut, 1) + go func() { + res, err := l.Run(context.Background(), "web", []model.Message{model.TextMessage(model.RoleUser, "你好")}) + done <- runOut{res: res, err: err} + }() + + select { + case <-guard.entered: + case <-time.After(2 * time.Second): + t.Fatal("输出侧检查始终没有启动") + } + var out runOut + select { + case out = <-done: + case <-time.After(2 * time.Second): + t.Fatal("回合被输出侧护栏阻塞了：检查未放行时 Run 必须已经返回（裁决 13.4.1 真并发）") + } + if out.err != nil { + t.Fatalf("Run: %v", out.err) + } + if out.res.Text != "回答正文" { + t.Fatalf("Text = %q，want 模型产出（输出侧检查不得改写已流转的回复）", out.res.Text) + } + + close(guard.release) + waitUntil(t, "输出侧拦截闭环", func() bool { + mu.Lock() + audited := len(audits) == 1 && audits[0].What == auditGuardrailOutputBlock + mu.Unlock() + if !audited { + return false + } + g := rec.last(t, "web") + for _, n := range nodesOfKind(g, "guardrail") { + if n.Props["action"] == spanNameGuardrailOutput && n.Props["allow"] == false { + return n.Status == trace.FlowStatusError + } + } + return false + }) +} + +// denyingInputGuard 直接拒掉输入侧检查。 +type denyingInputGuard struct{} + +func (denyingInputGuard) CheckInput(context.Context, *model.Request) (*guardrail.Verdict, error) { + return &guardrail.Verdict{Allow: false, Reason: "输入含敏感内容"}, nil +} + +func (denyingInputGuard) CheckOutput(context.Context, *model.Response) (*guardrail.Verdict, error) { + return &guardrail.Verdict{Allow: true}, nil +} + +// TestStreamStep_InputGuardrailBlocksBeforeModel 是裁决 13.4.1 的输入侧用例： +// 输入侧检查仍是模型调用之前的同步前置门 —— 拒绝时模型一次都不许被调用，回合 +// 直接带着拦截理由返回，且图上留下一个显红的 guardrail 节点。 +func TestStreamStep_InputGuardrailBlocksBeforeModel(t *testing.T) { + t.Parallel() + rec := newFlowRecorder() + ad := &mockAdapter{stream: []mgResp{{resp: model.Response{Model: "mock-1", Content: "不该出现"}}}} + td := dispatcherWith(t) + a := newTestAgent(t, ad, td) + l := New(Options{ + Agent: a, Tools: td, MaxRounds: 8, + Tracer: trace.New(nil), Flow: rec.publisher(), + Guards: guardrail.New(guardrail.WithGuard(denyingInputGuard{})), + OnStream: func(model.StreamEvent) {}, + }) + + res, err := l.Run(context.Background(), "web", []model.Message{model.TextMessage(model.RoleUser, "危险输入")}) + if err != nil { + t.Fatalf("Run: %v", err) + } + if res.Text != "输入含敏感内容" || res.StopReason != agent.StopReasonGuardrailInput { + t.Fatalf("res = %+v，want 输入侧拦截理由与 guardrail_input", res) + } + if ad.streamCalls != 0 { + t.Fatalf("模型被调用了 %d 次，输入侧前置门必须在模型之前拦下", ad.streamCalls) + } + + g := rec.last(t, "web") + node := nodeByKind(t, g, "guardrail") + if node.Props["action"] != spanNameGuardrailInput { + t.Fatalf("guardrail 节点 action = %v，want %s", node.Props["action"], spanNameGuardrailInput) + } + if allow, _ := node.Props["allow"].(bool); allow { + t.Fatalf("被拦下的 guardrail 节点 allow = %v，want false（props：%+v）", node.Props["allow"], node.Props) + } + if node.Status != trace.FlowStatusError { + t.Fatalf("guardrail 节点状态 = %q，want error", node.Status) + } +} diff --git a/internal/runtime/loop/memory.go b/internal/runtime/loop/memory.go index d86b33e..86cd909 100644 --- a/internal/runtime/loop/memory.go +++ b/internal/runtime/loop/memory.go @@ -7,6 +7,7 @@ import ( "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/memory" "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/model" + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/trace" ) // memoryExtractTimeout 限制单次后台提取。它只管辖 @@ -75,12 +76,21 @@ func (l *Loop) memoryBlock(ctx context.Context, messages []model.Message) string // 角色日常记忆（ScopePersona）属敏感项，绝不注入 // （design/03 §5「不泄露主人敏感信息」+ 主人 2026-10-03 // 裁决「用户习惯记忆与不敏感数据：可查、不可写」）。 +// +// 召回本身包在一个 memory span 里（裁决 11.4.2，design/19 §1 +// 旁路节点）：span 只记元数据（作用域/命中条数/注入块字节数/耗时）， +// 绝不记召回了什么内容。 func (l *Loop) retrieveMemoryBlock(ctx context.Context, env SessionEnv, personaID, query string) (string, error) { if !env.Restricted() { - return l.retriever.Inject(ctx, personaID, query, 0) + rctx, span := l.span(ctx, spanNameMemoryRecall, trace.SpanKindMemory) + block, err := l.retriever.Inject(rctx, personaID, query, 0) + l.finishMemoryRecallSpan(span, "owner", 0, len(block), err) + return block, err } - results, err := l.retriever.Retrieve(ctx, personaID, query, 0) + rctx, span := l.span(ctx, spanNameMemoryRecall, trace.SpanKindMemory) + results, err := l.retriever.Retrieve(rctx, personaID, query, 0) if err != nil { + l.finishMemoryRecallSpan(span, "restricted", 0, 0, err) return "", err } shared := make([]memory.Result, 0, len(results)) @@ -90,9 +100,12 @@ func (l *Loop) retrieveMemoryBlock(ctx context.Context, env SessionEnv, personaI } } if len(shared) == 0 { + l.finishMemoryRecallSpan(span, "restricted", 0, 0, nil) return "", nil } - return formatMemoryBlock(shared, "群聊隔离"), nil + block := formatMemoryBlock(shared, "群聊隔离") + l.finishMemoryRecallSpan(span, "restricted", len(shared), len(block), nil) + return block, nil } // sharedScope 报告 scope 是否为群聊/陌生人可查的非敏感共享域： @@ -278,18 +291,31 @@ func (l *Loop) extractMemoryAsync(ctx context.Context, messages []model.Message, // （memory.FormatSourceConversation），无条件覆盖 extractor // 产出值 —— 模型自由文本绝无机会落库为来源。store // 失败会被记录日志，绝不致命，因此一条被拒的条目无法中止一个 turn。 +// +// 整批写入包在一个 memory span 里（裁决 11.4.2，design/19 §1 旁路节点）： +// 只记写入/失败条数与耗时，绝不记写入了什么内容；没有条目时不产生节点 +// （span 绝不宣告未运行的工作）。 func (l *Loop) persistMemories(ctx context.Context, items []memory.MemoryItem, step int) { + if len(items) == 0 { + return + } personaID := l.personaFor(ctx) source := memory.FormatSourceConversation(loopSession(ctx), loopTurn(ctx)) + wctx, span := l.span(ctx, spanNameMemoryWrite, trace.SpanKindMemory) + written, failed := 0, 0 for _, it := range items { if personaID != "" { it.PersonaID = personaID } it.SourceConversation = source - if _, err := l.memory.Add(ctx, it); err != nil { + if _, err := l.memory.Add(wctx, it); err != nil { + failed++ l.log.Warn("memory store", "step", step, "error", err) + continue } + written++ } + l.finishMemoryWriteSpan(span, written, failed, nil) } // queryFromMessages 从对话导出检索查询：最后一个用户 diff --git a/internal/runtime/loop/spanlog.go b/internal/runtime/loop/spanlog.go index 164f6d9..cabfa7b 100644 --- a/internal/runtime/loop/spanlog.go +++ b/internal/runtime/loop/spanlog.go @@ -1,10 +1,12 @@ package loop import ( + "fmt" "sort" - "time" + "strings" "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/agent" + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/guardrail" "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/trace" ) @@ -16,12 +18,49 @@ const ( EventLLMDone = "llm.done" // EventToolDone 在工具调用落定后记录到 tool_call span 上。 EventToolDone = "tool.done" + // eventGuardrailDone 在输入/输出护栏检查落定后记录到 guardrail span 上。 + eventGuardrailDone = "guardrail.done" + // eventMemoryDone 在记忆召回/写入落定后记录到 memory span 上。 + eventMemoryDone = "memory.done" +) + +// 旁路节点（design/19 §1，裁决 11.4.2）的 span 名：guardrail 记输入/输出 +// 检查，memory 记召回/写入。名字即图上的节点标签。 +const ( + spanNameGuardrailInput = "guardrail.input" + spanNameGuardrailOutput = "guardrail.output" + spanNameMemoryRecall = "memory.recall" + spanNameMemoryWrite = "memory.write" ) // maxLogSummaryRunes 在发布前限制日志取值的长度。图在 // 输出时还会再次脱敏与截断；这里只是让单个 span 保持小。 const maxLogSummaryRunes = 240 +// 工具参数摘要的规格（裁决 11.4.3，权威定义见 DevDocs/design/19 §1）： +// 单值最多 maxArgValueRunes 个 rune，一次调用最多放行 maxArgSummaryKeys 个参数值。 +const ( + maxArgValueRunes = 80 + maxArgSummaryKeys = 24 +) + +// argValueKeys 是「值可显示」的参数键放行表（裁决 11.4.3）：只放行结构/元数据类 +// 键，表外键一律只出键名。工具参数正是凭据、对话正文或含隐私的文件路径最容易 +// 出现的地方，因此放行表取保守口径 —— 想放行一个新键，必须先改设计再改这里。 +var argValueKeys = map[string]bool{ + // 文件与路径（读写目标本身是运维信息，不是内容） + "path": true, "file": true, "file_path": true, "filename": true, + "dir": true, "directory": true, "input_path": true, "output_path": true, + // 检索/匹配（查询词与模式，出图前仍经凭据形状清洗） + "query": true, "q": true, "pattern": true, "name": true, "id": true, "url": true, + // 分类与格式 + "kind": true, "scope": true, "format": true, "encoding": true, "mode": true, + "project": true, "project_id": true, "timezone": true, + // 时间窗与分页（纯数值） + "date": true, "from": true, "to": true, "limit": true, "offset": true, "page": true, + "top_k": true, "max_results": true, "count": true, +} + // recordLLMDone 把一次已完成模型调用的脱敏摘要追加到其 // span：服务该 turn 的模型、两个 stop reason、请求了多少工具调用 // 以及该调用耗时。不记录任何提示词或补全文本， @@ -52,16 +91,30 @@ func (l *Loop) recordLLMDone(span *trace.Span, out *stepOutcome, ag *agent.Agent // 刻意省略 —— 工具参数正是凭据、消息正文 // 或含机密的文件路径会出现的地方 —— 且这里记录的一切 // 在发布前都会由图再次清洗。 +// +// 参数摘要（裁决 11.4.3，规格见 design/19 §1）：除全量参数名 +// （args_keys）外，另生成 args_summary —— 只含放行表内、且非密钥类 +// 键的值，单值截断到 maxArgValueRunes。摘要同时写进 span prop +// （图的 safePropKeys 放行它，面板节点详情直接读），使「可视化思考 +// 链路」看得到参数而看不到凭据。 func (l *Loop) recordToolDone(span *trace.Span, tc agent.ToolCall, out ToolOutcome) { if span == nil { return } + keys := argKeys(tc.Args) props := map[string]any{ "tool": tc.Name, "ok": out.Err == nil, - "args_keys": argKeys(tc.Args), + "args_keys": keys, "elapsed_ms": elapsedMS(span), } + // 键名清单同时写进 span prop：面板节点详情要列出全部参数名， + // 再对放行的键显示截断后的值、对其余键显示「值未展示」。 + _ = span.SetProp("args_keys", keys) + if summary := argSummary(tc.Args); len(summary) > 0 { + props["args_summary"] = summary + _ = span.SetProp("args_summary", summary) + } if out.Err != nil { if out.ErrKind != "" { props["error_kind"] = out.ErrKind @@ -73,6 +126,156 @@ func (l *Loop) recordToolDone(span *trace.Span, tc agent.ToolCall, out ToolOutco _ = span.Record(trace.Event{Type: EventToolDone, Props: props}) } +// argSummary 生成工具参数的「值摘要」：只放行 argValueKeys 表内的键，且 +// 密钥类键一律不给值（trace.IsSecretKey 与出图清洗同一判据）；非标量值只给 +// 占位（数组(N)/对象(N)），因为容器里可能藏着正文或凭据。键按字典序处理， +// 使摘要只随参数变化，不随 map 迭代顺序抖动。 +func argSummary(args map[string]any) map[string]any { + if len(args) == 0 { + return nil + } + out := make(map[string]any) + for _, k := range argKeys(args) { + if len(out) >= maxArgSummaryKeys { + break + } + if !argValueVisible(k) { + continue + } + if s := summarizeArgValue(args[k]); s != "" { + out[k] = ellipsize(s, maxArgValueRunes) + } + } + if len(out) == 0 { + return nil + } + return out +} + +// argValueVisible 报告某参数键的值是否允许出图：必须是放行表内的结构类键， +// 且不得是密钥类键（放行表本身不含密钥名，这里是纵深防御）。 +func argValueVisible(key string) bool { + if trace.IsSecretKey(key) { + return false + } + return argValueKeys[strings.ToLower(strings.TrimSpace(key))] +} + +// summarizeArgValue 把参数值渲染成一行短文本：字符串原样，数字/布尔按字面量， +// 数组与对象只给占位而不展开（不把容器内的正文或凭据带出来），其余类型返回空串。 +func summarizeArgValue(v any) string { + switch x := v.(type) { + case nil: + return "" + case string: + return x + case bool, float64, float32, int, int64, int32, uint, uint64, uint32: + return fmt.Sprintf("%v", x) + case []any: + return fmt.Sprintf("数组(%d)", len(x)) + case []string: + return fmt.Sprintf("数组(%d)", len(x)) + case map[string]any: + return fmt.Sprintf("对象(%d)", len(x)) + default: + return "" + } +} + +// finishGuardrailSpan 把一次护栏检查（输入侧同步门 / 输出侧异步检查）的结论 +// 落到其 span 上（裁决 11.4.2，design/19 §1 旁路节点）：props 只放检查名、 +// 裁决与耗时（全部在 safePropKeys 白名单内，出图时仍会再清洗一遍），事件带 +// 一条脱敏后的 Reason 供节点日志查阅。拦截时 span 终态记为 error，面板上显红。 +// err 非 nil 表示检查本身失败（不是裁决），同样记为 error 并把原因脱敏后留下。 +func (l *Loop) finishGuardrailSpan(span *trace.Span, action string, verdict *guardrail.Verdict, err error) { + if span == nil { + return + } + allow := err == nil && (verdict == nil || verdict.Allow) + props := map[string]any{ + "action": action, + "allow": allow, + "elapsed_ms": elapsedMS(span), + } + reason := "" + if verdict != nil { + reason = verdict.Reason + } + if err != nil { + props["error"] = err.Error() + reason = err.Error() + } + for k, v := range props { + _ = span.SetProp(k, v) + } + // 一条事件同时覆盖「拦截原因」与「标记放行」（SensitiveMark 会带 Reason + // 放行），使 mark 模式在面板上有消费者而不是无声通过。 + if !allow || reason != "" { + _ = span.Record(trace.Event{Type: eventGuardrailDone, Props: map[string]any{ + "action": action, + "allow": allow, + "reason": trimSummary(reason), + }}) + } + if allow { + l.finishSpan(span, trace.SpanStatusOK) + return + } + l.finishSpan(span, trace.SpanStatusError) +} + +// finishMemoryRecallSpan 记录一次记忆召回的元数据：作用域、命中条数、注入块 +// 字节数与耗时。绝不记录召回了什么内容（design/19 §1：memory 旁路节点只记 +// 元数据，不含敏感内容）—— 记忆正文本身就可能是主人的隐私。 +func (l *Loop) finishMemoryRecallSpan(span *trace.Span, scope string, hits, size int, err error) { + if span == nil { + return + } + props := map[string]any{ + "action": "recall", + "scope": scope, + "bytes": size, + "hits": hits, + "elapsed_ms": elapsedMS(span), + } + l.finishMemorySpan(span, props, err) +} + +// finishMemoryWriteSpan 记录一次记忆写入的元数据：写入与失败的条数、耗时。 +// 绝不记录写入了什么内容（同 recall 的口径）。 +func (l *Loop) finishMemoryWriteSpan(span *trace.Span, written, failed int, err error) { + if span == nil { + return + } + props := map[string]any{ + "action": "write", + "count": written, + "failed": failed, + "elapsed_ms": elapsedMS(span), + } + l.finishMemorySpan(span, props, err) +} + +// finishMemorySpan 是两次记忆读写的共同收尾：写 props、记一条事件、按结果 +// 关闭 span（失败记 error 终态）。props 由调用方给全，收尾只负责落盘与终态。 +func (l *Loop) finishMemorySpan(span *trace.Span, props map[string]any, err error) { + if span == nil { + return + } + if err != nil { + props["error"] = err.Error() + } + for k, v := range props { + _ = span.SetProp(k, v) + } + _ = span.Record(trace.Event{Type: eventMemoryDone, Props: props}) + if err != nil { + l.finishSpan(span, trace.SpanStatusError) + return + } + l.finishSpan(span, trace.SpanStatusOK) +} + // modelName 报告服务某步的模型：流式路径知道时用路由出的 // 请求模型，否则用该 turn 的 agent 配置的模型。两者 // 都不可用时为空。 @@ -88,11 +291,9 @@ func modelName(out *stepOutcome, ag *agent.Agent) string { // elapsedMS 报告 span 迄今已打开多久。span 的 Duration 仅由 // Finish 打上，因此在 Finish 之前记录的日志事件在此测量它。 +// 实现委托给 trace.ElapsedMS，使 agent 侧的护栏 span 与本包用同一算法。 func elapsedMS(span *trace.Span) int64 { - if span == nil || span.StartedAt.IsZero() { - return 0 - } - return time.Since(span.StartedAt).Milliseconds() + return trace.ElapsedMS(span) } // argKeys 按排序返回工具调用的参数名。只记录 @@ -110,11 +311,16 @@ func argKeys(args map[string]any) []string { } // trimSummary 把日志取值缩短到 maxLogSummaryRunes 个 rune， -// 且不切开多字节 rune。 +// 且不切开多字节 rune（被截断时末位换成省略号）。 func trimSummary(s string) string { - r := []rune(s) - if len(r) <= maxLogSummaryRunes { + return ellipsize(s, maxLogSummaryRunes) +} + +// ellipsize 把 s 截到最多 max 个 rune；确实被截断时末位换成省略号， +// 使面板上的一眼能看出「这里被截过」。max 非正时原样返回。 +func ellipsize(s string, max int) string { + if max <= 0 || len([]rune(s)) <= max { return s } - return string(r[:maxLogSummaryRunes]) + "…" + return truncateRunes(s, max-1) + "…" } diff --git a/internal/runtime/loop/stream.go b/internal/runtime/loop/stream.go index 71669f1..240a876 100644 --- a/internal/runtime/loop/stream.go +++ b/internal/runtime/loop/stream.go @@ -9,14 +9,29 @@ import ( "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/agent" "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/model" + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/trace" ) +// guardrailCheckTimeout 限制一次后台输出侧护栏检查。它只管辖脱离 +// 出来的检查 context，因此一个挂起的检查不会让 goroutine 泄露到进程 +// 生命周期结束，也不会拖住任何回合。 +const guardrailCheckTimeout = 30 * time.Second + +// auditGuardrailOutputBlock 是输出侧护栏在后台判定拦截时发出的审计 +// "what"。输出侧检查已移出关键路径（裁决 13.4.1），拦截因此必须闭环： +// 一条审计事件 + guardrail 节点上的 verdict，「异步」不等于「可绕过」。 +const auditGuardrailOutputBlock = "guardrail.output.block" + // streamStep 驱动一次流式模型调用。它是 // agent.RunStep 的流式对应物：它从 agent 的公开表面构建请求、 // 运行循环的输入护栏、把增量流经 OnStream、运行输出 // 护栏并解码所有工具调用。由于 RunStep 不暴露流式 // 路径，该方法归循环所有；它镜像 agent 的语义，使两者 // 除增量回调外行为一致。 +// +// 两侧护栏的分工（裁决 13.4.1，design/04 §5）：**输入侧是模型调用前的同步 +// 前置门**（恶意输入不得先消耗一次模型调用），**输出侧是异步并行检查** +// （模型产出不等待检查即可继续流转，结果异步落定）。 func (l *Loop) streamStep(ctx context.Context, messages []model.Message) (*stepOutcome, error) { ag := l.agentFor(ctx) adapter := ag.Adapter() @@ -30,7 +45,11 @@ func (l *Loop) streamStep(ctx context.Context, messages []model.Message) (*stepO req := l.buildRequest(ctx, messages, ag, knobs) if l.guards != nil { - verdict, err := l.guards.CheckInput(ctx, &req) + // 输入侧同步前置门：检查包在自己的 guardrail span 里 + // （裁决 11.4.2），因此护栏也进了全链路流程图。 + gctx, gspan := l.span(ctx, spanNameGuardrailInput, trace.SpanKindGuardrail) + verdict, err := l.guards.CheckInput(gctx, &req) + l.finishGuardrailSpan(gspan, spanNameGuardrailInput, verdict, err) if err != nil { return nil, fmt.Errorf("loop: check input: %w", err) } @@ -65,15 +84,8 @@ func (l *Loop) streamStep(ctx context.Context, messages []model.Message) (*stepO return nil, fmt.Errorf("loop: chat stream: %w", err) } - if l.guards != nil { - verdict, err := l.guards.CheckOutput(ctx, &resp) - if err != nil { - return nil, fmt.Errorf("loop: check output: %w", err) - } - if verdict != nil && !verdict.Allow { - return &stepOutcome{final: true, text: verdict.Reason, stopReason: agent.StopReasonGuardrailOutput}, nil - } - } + // 输出侧不阻塞回合：检查在后台跑，本回合带着模型产出继续。 + l.checkOutputAsync(ctx, &resp) if calls, ok := decodeToolCalls(resp); ok { return &stepOutcome{final: false, toolCalls: calls, stopReason: agent.StopReasonToolCalls, @@ -83,6 +95,37 @@ func (l *Loop) streamStep(ctx context.Context, messages []model.Message) (*stepO modelStopReason: resp.StopReason, model: resp.Model}, nil } +// checkOutputAsync 在后台运行输出侧护栏检查（裁决 13.4.1：真并发、不阻塞 +// 主会话）。启动 span 与「运行中」占位节点留在同步路径上，节点的出现顺序 +// 才与真实发生顺序一致；检查本身脱离本步的 context 跑（保留会话/actor 值， +// 另加超时兜底），因此慢检查绝不拖住回合。 +// +// 结果闭环在 guardrail 节点上（allow / 耗时 / 脱敏后的 Reason，裁决 11.4.2）， +// 判定拦截时再发一条审计事件 —— 输出侧的拦截始终留痕，只是不再挡住已产出的 +// 回复（这正是「真并发」与「可绕过」的区别：输入侧的前置门依旧同步拦下）。 +func (l *Loop) checkOutputAsync(ctx context.Context, resp *model.Response) { + if l.guards == nil || resp == nil { + return + } + gctx, span := l.span(ctx, spanNameGuardrailOutput, trace.SpanKindGuardrail) + bg, cancel := context.WithTimeout(context.WithoutCancel(gctx), guardrailCheckTimeout) + go func() { + defer cancel() + verdict, err := l.guards.CheckOutput(bg, resp) + l.finishGuardrailSpan(span, spanNameGuardrailOutput, verdict, err) + if verdict != nil && !verdict.Allow { + // Reason 由 guardrail 包保证不含被匹配的密钥（其契约）， + // 这里再经 RedactText 走一遍凭据形状清洗 + 截断，使审计面 + // 与出图面用同一判据。 + l.emitAudit(AuditEvent{ + Who: string(sessionEnvFrom(bg).ActorEnv), + What: auditGuardrailOutputBlock, + Result: trace.RedactText(verdict.Reason), + }) + } + }() +} + // broadcasts 报告某流式事件是否可推给 UI。只有 // 思考帧会被扣下，且仅当该 turn 的角色把 // showToUser 解析为 false 时；文本、工具调用与工具进度帧 diff --git a/internal/runtime/orchestration/flowround_test.go b/internal/runtime/orchestration/flowround_test.go new file mode 100644 index 0000000..06f7323 --- /dev/null +++ b/internal/runtime/orchestration/flowround_test.go @@ -0,0 +1,274 @@ +package orchestration + +import ( + "context" + "encoding/json" + "io" + "log/slog" + "strings" + "sync" + "testing" + + "git.yeij.top/AskaEth/Cyrene_v2/internal/capability/registry" + "git.yeij.top/AskaEth/Cyrene_v2/internal/db" + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/agent" + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/guardrail" + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/loop" + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/memory" + "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" +) + +// roundExtractor 是确定性的记忆提取器：一轮结束后必然产生一次记忆写入， +// 使「真实跑一轮」里 memory 的写侧也有节点（不实现 asyncExtractor，因此 +// 提取留在同步路径上，测试不用等后台）。 +type roundExtractor struct{} + +func (roundExtractor) Extract(context.Context, []model.Message) ([]memory.MemoryItem, error) { + return []memory.MemoryItem{{Content: "子代理记住了这次委托"}}, nil +} + +// delegateTool 模拟「派活」工具：它把前台会话当作 ParentSession 派一个后台 +// 子代理（与生产里的 subagent_spawn 同一个归属口径：会话来自 ctx，绝不从 +// 模型输出推断），并记下被派发的子代理 id 供测试等待。 +type delegateTool struct { + sched *Scheduler + mu sync.Mutex + spawn []string + err error +} + +func (d *delegateTool) Name() string { return "delegate" } +func (d *delegateTool) Description() string { return "dispatch a background sub-agent" } +func (d *delegateTool) Parameters() (map[string]any, error) { return nil, nil } + +func (d *delegateTool) Execute(ctx context.Context, _ any) (any, error) { + session := loop.SessionFromContext(ctx) + if session == "" { + d.setErr(context.Canceled) + return "no session in context", nil + } + // 与生产的 subagent_spawn 同一口径：后台子代理必须活得比前台回合久， + // 因此派发用一个脱离取消但保留会话值的 context。 + spawnCtx := context.WithoutCancel(ctx) + sub, err := d.sched.Spawn(spawnCtx, AgentSpec{Goal: "调研一下", ParentSession: session}) + if err != nil { + d.setErr(err) + return "dispatch failed", nil + } + d.mu.Lock() + d.spawn = append(d.spawn, sub.ID) + d.mu.Unlock() + return "已派发 " + sub.ID, nil +} + +func (d *delegateTool) setErr(err error) { + d.mu.Lock() + defer d.mu.Unlock() + if d.err == nil { + d.err = err + } +} + +func (d *delegateTool) spawned() []string { + d.mu.Lock() + defer d.mu.Unlock() + return append([]string(nil), d.spawn...) +} + +// roundDiscardLogger 返回一个不写任何地方的 logger，保持测试输出干净。 +func roundDiscardLogger() *slog.Logger { + return slog.New(slog.NewTextHandler(io.Discard, nil)) +} + +// flowRound 是一轮「前台 + 后台子代理」的真实运行装置：它把两条真实循环 +// （前台流式路径、子代理模板）、调度器与派活工具串起来。 +type flowRound struct { + front *loop.Loop + flow *loop.FlowPublisher + capture *graphCapture + sched *Scheduler + delegate *delegateTool + ctx context.Context +} + +// newFlowRound 组装装置：真实 SQLite 内存库上的记忆栈、真实护栏、实时 +// tracer/flow，前台走流式路径（生产 chat 路径），子代理模板同样带记忆。 +func newFlowRound(t *testing.T) *flowRound { + t.Helper() + capture := &graphCapture{} + flow := loop.NewFlowPublisher(capture.publish, nil) + tracer := trace.New(nil) + + ctx := context.Background() + sqlDB, oerr := db.Open(ctx, ":memory:", roundDiscardLogger()) + if oerr != nil { + t.Fatalf("db.Open: %v", oerr) + } + t.Cleanup(func() { _ = sqlDB.Close() }) + memStore, merr := memory.NewStore(ctx, sqlDB, roundDiscardLogger()) + if merr != nil { + t.Fatalf("memory.NewStore: %v", merr) + } + retriever := memory.NewRetriever(memStore) + + front, frontDispatch := newFrontRoundLoop(t, flow, tracer, memStore, retriever) + subLoop := newSubRoundLoop(t, flow, tracer, memStore, retriever) + sched := newScheduler(t, Options{MaxParallel: 1, Loop: subLoop, Tracer: tracer, Flow: flow, Memory: memStore}) + + delegate := &delegateTool{sched: sched} + if rerr := frontDispatch.Register(delegate); rerr != nil { + t.Fatalf("Register delegate: %v", rerr) + } + return &flowRound{front: front, flow: flow, capture: capture, sched: sched, delegate: delegate, ctx: ctx} +} + +// newFrontRoundLoop 构建前台循环：派活工具 + 流式路径（生产 chat 路径）+ 真实 +// 记忆/护栏。它把 dispatcher 一并返回，调用方才能挂上需要调度器的派活工具。 +func newFrontRoundLoop(t *testing.T, flow *loop.FlowPublisher, tracer *trace.Tracer, memStore *memory.Store, retriever *memory.Retriever) (*loop.Loop, *tool.Dispatcher) { + t.Helper() + dispatch, derr := tool.New(registry.New()) + if derr != nil { + t.Fatalf("tool.New: %v", derr) + } + adapter := &toolThenFinalAdapter{toolName: "delegate", args: map[string]any{"goal": "调研"}, final: "已派发，稍后回填"} + ag, aerr := agent.New("front", dispatch, nil, + agent.WithModel(adapter, nil), agent.WithOptions(agent.Options{Model: "mock-1"})) + if aerr != nil { + t.Fatalf("agent.New: %v", aerr) + } + front := loop.New(loop.Options{ + Agent: ag, Tools: dispatch, Tracer: tracer, Flow: flow, MaxRounds: 8, + Guards: guardrail.New(), Memory: memStore, Retriever: retriever, + Extractor: roundExtractor{}, PersonaID: "cyrene", + OnStream: func(model.StreamEvent) {}, + }) + return front, dispatch +} + +// newSubRoundLoop 构建子代理模板：真实循环 + 真实记忆检索，因此泳道里也有 +// memory 节点，且它的 llm/tool 步骤会折进派生它的会话的图。 +func newSubRoundLoop(t *testing.T, flow *loop.FlowPublisher, tracer *trace.Tracer, memStore *memory.Store, retriever *memory.Retriever) *loop.Loop { + t.Helper() + dispatch, serr := tool.New(registry.New()) + if serr != nil { + t.Fatalf("tool.New: %v", serr) + } + if rerr := dispatch.Register(&echoTool{name: "echo", result: map[string]any{"ok": true}}); rerr != nil { + t.Fatalf("Register: %v", rerr) + } + adapter := &toolThenFinalAdapter{toolName: "echo", args: map[string]any{ + "path": "/tmp/sub-report.txt", + "api_key": secretArg, + }, final: "子代理完成"} + ag, aerr := agent.New("be brief", dispatch, nil, + agent.WithModel(adapter, nil), agent.WithOptions(agent.Options{Model: "mock-1"})) + if aerr != nil { + t.Fatalf("agent.New: %v", aerr) + } + return loop.New(loop.Options{ + Agent: ag, Tools: dispatch, Tracer: tracer, Flow: flow, MaxRounds: 8, + Memory: memStore, Retriever: retriever, Extractor: roundExtractor{}, PersonaID: "cyrene", + }) +} + +// run 跑完前台一轮、等子代理收尾，并返回前台会话的最后一份图。 +func (r *flowRound) run(t *testing.T) (*trace.FlowGraph, string) { + t.Helper() + r.flow.StartRound(parentSession, "帮我派个活") + res, err := r.front.Run(r.ctx, parentSession, []model.Message{model.TextMessage(model.RoleUser, "帮我派个活")}) + if err != nil { + t.Fatalf("front Run: %v", err) + } + r.flow.EndRound(parentSession, res.Text) + if len(r.delegate.spawned()) != 1 { + t.Fatalf("派发次数 = %v，want 1（delegate 错误：%v）", r.delegate.spawned(), r.delegate.err) + } + subID := r.delegate.spawned()[0] + merged, werr := r.sched.Wait(r.ctx, []string{subID}) + if werr != nil { + t.Fatalf("Wait: %v", werr) + } + if merged.Succeeded != 1 { + t.Fatalf("子代理未成功：%+v（summary=%s）", merged, merged.Summary) + } + return r.capture.graph(t, parentSession), subID +} + +// TestRealRound_FlowGraphHasAllThreeNodeKinds 是一条端到端的「真实跑一轮」用例 +// （裁决 11.4.2）：前台会话带着真实记忆栈、真实护栏与实时 tracer/flow 跑完一轮， +// 期间经工具派发一个后台子代理 —— 同一张图里必须同时出现 memory（记忆召回/写入）、 +// guardrail（输入/输出护栏）与 orchestration（派发标记）三类节点，且工具参数里的 +// 密钥值绝不出图（裁决 11.4.3 红线）。 +func TestRealRound_FlowGraphHasAllThreeNodeKinds(t *testing.T) { + t.Parallel() + r := newFlowRound(t) + g, subID := r.run(t) + + kinds := make(map[string]int) + for _, n := range g.Nodes { + kinds[n.Kind]++ + } + for _, want := range []string{"memory", "guardrail", "orchestration"} { + if kinds[want] == 0 { + t.Fatalf("图中缺 %q 节点（实得 %v）", want, kinds) + } + } + assertOrchestrationNode(t, g, subID) + assertSubagentLane(t, g, subID) + assertRoundGraphRedaction(t, g) +} + +// assertOrchestrationNode 断言派发标记：带被派发的子代理 id、收尾为 done。 +func assertOrchestrationNode(t *testing.T, g *trace.FlowGraph, subID string) { + t.Helper() + orch := flowKindNodes(g, "orchestration") + if len(orch) != 1 { + t.Fatalf("orchestration 节点 = %d 个，want 1", len(orch)) + } + if got := orch[0].Props["subagent"]; got != subID { + t.Fatalf("orchestration 节点 subagent = %v，want %s（props：%+v）", got, subID, orch[0].Props) + } + if orch[0].Status != trace.FlowStatusDone { + t.Fatalf("orchestration 节点状态 = %q，want done（props：%+v）", orch[0].Status, orch[0].Props) + } +} + +// assertSubagentLane 断言泳道仍在：子代理根 + 它自己的步骤链（含记忆召回与工具调用）。 +func assertSubagentLane(t *testing.T, g *trace.FlowGraph, subID string) { + t.Helper() + subNode := flowSubagentNode(t, g, subID) + lane := flowReachable(g, subNode.ID) + hasMemoryInLane, hasToolInLane := false, false + for _, n := range g.Nodes { + if !lane[n.ID] { + continue + } + if n.Kind == "memory" { + hasMemoryInLane = true + } + if n.Kind == "tool" { + hasToolInLane = true + } + } + if !hasMemoryInLane || !hasToolInLane { + t.Fatalf("子代理泳道缺 memory/tool 步骤（memory=%v tool=%v）", hasMemoryInLane, hasToolInLane) + } +} + +// assertRoundGraphRedaction 断言红线：工具参数里的密钥不得出现在发布的图里， +// 而放行的结构类键仍要可读（可视化要能看参数）。 +func assertRoundGraphRedaction(t *testing.T, g *trace.FlowGraph) { + t.Helper() + blob, jerr := json.Marshal(g) + if jerr != nil { + t.Fatalf("marshal graph: %v", jerr) + } + if strings.Contains(string(blob), secretArg) { + t.Fatalf("发布的图泄漏了工具参数：%s", blob) + } + if !strings.Contains(string(blob), "/tmp/sub-report.txt") { + t.Fatalf("放行键 path 的值没出图：%s", blob) + } +} diff --git a/internal/runtime/orchestration/scheduler.go b/internal/runtime/orchestration/scheduler.go index 3d8fb03..7b1191a 100644 --- a/internal/runtime/orchestration/scheduler.go +++ b/internal/runtime/orchestration/scheduler.go @@ -322,6 +322,9 @@ func (s *Scheduler) run(ctx context.Context, sub *SubAgent) { // 工具白名单（design/05 §7 裁决 02.4.1）随运行上下文传递：名单非空时 // 只放行所列工具；空名单＝全开，与挂载前的行为一致。 runCtx = loop.WithToolWhitelist(runCtx, sub.ToolWhitelist) + // 派发标记挂在原始 ctx 上（不是 runCtx）：它是这一批派发/收束的 + // orchestration 节点（裁决 11.4.2），与子代理泳道根并列成链。 + orchSpan := s.orchSpan(ctx, sub) runCtx, span := s.subSpan(runCtx, sub) sub.setCancel(cancel) defer cancel() @@ -329,10 +332,12 @@ func (s *Scheduler) run(ctx context.Context, sub *SubAgent) { res, err := s.scopedRunner(sub).Run(runCtx, sub.ID, sub.Context) if err != nil { s.spanErr(sub, span, err) + s.finishOrchSpan(sub, orchSpan, err) s.abortOrFail(sub, err) return } s.spanOK(sub, span, res) + s.finishOrchSpan(sub, orchSpan, nil) sub.setResult(res) s.finish(sub, StatusDone, doneSummary(res), nil) } diff --git a/internal/runtime/orchestration/scheduler_spans.go b/internal/runtime/orchestration/scheduler_spans.go new file mode 100644 index 0000000..e9dfa4d --- /dev/null +++ b/internal/runtime/orchestration/scheduler_spans.go @@ -0,0 +1,65 @@ +package orchestration + +import ( + "context" + + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/trace" +) + +// 本文件是调度层的「编排 span」：把每次子代理派发与收束记成一个 +// orchestration 节点（design/19 §1 旁路节点，裁决 11.4.2）。它从 +// scheduler.go 拆出，因为该文件已接近 AGENTS §10 的 500 行上限， +// 而 span 的开启/收尾是一块独立职责。 + +// orchestration.dispatch 的 span 名与收尾事件类型：此前调度只发 +// SpanKindSubagent，因此全仓没有任何 orchestration 节点。 +const ( + spanNameOrchestrationSpawn = "orchestration.spawn" + orchestrationEventDone = "orchestration.done" +) + +// orchSpan 为一次子代理派发开一个 orchestration span。它是挂在主干上的 +// 「派发标记」：与泳道根并列而不是它的父级 —— 若把它当作子代理 span 的父级， +// 泳道链会多出一节（图构建器把非 subagent 祖先一路穿透解析泳道），既有泳道 +// 契约就被改写了。span 只带非敏感属性（被派发的子代理 id、目标角色 id）， +// 声明了归属会话时同时发布一个「运行中」占位节点。未配置 tracer 时返回 nil。 +func (s *Scheduler) orchSpan(ctx context.Context, sub *SubAgent) *trace.Span { + if s.tracer == nil || sub == nil { + return nil + } + _, span := s.tracer.StartSpan(ctx, spanNameOrchestrationSpawn, trace.SpanKindOrchestration) + if sub.ID != "" { + _ = span.SetProp("subagent", sub.ID) + } + if sub.PersonaID != "" { + _ = span.SetProp("persona", sub.PersonaID) + } + if sess := sub.ParentSession; sess != "" && s.flow != nil { + _ = span.SetProp("session", sess) + s.flow.SpanStarted(sess, span) + } + return span +} + +// finishOrchSpan 收尾一次派发：记录耗时（失败另记原因）与一条脱敏事件， +// 失败时终态为 error，然后把已关闭的 span 折回所属会话的图。 +func (s *Scheduler) finishOrchSpan(sub *SubAgent, span *trace.Span, err error) { + if span == nil { + return + } + props := map[string]any{"elapsed_ms": trace.ElapsedMS(span)} + if err != nil { + props["error"] = err.Error() + _ = span.SetProp("error", err.Error()) + } + _ = span.Record(trace.Event{Type: orchestrationEventDone, Props: props}) + status := trace.SpanStatusOK + if err != nil { + status = trace.SpanStatusError + } + span.Finish(status) + if s.flow == nil || sub == nil || sub.ParentSession == "" { + return + } + s.flow.SpanFinished(sub.ParentSession, span) +} diff --git a/internal/runtime/trace/graph_build.go b/internal/runtime/trace/graph_build.go index 50b0deb..379c7cd 100644 --- a/internal/runtime/trace/graph_build.go +++ b/internal/runtime/trace/graph_build.go @@ -70,17 +70,23 @@ func (g *RunGraph) Upsert(spec NodeSpec) { if from != "" && from != spec.ID { g.addEdge(from, spec.ID) } - // A sub-agent root is drawn on the trunk it hangs off, but it does not become - // the trunk's continuation point: the main chain resumes where the branch left - // it (the spawn node). Without this, every node after a spawn would look like a - // step of the last branch (the frontend derives lanes from the edges). The - // first trunk node still advances the cursor, so a graph built without an input - // seed cannot orphan its later trunk nodes. - if lane != "" || n.Kind != nodeKindSubagent || g.laneLast[trunkLane] == "" { + // 两类节点是挂在主干上的「标记」而非主干的续点，因此不推进主干游标： + // subagent 根开出一条泳道（见下），orchestration 记录一次派发/收束 + // （design/19 §1 旁路节点，裁决 11.4.2）—— 它们之后的主干节点仍从原 + // 位置续链，标记本身不会被误当成主干的一步。泳道内的节点（lane != ""） + // 照常推进本泳道游标。 + if lane != "" || !branchMarker(n.Kind) || g.laneLast[trunkLane] == "" { g.laneLast[lane] = spec.ID } } +// branchMarker 报告某类节点是否为「主干上的标记」：子代理泳道根与编排 +// 派发节点都挂在主干上却不接替主干链，使主干（input→llm→tool→…→output） +// 与并行分支、派发标记各成一条不会被彼此打断的链。 +func branchMarker(kind string) bool { + return kind == nodeKindSubagent || kind == nodeKindOrchestrate +} + // rememberParent records a span's parent link even when the span is not rendered // as a node, so laneOf can walk through skipped containers. func (g *RunGraph) rememberParent(id, parent string) { diff --git a/internal/runtime/trace/graph_nodeprod_test.go b/internal/runtime/trace/graph_nodeprod_test.go new file mode 100644 index 0000000..68c1ead --- /dev/null +++ b/internal/runtime/trace/graph_nodeprod_test.go @@ -0,0 +1,137 @@ +package trace + +import ( + "context" + "encoding/json" + "strings" + "testing" + "time" +) + +// graphNodeKinds 返回图里出现的节点 kind 集合。 +func graphNodeKinds(g *FlowGraph) map[string]bool { + out := make(map[string]bool) + for _, n := range g.Nodes { + out[n.Kind] = true + } + return out +} + +// spanRecordOn 在实时 tracer 上开一个 span、挂上 props、关掉它，并返回完成记录。 +// id 由 tracer 生成，因此每个记录天然唯一。 +func spanRecordOn(t *testing.T, tr *Tracer, name string, kind SpanKind, props map[string]any) SpanRecord { + t.Helper() + _, sp := tr.StartSpan(context.Background(), name, kind) + for k, v := range props { + if err := sp.SetProp(k, v); err != nil { + t.Fatalf("SetProp(%q): %v", k, err) + } + } + sp.Finish(SpanStatusOK) + return sp.Snapshot() +} + +// TestGraphBuilder_MemoryGuardrailOrchestrationNodes 钉住三类旁路节点真的能 +// 成为图上的节点（裁决 11.4.2）：memory / guardrail / orchestration 三类 span +// 各有生产者后，kindForSpan 的映射必须把它们原样折进图，而不是当未知 kind 丢掉。 +func TestGraphBuilder_MemoryGuardrailOrchestrationNodes(t *testing.T) { + t.Parallel() + tr := New(nil) + g := NewRunGraph("web") + g.Start(1, time.Now()) + g.SeedInput("帮我看看") + + g.AddSpan(spanRecordOn(t, tr, "memory.recall", SpanKindMemory, map[string]any{"action": "recall", "scope": "owner"})) + g.AddSpan(spanRecordOn(t, tr, "guardrail.input", SpanKindGuardrail, map[string]any{"action": "guardrail.input", "allow": true})) + g.AddSpan(spanRecordOn(t, tr, "orchestration.spawn", SpanKindOrchestration, map[string]any{"subagent": "sub-1"})) + + got := graphNodeKinds(g.Snapshot()) + for _, want := range []string{"memory", "guardrail", "orchestration"} { + if !got[want] { + t.Fatalf("图中缺 %q 节点（实得 %v）", want, got) + } + } +} + +// TestGraphBuilder_OrchestrationMarkerKeepsTrunkCursor 钉住编排派发标记的 +// 定位：它挂在主干上但不接替主干链 —— 之后的 llm/tool 仍从原位置续链， +// 派发标记与泳道根一样只是主干上的一个标记（否则整条主干会被标记截断）。 +func TestGraphBuilder_OrchestrationMarkerKeepsTrunkCursor(t *testing.T) { + t.Parallel() + tr := New(nil) + g := NewRunGraph("web") + g.Start(1, time.Now()) + g.SeedInput("派活") + + orch := spanRecordOn(t, tr, "orchestration.spawn", SpanKindOrchestration, nil) + g.AddSpan(orch) + llm := spanRecordOn(t, tr, "llm.call", SpanKindLLM, nil) + g.AddSpan(llm) + + graph := g.Snapshot() + if !hasEdge(graph, "input", orch.ID) { + t.Fatalf("编排标记未挂在主干 input 上：%+v", graph.Edges) + } + if !hasEdge(graph, "input", llm.ID) { + t.Fatalf("主干被编排标记截断：llm 应从 input 续链，实得 %+v", graph.Edges) + } +} + +// TestSafeSpanProps_ToolArgsSummaryKeepsSecretsOut 是发布面的脱敏用例 +// （裁决 11.4.3）：把带密钥的参数摘要喂进 tool span，断言出图后没有密钥值， +// 只留下放行键的摘要与键名清单，且密钥类键在摘要里被替换为 [redacted]。 +func TestSafeSpanProps_ToolArgsSummaryKeepsSecretsOut(t *testing.T) { + t.Parallel() + const secret = "sk-live-0123456789abcdef" + tr := New(nil) + g := NewRunGraph("web") + g.Start(1, time.Now()) + g.SeedInput("读文件") + + g.AddSpan(spanRecordOn(t, tr, "tool.fs.read", SpanKindTool, map[string]any{ + "args_keys": []string{"api_key", "content", "path"}, + "args_summary": map[string]any{ + "path": "/tmp/report.txt", + "api_key": secret, + }, + "result": secret, + })) + + blob, err := json.Marshal(g.Snapshot()) + if err != nil { + t.Fatalf("marshal graph: %v", err) + } + out := string(blob) + if strings.Contains(out, secret) { + t.Fatalf("出图泄漏了密钥值：%s", out) + } + if !strings.Contains(out, "/tmp/report.txt") { + t.Fatalf("放行键的值没出图（可视化要能看参数）：%s", out) + } + if !strings.Contains(out, "args_summary") || !strings.Contains(out, "args_keys") { + t.Fatalf("摘要键未随节点发布：%s", out) + } + if !strings.Contains(out, redactedMarker) { + t.Fatalf("密钥类键的值应被替换为 %s：%s", redactedMarker, out) + } + if strings.Contains(out, "\"content\"") && strings.Contains(out, "args_summary\":{\"content") { + t.Fatalf("表外键的值不应进入摘要：%s", out) + } +} + +// TestRedactValue_StringKeyedSummaryIsScrubbed 钉住 map[string]string 形态的 +// 摘要同样逐键清洗：生产者若改用字符串键值对，密钥仍不能原样出图。 +func TestRedactValue_StringKeyedSummaryIsScrubbed(t *testing.T) { + t.Parallel() + const secret = "sk-live-0123456789abcdef" + out, ok := redactValue(map[string]string{"api_key": secret, "path": "/tmp/a"}).(map[string]any) + if !ok { + t.Fatal("map[string]string 未被清洗为 map[string]any") + } + if out["api_key"] != redactedMarker { + t.Fatalf("密钥类键 = %v，want %s", out["api_key"], redactedMarker) + } + if out["path"] != "/tmp/a" { + t.Fatalf("放行键值被改坏：%v", out["path"]) + } +} diff --git a/internal/runtime/trace/graph_props.go b/internal/runtime/trace/graph_props.go index 2f73a9a..68f3c9a 100644 --- a/internal/runtime/trace/graph_props.go +++ b/internal/runtime/trace/graph_props.go @@ -39,6 +39,17 @@ func safeSpanProps(rec SpanRecord) map[string]any { // safePropKeys are the span prop keys forwarded to a node's props. // +// 键分五类，全部非敏感（值仍会逐个经 redactValue 清洗）： +// +// - 既有：stop_reason / step / error / error_kind / tool / persona； +// - tool 参数摘要（裁决 11.4.3）：args_keys（全量参数名）、args_summary +// （仅放行键的值，长度已在 loop/spanlog.go 侧截断），密钥类键的值绝不在其中； +// - guardrail（裁决 11.4.2）：allow（裁决结果）、action（检查名）、elapsed_ms（耗时）； +// - memory（裁决 11.4.2）：hits / count / failed（命中、写入与失败条数）、 +// bytes（注入块字节数）、scope / action（记忆作用域与操作名）—— 只记元数据， +// 不含任何记忆正文； +// - orchestration（裁决 11.4.2）：subagent（被派发的子代理 id）。 +// // persona 是子代理分支根节点上的「目标角色 id」：orchestration/scheduler.go 只在 // subagent_spawn 显式指定角色时才设置它，且该 id 必须先通过角色注册表校验 // （ErrUnknownPersona 会拒绝未注册的 id，orchestration/persona.go），因此它是注册表里 @@ -46,7 +57,13 @@ func safeSpanProps(rec SpanRecord) map[string]any { // 与会话材料对外可见。它是 Docs/api-client.md §3.5（v0.47）对客户端的承诺 // （design/22 §8-2：让流程图显示「这个子代理是以谁的身份在替班」），漏出白名单就会被 // safeSpanProps 静默丢掉。未指定角色派发时该键不存在，符合 v0.47「纯增量」的约定。 -var safePropKeys = []string{"stop_reason", "step", "error", "error_kind", "tool", "persona"} +var safePropKeys = []string{ + "stop_reason", "step", "error", "error_kind", "tool", "persona", + "args_keys", "args_summary", + "allow", "action", "elapsed_ms", + "hits", "count", "failed", "bytes", "scope", + "subagent", +} // summarize renders an arbitrary value as a short text string for a prop summary. func summarize(v any) string { diff --git a/internal/runtime/trace/graph_span.go b/internal/runtime/trace/graph_span.go new file mode 100644 index 0000000..15923d1 --- /dev/null +++ b/internal/runtime/trace/graph_span.go @@ -0,0 +1,62 @@ +package trace + +import ( + "context" + "time" +) + +// 本文件是 span 生产者侧的桥接层：它让不持有 *Tracer 句柄的叶子包 +// （internal/runtime/agent 的护栏检查等）也能在当前 span 之下再开一个 +// 子 span，从而把「尽可能深」的结构化追踪补齐（design/04 §9：LLM / +// 工具 / 编排 / 状态转移 / 子代理 / 记忆读写 / guardrail 全都要有生产者）。 +// +// 图谱（graph*.go）与生产者（span 生产点）在这里分工：图构建器只负责把 +// span 折成节点，生产者只负责如实记录发生了什么。两者都不需要对方存在 +// —— 没有 tracer 时 ChildSpan 返回 nil，生产点照常跑完业务。 + +// ChildSpan 在当前 ctx 携带的 span 之下开一个具名字 span：新 span 复用其父 +// span 的 Tracer（因此 Finish 时同样交给 Writer 落盘），ParentID 指向父 +// span，返回的 ctx 把新 span 作为当前 span 继续往下传。 +// +// ctx 未携带 span（未开追踪、或在根 span 之外调用）时返回 (ctx, nil)： +// 调用方必须容忍 nil span（不要用它，也不要把 nil span 的 ctx 当真）。 +// 空 ctx 按 context.Background() 处理。 +func ChildSpan(ctx context.Context, name string, kind SpanKind) (context.Context, *Span) { + if ctx == nil { + ctx = context.Background() + } + parent := SpanFromContext(ctx) + if parent == nil { + return ctx, nil + } + parent.mu.RLock() + tracer := parent.tracer + parent.mu.RUnlock() + if tracer == nil { + return ctx, nil + } + span := tracer.newSpan(name, kind, parent) + return context.WithValue(ctx, spanKey{}, span), span +} + +// FinishChild 关闭一个由 ChildSpan 开出的 span 并记录耗时与结果：span 为 +// nil（未开追踪）时是空操作，因此生产点可以无条件调用它，不必在调用处 +// 再写一遍 nil 判断。status 为 ok/error，err 为 nil 时代码不记录 error 属性。 +func FinishChild(span *Span, status SpanStatus, err error) { + if span == nil { + return + } + if err != nil { + _ = span.SetProp("error", err.Error()) + } + span.Finish(status) +} + +// ElapsedMS 报告 span 自开始以来经过的毫秒数；nil span 或零 StartAt 返回 0。 +// 它让叶子包在不依赖 loop 内部助手的情况下记录「耗时」属性。 +func ElapsedMS(span *Span) int64 { + if span == nil || span.StartedAt.IsZero() { + return 0 + } + return time.Since(span.StartedAt).Milliseconds() +} diff --git a/internal/runtime/trace/redact.go b/internal/runtime/trace/redact.go index 46621b3..563adbe 100644 --- a/internal/runtime/trace/redact.go +++ b/internal/runtime/trace/redact.go @@ -42,8 +42,10 @@ var secretKeyNames = map[string]bool{ // secretKeySuffixes catch prefixed variants such as openai_api_key. var secretKeySuffixes = []string{"_api_key", "_access_token", "_auth_token", "_secret", "_token"} -// isSecretKey reports whether a map key names a credential. -func isSecretKey(key string) bool { +// IsSecretKey 报告某个 map 键是否命名了一个凭据。它是「密钥类键」的唯一判据： +// 出图前的清洗与参数摘要的生产者（loop/spanlog.go，裁决 11.4.3「密钥类参数 +// 一律不显示值」）共用它，因此生产侧与最后一道防线不可能判得不一致。 +func IsSecretKey(key string) bool { k := strings.NewReplacer("-", "_", ".", "_").Replace(strings.ToLower(strings.TrimSpace(key))) if secretKeyNames[k] { return true @@ -80,7 +82,7 @@ func RedactEvent(ev Event) Event { func redactProps(props map[string]any) map[string]any { out := make(map[string]any, len(props)) for k, v := range props { - if isSecretKey(k) { + if IsSecretKey(k) { out[k] = redactedMarker continue } @@ -135,6 +137,18 @@ func redactValue(v any) any { out[i] = RedactText(item) } return out + case map[string]string: + // 参数摘要以 map[string]string 生产时也必须逐键清洗， + // 否则密钥会绕过键名规则原样出图（裁决 11.4.3 红线）。 + out := make(map[string]any, len(x)) + for k, v := range x { + if IsSecretKey(k) { + out[k] = redactedMarker + continue + } + out[k] = RedactText(v) + } + return out default: return v }