feat(trace): M40 span 写侧脱敏 + 共享上游错误收敛 helper
stderr 的 JSONL 是无主人在场的落盘面，却在此前原样落工具结果（result）、
子代理回复（text/summary）与上游报文，等于一次 tail 就能拷走整段对话。
本切片交付两件地基：
1) 写侧脱敏（trace.JSONRecorder）
- 新增 trace/span_props.go：safePropKeys 白名单与 projectSpanProps 投影，
读侧（safeSpanProps）与写侧共用同一张表、同一个投影函数；
- WriteSpan 落盘前经 trace.SanitizeRecord 投影：白名单键清洗，表外键只留
Digest（bytes/runes/sha256 前缀，收进 omitted），事件先 RedactEvent 再把
正文键折成 Digest。落到磁盘的只有元数据，正文一个字节都不留；
- 新增 trace/digest.go：Digest/DigestOf/DigestOfValue，给「脱敏后仍要能诊断」
一个构造上不含正文的取值形态（体积 + 指纹），两条投影都按类型放行它。
2) 共享上游错误体收敛 helper（给 M41/M42，禁止三家各写一份）
- trace.UpstreamError(upstream, status, contentType, body)：只保留 status /
contentType / bodyLen / body sha256 前缀 / 上游 JSON 的 error.code /
error.type / 截断 128 rune 的 error.message，其余一律丢弃，响应体原文绝不
进 error 或日志；*UpstreamErrorInfo 提供 Error()/Attrs()/Unwrap()（errors.Is
可判 ErrUpstream 走降级链）；
- trace.HostOf / trace.MaskQuery：日志里的 URL 只回来源，凭据类 query 与
fragment 值打码为 ***，user:pass@ 整段打码。
3) 剥掉三处正文上 span
- loop：tool span 不再放 result，改放 result_summary（与 tool_execution.end
帧同源、已脱敏截断）与 result_digest；
- orchestration：subagent span 不再放 text，subagent.done 事件不再放 summary，
改记 status/steps/chars/sha256；
- automations：AddRuleByName 不再 %v 整个 payload，改记 subject/type/bytes/sha256。
测试：以哨兵正文在落盘字节上断言（JSONL / slog 输出），并分别对「编排层收到的
原始 span」与「写侧投影」两层各钉一遍；UpstreamError 用回显 input 的 422 报文
与 HTML 错误页各钉一条；解析出/未解析出 JSON 都不回原文。以上均做过变异验证
（去掉保护即变红）。
验收：go build ./... + go vet ./... + go test -count=1 ./... 全绿；
trace/loop/orchestration 另跑 -race 全绿。
This commit is contained in:
AskaEth committed
1 parent
240b59708d
commit
003d12a712
18 files changed
+1612
-71
No files matched your search
@@ -394,21 +394,29 @@ func (s *Scheduler) subSpan(ctx context.Context, sub *SubAgent) (context.Context
|
||||
}
|
||||
|
||||
// subagentEventDone 是子代理运行落定时记在其 span 上的日志事件类型，使面板节点带上
|
||||
// 结果摘要（design/19 §1 节点实时日志）。
|
||||
// 结果的元数据（状态/步数/回复体积与指纹，绝无正文；design/19 §1 节点实时日志）。
|
||||
const subagentEventDone = "subagent.done"
|
||||
|
||||
// spanOK 把子代理 span 以 ok 收尾，记录结果文本与一条脱敏的结果事件，随后把已关闭的
|
||||
// span 折回父 flow。
|
||||
// spanOK 把子代理 span 以 ok 收尾，记录**结果的体积与指纹**（不是正文）与一条同口径
|
||||
// 的结果事件，随后把已关闭的 span 折回父 flow。
|
||||
//
|
||||
// 🔴 子代理的回复正文不进 span：它本来就要留在会话库（那是主人要读的对话），
|
||||
// span 与 stderr JSONL 只是落盘的诊断面 —— 把整段回复再抄一遍，等于给一次
|
||||
// stderr tail 就能拷走全部对话开了口子（M40 修掉的最严重一条泄漏）。这里留
|
||||
// text_digest（字节/rune/sha256 前缀）与事件里的 status/steps/chars/sha256，
|
||||
// 足以分辨「这次子代理答了多久、答了多少、是不是同一份」，却拿不到一个字。
|
||||
func (s *Scheduler) spanOK(sub *SubAgent, span *trace.Span, res *loop.Result) {
|
||||
if span == nil {
|
||||
return
|
||||
}
|
||||
if res != nil {
|
||||
_ = span.SetProp("text", res.Text)
|
||||
digest := trace.DigestOf(res.Text)
|
||||
_ = span.SetProp("text_digest", digest)
|
||||
_ = span.Record(trace.Event{Type: subagentEventDone, Props: map[string]any{
|
||||
"status": string(StatusDone),
|
||||
"steps": res.Steps,
|
||||
"summary": compactText(res.Text, summaryTextMax),
|
||||
"status": string(StatusDone),
|
||||
"steps": res.Steps,
|
||||
"chars": digest.Runes,
|
||||
"sha256": digest.SHA256,
|
||||
}})
|
||||
}
|
||||
span.Finish(trace.SpanStatusOK)
|
||||
|
||||
@@ -0,0 +1,176 @@
|
||||
package orchestration
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"strings"
|
||||
"sync"
|
||||
"testing"
|
||||
|
||||
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/loop"
|
||||
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/trace"
|
||||
)
|
||||
|
||||
// 本文件钉住 M40 的子代理侧写盘边界：子代理的**整段回复**曾经以 span prop
|
||||
// （`text`）与 `subagent.done` 事件值（`summary`）两种形态原样落到 stderr 的
|
||||
// JSONL 上 —— 一次 `tail` 就能拷走整段对话。现在落盘只有元数据：
|
||||
// text_digest（字节/rune/sha256 前缀）与事件里的 status/steps/chars/sha256。
|
||||
//
|
||||
// 🔴 断言打在落盘字节上（真派发 + 真 tracer + 真 JSONRecorder），不打在
|
||||
// 「我们没调用 SetProp("text")」这种实现细节上：以后谁把正文塞回来，这里就红。
|
||||
|
||||
// spannedScheduler 建一个带 tracer 的调度器：span 同时进「原始记录捕获器」与
|
||||
// stderr JSONL 记录器，使编排层与写侧能各自被断言（只测 JSONL 的话，写侧白名单
|
||||
// 会替编排层的漏做遮挡）。
|
||||
func spannedScheduler(t *testing.T) (*Scheduler, *captureWriter, *bytes.Buffer) {
|
||||
t.Helper()
|
||||
capture := &captureWriter{}
|
||||
buf := &bytes.Buffer{}
|
||||
tr := trace.New(teeWriter{capture, trace.NewJSONRecorder(buf)})
|
||||
return newScheduler(t, Options{MaxParallel: 1, Tracer: tr}), capture, buf
|
||||
}
|
||||
|
||||
// captureWriter 收下每个 span 的**原始**记录（未经写侧投影）。
|
||||
type captureWriter struct {
|
||||
mu sync.Mutex
|
||||
recs []trace.SpanRecord
|
||||
}
|
||||
|
||||
// WriteSpan 记录一份 span。
|
||||
func (c *captureWriter) WriteSpan(rec trace.SpanRecord) error {
|
||||
c.mu.Lock()
|
||||
defer c.mu.Unlock()
|
||||
c.recs = append(c.recs, rec)
|
||||
return nil
|
||||
}
|
||||
|
||||
// subagent 返回已捕获记录里的那条子代理 span。
|
||||
func (c *captureWriter) subagent(t *testing.T) trace.SpanRecord {
|
||||
t.Helper()
|
||||
c.mu.Lock()
|
||||
defer c.mu.Unlock()
|
||||
for _, rec := range c.recs {
|
||||
if rec.Kind == trace.SpanKindSubagent {
|
||||
return rec
|
||||
}
|
||||
}
|
||||
t.Fatalf("捕获记录里没有 subagent span：%+v", c.recs)
|
||||
return trace.SpanRecord{}
|
||||
}
|
||||
|
||||
// teeWriter 把同一份 span 交给多个出口，返回第一个错误。
|
||||
type teeWriter []trace.Writer
|
||||
|
||||
// WriteSpan 依次写入每个出口。
|
||||
func (t teeWriter) WriteSpan(rec trace.SpanRecord) error {
|
||||
var firstErr error
|
||||
for _, w := range t {
|
||||
if err := w.WriteSpan(rec); err != nil && firstErr == nil {
|
||||
firstErr = err
|
||||
}
|
||||
}
|
||||
return firstErr
|
||||
}
|
||||
|
||||
// TestSpawn_SubagentReplyNeverLandsInJSONL 是最直接的泄漏钉：编排层收到的原始
|
||||
// span 里没有回复原文，落盘的 JSONL 里也没有。
|
||||
func TestSpawn_SubagentReplyNeverLandsInJSONL(t *testing.T) {
|
||||
t.Parallel()
|
||||
const reply = "SENTINEL-SUBAGENT-REPLY-2f91-她说的一整段话"
|
||||
s, capture, buf := spannedScheduler(t)
|
||||
s.runner = &mockRunner{result: &loop.Result{Text: reply, Steps: 3, StopReason: "stop"}}
|
||||
|
||||
sub, err := s.Spawn(context.Background(), AgentSpec{Goal: "查一下天气"})
|
||||
if err != nil {
|
||||
t.Fatalf("Spawn: %v", err)
|
||||
}
|
||||
waitFor(t, func() bool { return sub.Status() == StatusDone }, waitTimeout)
|
||||
|
||||
raw := buf.String()
|
||||
if strings.Contains(raw, reply) {
|
||||
t.Fatalf("子代理回复原文落进了 JSONL：%s", raw)
|
||||
}
|
||||
// 编排层：span 的 props 里没有 text 原文，只有 text_digest。
|
||||
span := capture.subagent(t)
|
||||
if _, ok := span.Props["text"]; ok {
|
||||
t.Fatalf("子代理 span 仍带着 text 原文：%+v", span.Props)
|
||||
}
|
||||
digest, ok := span.Props["text_digest"].(trace.Digest)
|
||||
if !ok {
|
||||
t.Fatalf("子代理 span 没有 text_digest：%+v", span.Props)
|
||||
}
|
||||
if digest.Runes != len([]rune(reply)) {
|
||||
t.Fatalf("text_digest.Runes = %d, want %d（回复体积必须仍可读）", digest.Runes, len([]rune(reply)))
|
||||
}
|
||||
if digest.Bytes != len(reply) {
|
||||
t.Fatalf("text_digest.Bytes = %d, want %d", digest.Bytes, len(reply))
|
||||
}
|
||||
}
|
||||
|
||||
// TestSpawn_SubagentDoneEventCarriesMetadataOnly 钉住事件侧：`subagent.done` 只带
|
||||
// 状态/步数/字符数/指纹，不再带一段回复摘要 —— 摘要也是正文。断言打在**编排层收到的
|
||||
// 原始事件**上（写侧另有自己的投影，不能替这一层担保）。
|
||||
func TestSpawn_SubagentDoneEventCarriesMetadataOnly(t *testing.T) {
|
||||
t.Parallel()
|
||||
const reply = "SENTINEL-SUBAGENT-DONE-77aa"
|
||||
s, capture, buf := spannedScheduler(t)
|
||||
s.runner = &mockRunner{result: &loop.Result{Text: reply, Steps: 2}}
|
||||
|
||||
sub, err := s.Spawn(context.Background(), AgentSpec{Goal: "go"})
|
||||
if err != nil {
|
||||
t.Fatalf("Spawn: %v", err)
|
||||
}
|
||||
waitFor(t, func() bool { return sub.Status() == StatusDone }, waitTimeout)
|
||||
|
||||
if raw := buf.String(); strings.Contains(raw, reply) {
|
||||
t.Fatalf("事件里带着子代理回复：%s", raw)
|
||||
}
|
||||
rec := capture.subagent(t)
|
||||
if len(rec.Events) != 1 || rec.Events[0].Type != subagentEventDone {
|
||||
t.Fatalf("events = %+v, want one %s", rec.Events, subagentEventDone)
|
||||
}
|
||||
props := rec.Events[0].Props
|
||||
if _, ok := props["summary"]; ok {
|
||||
t.Fatalf("subagent.done 仍带 summary（摘要也是正文）：%+v", props)
|
||||
}
|
||||
if _, ok := props["text"]; ok {
|
||||
t.Fatalf("subagent.done 仍带 text：%+v", props)
|
||||
}
|
||||
if props["status"] != string(StatusDone) || props["chars"] == nil || props["sha256"] == nil || props["steps"] == nil {
|
||||
t.Fatalf("subagent.done props = %+v，want status/steps/chars/sha256", props)
|
||||
}
|
||||
if got := props["chars"]; got != len([]rune(reply)) {
|
||||
t.Fatalf("chars = %v, want %d（回复体积必须仍可读）", got, len([]rune(reply)))
|
||||
}
|
||||
}
|
||||
|
||||
// TestSpawn_SubagentFailureStillDiagnosable 钉住失败侧仍可诊断：落盘里仍能看出
|
||||
// 原因（"boom"），且不带凭据原文 —— 脱敏不等于把 span 变哑。
|
||||
func TestSpawn_SubagentFailureStillDiagnosable(t *testing.T) {
|
||||
t.Parallel()
|
||||
const secret = "abcdef1234567890"
|
||||
s, capture, buf := spannedScheduler(t)
|
||||
s.runner = &mockRunner{err: &runnerErr{msg: "boom: token=" + secret}}
|
||||
|
||||
sub, err := s.Spawn(context.Background(), AgentSpec{Goal: "go"})
|
||||
if err != nil {
|
||||
t.Fatalf("Spawn: %v", err)
|
||||
}
|
||||
waitFor(t, func() bool { return sub.Status() == StatusFailed }, waitTimeout)
|
||||
|
||||
raw := buf.String()
|
||||
if strings.Contains(raw, secret) {
|
||||
t.Fatalf("JSONL 带着凭据原文：%s", raw)
|
||||
}
|
||||
if !strings.Contains(raw, "boom") {
|
||||
t.Fatalf("失败落盘里没有可诊断的原因：%s", raw)
|
||||
}
|
||||
if span := capture.subagent(t); span.Status != trace.SpanStatusError {
|
||||
t.Fatalf("子代理 span 终态 = %q, want error", span.Status)
|
||||
}
|
||||
}
|
||||
|
||||
// runnerErr 是只提供 Error() 的运行失败（模拟子代理内部／上游拼出的错误文本）。
|
||||
type runnerErr struct{ msg string }
|
||||
|
||||
func (e *runnerErr) Error() string { return e.msg }
|
||||
Reference in new issue
Block a user