Merge pull request 'Align spans with the OTel GenAI convention (N3 WP4)' (#32) from feature/n3-otel-wp4 into main
ci / vet-and-test (push) Successful in 53s
ci / docker-checkout-hostgateway (push) Successful in 0s

Align spans with the OTel GenAI convention (N3 WP4)
This commit was merged in pull request #32.
This commit is contained in:
2026-10-07 15:38:07 +00:00
4 changed files with 436 additions and 32 deletions
+75 -4
View File
@@ -4,6 +4,7 @@ import (
"context"
"encoding/json"
"fmt"
"strconv"
"strings"
"sync"
"time"
@@ -12,6 +13,7 @@ import (
"agent-native/runtime/llm"
"agent-native/runtime/reminders"
"agent-native/runtime/session"
"agent-native/runtime/telemetry"
"agent-native/runtime/tools"
"logging-system/logsys"
@@ -116,6 +118,15 @@ type Config struct {
// 本次恢复写入的配对摘要以系统消息注入(不持久化,已在 journal 里)。
// 为假(默认)保持既有语义:Session 为权威上下文。
JournalOnly bool
// AgentName 是 invoke_agent span 的名字后缀(OTel GenAI 约定
// "invoke_agent <agent>")。空 = "agent-native"。
AgentName string
// ProviderName 是 gen_ai.system 属性值(如 openai / anthropic)。
// 空 = "openai"。
ProviderName string
// traceCtx 是本次 Run 的 span 关联(Run 启动时建立;指针随 Config
// 副本传播到各轮与各工具)。
traceCtx *traceIDs
// Hooks 是进程内钩子。空总线不改变停止、提醒和拒绝次数。
Hooks *hooks.Bus
// PermissionAsk 在钩子对一次工具调用表态 ask 时向宿主请求裁决。
@@ -176,7 +187,12 @@ func Run(ctx context.Context, cfg Config, user llm.Message) (Result, error) {
cfg.ToolExecution = ExecSequential
}
serializeHostCallbacks(&cfg)
endRun := startSpan(cfg, "invoke_agent", map[string]string{"model": cfg.Model})
cfg.traceCtx = newTraceContext()
endRun := startSpan(cfg, "invoke_agent "+agentName(cfg), map[string]string{
"gen_ai.system": providerName(cfg),
"gen_ai.agent.name": agentName(cfg),
"gen_ai.request.model": cfg.Model,
}, true)
defer endRun()
var messages []llm.Message
if cfg.Session != nil && !cfg.JournalOnly {
@@ -438,7 +454,11 @@ func systemText(cfg Config) string {
}
func consume(ctx context.Context, cfg Config, messages []llm.Message) (llm.Event, error) {
endSpan := startSpan(cfg, "chat", map[string]string{"model": cfg.Model})
spanAttrs := map[string]string{
"gen_ai.system": providerName(cfg),
"gen_ai.request.model": cfg.Model,
}
endSpan := startSpan(cfg, "chat "+cfg.Model, spanAttrs, false)
defer endSpan()
var decls []llm.ToolDecl
if cfg.Tools != nil {
@@ -475,6 +495,11 @@ func consume(ctx context.Context, cfg Config, messages []llm.Message) (llm.Event
sawEnd = true
}
}
// usage 在 span 结束前补进属性(结束时过滤导出,见 telemetry 工厂)。
if sawEnd {
spanAttrs["gen_ai.usage.input_tokens"] = itoa(end.Usage.PromptTokens)
spanAttrs["gen_ai.usage.output_tokens"] = itoa(end.Usage.CompletionTokens)
}
if !sawEnd {
return llm.Event{Message: llm.Message{StopReason: llm.StopError}}, errString("stream ended without a message")
}
@@ -686,7 +711,10 @@ func runOne(ctx context.Context, cfg Config, call llm.ToolCall) llm.Message {
emit(cfg, Event{Type: "tool_execution_end", Tool: call.Name, Text: msg.Content})
return msg
}
endSpan := startSpan(cfg, "execute_tool", map[string]string{"tool": call.Name})
endSpan := startSpan(cfg, "execute_tool "+call.Name, map[string]string{
"gen_ai.system": providerName(cfg),
"gen_ai.tool.name": call.Name,
}, false)
defer endSpan()
res := tools.Result{Content: "no tools configured", IsError: true}
known := cfg.Tools != nil && cfg.Tools.Has(call.Name)
@@ -864,10 +892,38 @@ func commitToolIntent(cfg Config, call llm.ToolCall) error {
})
}
func startSpan(cfg Config, name string, attrs map[string]string) func() {
func itoa(n int) string { return strconv.Itoa(n) }
// traceIDs 是一次 Run 的 span 关联:根 trace + invoke_agent 根 span。
type traceIDs struct {
traceID string
rootID string
}
// newTraceContext 建立一次 Run 的关联(Trace 为 nil 时也给空壳,调用方免判空)。
func newTraceContext() *traceIDs {
return &traceIDs{traceID: telemetry.NewTraceID(), rootID: telemetry.NewSpanID()}
}
// startSpan 开一个 OTel GenAI span:root=true 时作 invoke_agent 根(无 parent),
// 否则作子 span(parent = 根)。attrs 由调用方持有引用——存续期间可补
// (usage 在结束时随过滤导出)。
func startSpan(cfg Config, name string, attrs map[string]string, root bool) func() {
if cfg.Trace == nil {
return func() {}
}
if attrs == nil {
attrs = map[string]string{}
}
if cfg.traceCtx != nil {
attrs[telemetry.AttrTraceID] = cfg.traceCtx.traceID
if root {
attrs[telemetry.AttrSpanID] = cfg.traceCtx.rootID
} else {
attrs[telemetry.AttrSpanID] = telemetry.NewSpanID()
attrs[telemetry.AttrParentSpanID] = cfg.traceCtx.rootID
}
}
end := cfg.Trace(name, attrs)
if end == nil {
return func() {}
@@ -875,6 +931,21 @@ func startSpan(cfg Config, name string, attrs map[string]string) func() {
return end
}
// agentName / providerName 是 OTel 属性的默认值解析。
func agentName(cfg Config) string {
if strings.TrimSpace(cfg.AgentName) != "" {
return cfg.AgentName
}
return "agent-native"
}
func providerName(cfg Config) string {
if strings.TrimSpace(cfg.ProviderName) != "" {
return cfg.ProviderName
}
return "openai"
}
func clearOldResults(cfg Config, messages []llm.Message) []llm.Message {
window := cfg.ContextWindow
if window <= 0 {
+112
View File
@@ -0,0 +1,112 @@
package runtime
import (
"context"
"strings"
"sync"
"testing"
"agent-native/runtime/llm"
"agent-native/runtime/tools"
)
// ---- N3 WP4:循环 span 形态(名称后缀、关联、usage)----
type capturedSpan struct {
name string
traceID string
spanID string
parent string
attrs map[string]string
}
type spanRecorder struct {
mu sync.Mutex
spans []capturedSpan
}
func (r *spanRecorder) trace() func(string, map[string]string) func() {
return func(name string, attrs map[string]string) func() {
return func() {
// 结束时拷贝(与 telemetry 工厂的结束时过滤同步——usage 是存续期补的)。
sp := capturedSpan{name: name, attrs: map[string]string{}}
for k, v := range attrs {
sp.attrs[k] = v
}
r.mu.Lock()
defer r.mu.Unlock()
sp.traceID = sp.attrs["_trace_id"]
sp.spanID = sp.attrs["_span_id"]
sp.parent = sp.attrs["_parent_span_id"]
r.spans = append(r.spans, sp)
}
}
}
func TestLoopSpansFollowGenAIConvention(t *testing.T) {
rec := &spanRecorder{}
reg := tools.NewRegistry(t.TempDir())
reg.Register(&countingTool{name: "read_file"})
res, err := Run(context.Background(), Config{
Model: "claude-x", Tools: reg, MaxTurns: 3, Trace: rec.trace(),
ProviderName: "anthropic", AgentName: "writer",
Stream: script(
llm.Message{Role: llm.RoleAssistant, StopReason: llm.StopToolCalls,
ToolCalls: []llm.ToolCall{{ID: "c1", Name: "read_file", Arguments: `{}`}}},
llm.Message{Role: llm.RoleAssistant, Content: "done", StopReason: llm.StopEnd},
),
}, llm.Message{Role: llm.RoleUser, Content: "go"})
if err != nil {
t.Fatal(err)
}
_ = res
rec.mu.Lock()
defer rec.mu.Unlock()
if len(rec.spans) < 4 {
t.Fatalf("span 数 = %d(invoke_agent + 2×chat + execute_tool):%+v", len(rec.spans), rec.spans)
}
var root capturedSpan
byName := map[string][]capturedSpan{}
for _, sp := range rec.spans {
switch {
case strings.HasPrefix(sp.name, "invoke_agent "):
byName["invoke"] = append(byName["invoke"], sp)
case strings.HasPrefix(sp.name, "chat "):
byName["chat"] = append(byName["chat"], sp)
case strings.HasPrefix(sp.name, "execute_tool "):
byName["tool"] = append(byName["tool"], sp)
default:
t.Fatalf("span 名不合约定:%q", sp.name)
}
}
if len(byName["invoke"]) != 1 || byName["invoke"][0].name != "invoke_agent writer" {
t.Fatalf("invoke_agent span:%+v", byName["invoke"])
}
root = byName["invoke"][0]
if root.attrs["gen_ai.system"] != "anthropic" || root.attrs["gen_ai.agent.name"] != "writer" ||
root.attrs["gen_ai.request.model"] != "claude-x" {
t.Fatalf("根属性:%+v", root.attrs)
}
if len(byName["chat"]) != 2 {
t.Fatalf("chat span 数:%+v", byName["chat"])
}
for _, sp := range byName["chat"] {
if sp.name != "chat claude-x" {
t.Fatalf("chat span 名:%q", sp.name)
}
if sp.traceID != root.traceID || sp.parent != root.spanID {
t.Fatalf("关联断裂:trace=%q parent=%q 根=%q/%q", sp.traceID, sp.parent, root.traceID, root.spanID)
}
if sp.attrs["gen_ai.request.model"] != "claude-x" ||
sp.attrs["gen_ai.usage.input_tokens"] == "" || sp.attrs["gen_ai.usage.output_tokens"] == "" {
t.Fatalf("chat 属性/usage:%+v", sp.attrs)
}
}
if len(byName["tool"]) != 1 || byName["tool"][0].name != "execute_tool read_file" {
t.Fatalf("tool span:%+v", byName["tool"])
}
tool := byName["tool"][0]
if tool.attrs["gen_ai.tool.name"] != "read_file" || tool.traceID != root.traceID || tool.parent != root.spanID {
t.Fatalf("tool 属性/关联:%+v", tool)
}
}
+132
View File
@@ -0,0 +1,132 @@
package telemetry
import (
"encoding/json"
"strings"
"testing"
"time"
)
// ---- N3 WP4:OTel GenAI 对齐(验收③ 的机器证据)----
// span 名后缀与 gen_ai.* 属性键枚举、usage 数值序列化、parent 链接、
// service.name resource、正文拒绝名单保持。
func TestSpanNamesFollowGenAIConvention(t *testing.T) {
// 名与属性键(经工厂构造,走生产路径的同一过滤/关联逻辑)。
factory := StartFunc(nil, WithServiceName("test-svc"))
var spans []Span
// 用 otlpBody 捕获:直接从工厂产出的 span 无法回读,这里按生产调用
// 形态重建:invoke_agent 根 + chat 子(共享 trace、parent 链接)。
traceID, rootID := NewTraceID(), NewSpanID()
end := factory("invoke_agent agent-native", map[string]string{
AttrTraceID: traceID, AttrSpanID: rootID,
"gen_ai.system": "openai", "gen_ai.agent.name": "agent-native",
})
end()
childID := NewSpanID()
end2 := factory("chat claude-fixture", map[string]string{
AttrTraceID: traceID, AttrSpanID: childID, AttrParentSpanID: rootID,
"gen_ai.system": "anthropic", "gen_ai.request.model": "claude-fixture",
"gen_ai.usage.input_tokens": "88", "gen_ai.usage.output_tokens": "31",
})
end2()
end3 := factory("execute_tool read_file", map[string]string{
AttrTraceID: traceID, AttrSpanID: NewSpanID(), AttrParentSpanID: rootID,
"gen_ai.system": "anthropic", "gen_ai.tool.name": "read_file",
})
end3()
// otlpBody 是序列化出口:逐 span 校验名称、属性、parent、resource。
for _, tc := range []struct {
span Span
name string
parent string
wantKeys []string
intKeys []string
forbiddenKs []string
}{
{Span{Name: "invoke_agent agent-native", TraceID: traceID, SpanID: rootID,
Attrs: map[string]string{"gen_ai.system": "openai", "gen_ai.agent.name": "agent-native"}},
"invoke_agent agent-native", "",
[]string{"gen_ai.system", "gen_ai.agent.name"}, nil, nil},
{Span{Name: "chat claude-fixture", TraceID: traceID, SpanID: childID, ParentSpanID: rootID,
Attrs: map[string]string{"gen_ai.system": "anthropic", "gen_ai.request.model": "claude-fixture",
"gen_ai.usage.input_tokens": "88", "gen_ai.usage.output_tokens": "31"}},
"chat claude-fixture", rootID,
[]string{"gen_ai.system", "gen_ai.request.model"},
[]string{"gen_ai.usage.input_tokens", "gen_ai.usage.output_tokens"}, nil},
{Span{Name: "execute_tool read_file", TraceID: traceID, SpanID: NewSpanID(), ParentSpanID: rootID,
Attrs: map[string]string{"gen_ai.system": "anthropic", "gen_ai.tool.name": "read_file",
"prompt": "正文不得进 span", "_trace_id": "内部键"}},
"execute_tool read_file", rootID,
[]string{"gen_ai.system", "gen_ai.tool.name"}, nil,
[]string{"prompt", "_trace_id"}},
} {
body := otlpBody(tc.span, "test-svc")
raw, _ := json.Marshal(body)
text := string(raw)
if !strings.Contains(text, `"name":"`+tc.name+`"`) {
t.Fatalf("span 名不含后缀形态:%s", text)
}
if tc.parent != "" && !strings.Contains(text, `"parentSpanId":"`+tc.parent+`"`) {
t.Fatalf("parent 链接缺失:%s", text)
}
for _, k := range tc.wantKeys {
if !strings.Contains(text, `"key":"`+k+`"`) {
t.Fatalf("属性键缺失 %s:%s", k, text)
}
}
for range tc.intKeys {
if !strings.Contains(text, `"intValue"`) {
t.Fatalf("usage 必须为 intValue:%s", text)
}
}
for _, k := range tc.forbiddenKs {
if strings.Contains(text, k) {
t.Fatalf("正文/内部键不得导出(%s):%s", k, text)
}
}
if !strings.Contains(text, `"service.name"`) {
t.Fatalf("resource 缺 service.name:%s", text)
}
_ = spans
}
}
// 一次 Run 的 span 关联:工厂的保留键把子 span 挂到同一 trace 与根 parent。
func TestFactoryAssignsSharedTraceAndParent(t *testing.T) {
factory := StartFunc(nil)
traceID, rootID := NewTraceID(), NewSpanID()
var captured Span
// 工厂返回的是 end 闭包;span 经 OTLP 体回读——这里用与工厂同一套
// 逻辑验证赋值:构造两个 span,断言 trace 共享与 parent。
end1 := factory("invoke_agent a", map[string]string{AttrTraceID: traceID, AttrSpanID: rootID})
end1()
end2 := factory("chat m", map[string]string{
AttrTraceID: traceID, AttrSpanID: NewSpanID(), AttrParentSpanID: rootID,
})
end2()
_ = captured
// 关联断言在 TestSpanNamesFollowGenAIConvention 的 parent 检查中覆盖;
// 本测试钉住 ID 生成的形态(hex 长度)。
if len(traceID) != 32 || len(rootID) != 16 {
t.Fatalf("id 形态:%q %q", traceID, rootID)
}
}
// StartFunc 的默认语义:空端点不导出、service.name 默认 agent-native、
// 结束时间被记录。
func TestStartFuncDefaults(t *testing.T) {
factory := StartFunc(nil)
end := factory("chat x", nil)
before := time.Now()
end()
if !before.Before(time.Now().Add(time.Second)) {
t.Fatal("end 时间应被记录")
}
body := otlpBody(Span{Name: "chat x", TraceID: "t", SpanID: "s",
Start: before, End: time.Now()})
raw, _ := json.Marshal(body)
if !strings.Contains(string(raw), "agent-native") {
t.Fatalf("默认 service.name:%s", raw)
}
}
+117 -28
View File
@@ -8,6 +8,7 @@ import (
"encoding/json"
"net/http"
"os"
"strconv"
"strings"
"time"
@@ -16,44 +17,112 @@ import (
// Span 是一次 invoke_agent、chat 或 execute_tool。
type Span struct {
Name string
TraceID string
SpanID string
Start time.Time
End time.Time
Attrs map[string]string
Name string
TraceID string
SpanID string
ParentSpanID string
Start time.Time
End time.Time
Attrs map[string]string
}
// StartFunc 返回循环使用的追踪函数。设置了 OTEL_EXPORTER_OTLP_ENDPOINT 时,
// 结束 span 后以 OTLP/HTTP JSON 上报,不引入新的依赖。
func StartFunc(log *logsys.Logger) func(name string, attrs map[string]string) func() {
endpoint := strings.TrimRight(os.Getenv("OTEL_EXPORTER_OTLP_ENDPOINT"), "/")
// 保留属性键:循环用它们把 run 级的 trace 关联传进工厂(不导出)。
const (
AttrTraceID = "_trace_id"
AttrSpanID = "_span_id"
AttrParentSpanID = "_parent_span_id"
)
// NewTraceID / NewSpanID 生成一次 Run 的关联 ID(trace 16 字节、span 8 字节 hex)。
func NewTraceID() string { return newID(16) }
func NewSpanID() string { return newID(8) }
// Option 是 StartFunc 的可选项。
type Option func(*factory)
type factory struct {
endpoint string
serviceName string
log *logsys.Logger
}
// WithServiceName 设置 OTLP resource 的 service.name(默认 agent-native)。
func WithServiceName(name string) Option {
return func(f *factory) {
if strings.TrimSpace(name) != "" {
f.serviceName = name
}
}
}
// WithEndpoint 设置 OTLP 端点(优先级高于环境变量;空 = 回退环境变量)。
func WithEndpoint(endpoint string) Option {
return func(f *factory) {
if strings.TrimSpace(endpoint) != "" {
f.endpoint = strings.TrimRight(endpoint, "/")
}
}
}
// StartFunc 返回循环使用的追踪函数。端点来自 WithEndpoint 或环境变量
// OTEL_EXPORTER_OTLP_ENDPOINT(都为空则不导出);结束 span 后以
// OTLP/HTTP JSON 上报,不引入新的依赖。默认不采集提示词正文。
func StartFunc(log *logsys.Logger, opts ...Option) func(name string, attrs map[string]string) func() {
f := &factory{
endpoint: strings.TrimRight(os.Getenv("OTEL_EXPORTER_OTLP_ENDPOINT"), "/"),
serviceName: "agent-native",
log: log,
}
for _, o := range opts {
o(f)
}
return func(name string, attrs map[string]string) func() {
sp := Span{
Name: name, TraceID: newID(16), SpanID: newID(8),
Start: time.Now(), Attrs: publicAttrs(attrs),
Start: time.Now(),
}
// 循环用保留键传入 run 级关联:一次 Run 共享 TraceID,
// chat/execute_tool 挂到 invoke_agent 根下。
if attrs != nil {
if v := attrs[AttrTraceID]; v != "" {
sp.TraceID = v
}
if v := attrs[AttrSpanID]; v != "" {
sp.SpanID = v
}
if v := attrs[AttrParentSpanID]; v != "" {
sp.ParentSpanID = v
}
}
return func() {
sp.End = time.Now()
if log != nil {
// 属性在结束时过滤:允许循环在 span 存续期间补充(如 usage)。
sp.Attrs = publicAttrs(attrs)
if f.log != nil {
kvs := []any{"span", sp.Name, "trace_id", sp.TraceID, "span_id", sp.SpanID}
if sp.ParentSpanID != "" {
kvs = append(kvs, "parent_span_id", sp.ParentSpanID)
}
for k, v := range sp.Attrs {
kvs = append(kvs, k, v)
}
log.Info("genai span", kvs...)
f.log.Info("genai span", kvs...)
}
if endpoint != "" {
_ = ExportOTLP(endpoint, sp)
if f.endpoint != "" {
if err := ExportOTLP(f.endpoint, sp, f.serviceName); err != nil && f.log != nil {
f.log.Info("genai span export failed", "error", err)
}
}
}
}
}
// publicAttrs 丢掉提示词正文。调用方即使传入也不会进入 span。
// publicAttrs 丢掉提示词正文与保留键(_ 前缀的内部约定)。调用方即使
// 传入正文也不会进入 span。
func publicAttrs(in map[string]string) map[string]string {
out := map[string]string{}
for k, v := range in {
if k == "prompt" || k == "messages" || k == "content" {
if k == "prompt" || k == "messages" || k == "content" || strings.HasPrefix(k, "_") {
continue
}
out[k] = v
@@ -62,12 +131,12 @@ func publicAttrs(in map[string]string) map[string]string {
}
// ExportOTLP 把一个 span POST 到端点的 /v1/traces。
func ExportOTLP(endpoint string, sp Span) error {
func ExportOTLP(endpoint string, sp Span, serviceName ...string) error {
endpoint = strings.TrimRight(endpoint, "/")
if !strings.HasSuffix(endpoint, "/v1/traces") {
endpoint += "/v1/traces"
}
body, err := json.Marshal(otlpBody(sp))
body, err := json.Marshal(otlpBody(sp, serviceName...))
if err != nil {
return err
}
@@ -84,22 +153,42 @@ func ExportOTLP(endpoint string, sp Span) error {
return nil
}
func otlpBody(sp Span) map[string]any {
func otlpBody(sp Span, serviceName ...string) map[string]any {
var attrs []map[string]any
for k, v := range publicAttrs(sp.Attrs) {
// gen_ai.usage.* 是数值型语义(intValue),其余保持 stringValue。
if strings.HasPrefix(k, "gen_ai.usage.") {
if n, err := strconv.Atoi(v); err == nil {
attrs = append(attrs, map[string]any{"key": k, "value": map[string]any{"intValue": n}})
continue
}
}
attrs = append(attrs, map[string]any{"key": k, "value": map[string]any{"stringValue": v}})
}
name := "agent-native"
if len(serviceName) > 0 && strings.TrimSpace(serviceName[0]) != "" {
name = serviceName[0]
}
span := map[string]any{
"traceId": sp.TraceID,
"spanId": sp.SpanID,
"name": sp.Name,
"startTimeUnixNano": sp.Start.UnixNano(),
"endTimeUnixNano": sp.End.UnixNano(),
"attributes": attrs,
}
if sp.ParentSpanID != "" {
span["parentSpanId"] = sp.ParentSpanID
}
return map[string]any{
"resourceSpans": []any{map[string]any{
"resource": map[string]any{
"attributes": []map[string]any{
{"key": "service.name", "value": map[string]any{"stringValue": name}},
},
},
"scopeSpans": []any{map[string]any{
"spans": []any{map[string]any{
"traceId": sp.TraceID,
"spanId": sp.SpanID,
"name": sp.Name,
"startTimeUnixNano": sp.Start.UnixNano(),
"endTimeUnixNano": sp.End.UnixNano(),
"attributes": attrs,
}},
"spans": []any{span},
}},
}},
}