/agent/otel.go

package agent

import (
	"context"
	"encoding/json"
	"fmt"
	"sort"
	"strings"
	"time"

	"go-micro.dev/v6/ai"
	"go-micro.dev/v6/store"
	"go.opentelemetry.io/otel/attribute"
	"go.opentelemetry.io/otel/codes"
	"go.opentelemetry.io/otel/trace"
)

const agentInstrumentationName = "go-micro.dev/v6/agent"

const (
	spanNameRun       = "agent.run"
	spanNameModelCall = "agent.model.call"
	spanNameToolCall  = "agent.tool.call"

	AttrRunID          = "agent.run.id"
	AttrParentRunID    = "agent.run.parent_id"
	AttrAgentName      = "agent.name"
	AttrProvider       = "agent.model.provider"
	AttrModel          = "agent.model.name"
	AttrLatencyMS      = "agent.latency_ms"
	AttrInputTokens    = "agent.tokens.input"
	AttrOutputTokens   = "agent.tokens.output"
	AttrTotalTokens    = "agent.tokens.total"
	AttrToolName       = "agent.tool.name"
	AttrDelegate       = "agent.delegate"
	AttrGuardrailBlock = "agent.guardrail.block"
	AttrRefusal        = "agent.refusal"
)

type RunEvent struct {
	Time      time.Time `json:"time"`
	RunID     string    `json:"run_id"`
	ParentID  string    `json:"parent_id,omitempty"`
	TraceID   string    `json:"trace_id,omitempty"`
	SpanID    string    `json:"span_id,omitempty"`
	Agent     string    `json:"agent"`
	Kind      string    `json:"kind"`
	Name      string    `json:"name,omitempty"`
	Provider  string    `json:"provider,omitempty"`
	Model     string    `json:"model,omitempty"`
	LatencyMS int64     `json:"latency_ms,omitempty"`
	Tokens    Usage     `json:"tokens,omitempty"`
	Refused   string    `json:"refused,omitempty"`
	Error     string    `json:"error,omitempty"`
}

type Usage = ai.Usage

// RunListOptions controls how recorded agent run summaries are returned.
// Zero values preserve the full deterministic run list.
type RunListOptions struct {
	// Status, when set, keeps only runs with the matching status
	// (for example "running", "done", "error", or "refused").
	Status string
	// TraceID, when set, keeps only runs correlated with this trace id.
	// A prefix is accepted so operators can paste the shortened trace id
	// printed by `micro runs`.
	TraceID string
	// Limit, when positive, returns the most recently updated runs up to
	// the limit. Limited results are ordered newest first.
	Limit int
}

// RunSummary is a compact index entry for a recorded agent run.
type RunSummary struct {
	RunID      string    `json:"run_id"`
	Agent      string    `json:"agent"`
	ParentID   string    `json:"parent_id,omitempty"`
	TraceID    string    `json:"trace_id,omitempty"`
	SpanID     string    `json:"span_id,omitempty"`
	StartedAt  time.Time `json:"started_at"`
	UpdatedAt  time.Time `json:"updated_at"`
	DurationMS int64     `json:"duration_ms,omitempty"`
	Events     int       `json:"events"`
	Status     string    `json:"status,omitempty"`
	LastKind   string    `json:"last_kind,omitempty"`
	LastError  string    `json:"last_error,omitempty"`
}

func (a *agentImpl) tracer() trace.Tracer {
	return a.opts.TraceProvider.Tracer(agentInstrumentationName)
}

func (a *agentImpl) startRun(ctx context.Context, message string) (context.Context, func(error)) {
	info, _ := ai.RunInfoFrom(ctx)
	start := time.Now()

	if a.opts.TraceProvider == nil {
		a.recordRunEvent(RunEvent{Time: start, RunID: info.RunID, ParentID: info.ParentID, Agent: info.Agent, Kind: "run", Name: message})
		return ctx, func(err error) {
			latency := time.Since(start).Milliseconds()
			if err != nil {
				a.recordRunEvent(RunEvent{Time: time.Now(), RunID: info.RunID, ParentID: info.ParentID, Agent: info.Agent, Kind: "error", LatencyMS: latency, Error: err.Error()})
				return
			}
			a.recordRunEvent(RunEvent{Time: time.Now(), RunID: info.RunID, ParentID: info.ParentID, Agent: info.Agent, Kind: "done", LatencyMS: latency})
		}
	}

	ctx, span := a.tracer().Start(ctx, spanNameRun, trace.WithSpanKind(trace.SpanKindInternal), trace.WithAttributes(
		attribute.String(AttrRunID, info.RunID), attribute.String(AttrParentRunID, info.ParentID), attribute.String(AttrAgentName, info.Agent)))
	a.recordSpanEvent(span, RunEvent{Time: start, RunID: info.RunID, ParentID: info.ParentID, Agent: info.Agent, Kind: "run", Name: message})
	return ctx, func(err error) {
		latency := time.Since(start).Milliseconds()
		span.SetAttributes(attribute.Int64(AttrLatencyMS, latency))
		if err != nil {
			span.RecordError(err)
			span.SetStatus(codes.Error, err.Error())
			a.recordSpanEvent(span, RunEvent{Time: time.Now(), RunID: info.RunID, ParentID: info.ParentID, Agent: info.Agent, Kind: "error", LatencyMS: latency, Error: err.Error()})
		} else {
			span.SetStatus(codes.Ok, "")
			a.recordSpanEvent(span, RunEvent{Time: time.Now(), RunID: info.RunID, ParentID: info.ParentID, Agent: info.Agent, Kind: "done", LatencyMS: latency})
		}
		span.End()
	}
}

type tracedModel struct {
	ai.Model
	a *agentImpl
}

func (a *agentImpl) tracedModel(m ai.Model) ai.Model { return &tracedModel{Model: m, a: a} }
func (m *tracedModel) Generate(ctx context.Context, req *ai.Request, opts ...ai.GenerateOption) (*ai.Response, error) {
	info, _ := ai.RunInfoFrom(ctx)
	provider := m.String()
	model := m.Options().Model
	start := time.Now()

	if m.a.opts.TraceProvider == nil {
		resp, err := m.Model.Generate(ctx, req, opts...)
		dur := time.Since(start).Milliseconds()
		usage := ai.Usage{}
		if resp != nil {
			usage = resp.Usage
		}
		e := RunEvent{Time: time.Now(), RunID: info.RunID, ParentID: info.ParentID, Agent: info.Agent, Kind: "model", Provider: provider, Model: model, LatencyMS: dur, Tokens: usage}
		if err != nil {
			e.Error = err.Error()
		}
		m.a.recordRunEvent(e)
		return resp, err
	}

	ctx, span := m.a.tracer().Start(ctx, spanNameModelCall, trace.WithAttributes(
		attribute.String(AttrRunID, info.RunID),
		attribute.String(AttrParentRunID, info.ParentID),
		attribute.String(AttrAgentName, info.Agent),
		attribute.String(AttrProvider, provider),
		attribute.String(AttrModel, model),
	))
	resp, err := m.Model.Generate(ctx, req, opts...)
	dur := time.Since(start).Milliseconds()
	attrs := []attribute.KeyValue{attribute.Int64(AttrLatencyMS, dur)}
	usage := ai.Usage{}
	if resp != nil {
		usage = resp.Usage
		attrs = appendUsage(attrs, usage)
	}
	span.SetAttributes(attrs...)
	if err != nil {
		span.RecordError(err)
		span.SetStatus(codes.Error, err.Error())
	} else {
		span.SetStatus(codes.Ok, "")
	}
	span.End()
	e := RunEvent{Time: time.Now(), RunID: info.RunID, ParentID: info.ParentID, Agent: info.Agent, Kind: "model", Provider: provider, Model: model, LatencyMS: dur, Tokens: usage}
	if err != nil {
		e.Error = err.Error()
	}
	m.a.recordSpanEvent(span, e)
	return resp, err
}

func appendUsage(attrs []attribute.KeyValue, u ai.Usage) []attribute.KeyValue {
	if u.InputTokens > 0 {
		attrs = append(attrs, attribute.Int(AttrInputTokens, u.InputTokens))
	}
	if u.OutputTokens > 0 {
		attrs = append(attrs, attribute.Int(AttrOutputTokens, u.OutputTokens))
	}
	if u.TotalTokens > 0 {
		attrs = append(attrs, attribute.Int(AttrTotalTokens, u.TotalTokens))
	}
	return attrs
}

func (a *agentImpl) traceTool(next ai.ToolHandler) ai.ToolHandler {
	return func(ctx context.Context, call ai.ToolCall) ai.ToolResult {
		info, _ := ai.RunInfoFrom(ctx)
		start := time.Now()

		if a.opts.TraceProvider == nil {
			res := next(ctx, call)
			dur := time.Since(start).Milliseconds()
			a.recordRunEvent(RunEvent{Time: time.Now(), RunID: info.RunID, ParentID: info.ParentID, Agent: info.Agent, Kind: "tool", Name: call.Name, LatencyMS: dur, Refused: res.Refused, Error: resultError(res)})
			return res
		}

		ctx, span := a.tracer().Start(ctx, spanNameToolCall, trace.WithAttributes(
			attribute.String(AttrRunID, info.RunID),
			attribute.String(AttrParentRunID, info.ParentID),
			attribute.String(AttrAgentName, info.Agent),
			attribute.String(AttrToolName, call.Name),
			attribute.Bool(AttrDelegate, call.Name == toolDelegate),
		))
		res := next(ctx, call)
		dur := time.Since(start).Milliseconds()
		attrs := []attribute.KeyValue{attribute.Int64(AttrLatencyMS, dur)}
		if res.Refused != "" {
			attrs = append(attrs, attribute.Bool(AttrGuardrailBlock, true), attribute.String(AttrRefusal, res.Refused))
		}
		span.SetAttributes(attrs...)
		resErr := resultError(res)
		if res.Refused != "" {
			span.SetStatus(codes.Error, res.Refused)
		} else if resErr != "" {
			span.SetStatus(codes.Error, resErr)
		} else {
			span.SetStatus(codes.Ok, "")
		}
		span.End()
		a.recordSpanEvent(span, RunEvent{Time: time.Now(), RunID: info.RunID, ParentID: info.ParentID, Agent: info.Agent, Kind: "tool", Name: call.Name, LatencyMS: dur, Refused: res.Refused, Error: resErr})
		return res
	}
}

func resultError(res ai.ToolResult) string {
	if m, ok := res.Value.(map[string]string); ok {
		return m["error"]
	}
	if m, ok := res.Value.(map[string]any); ok {
		if err, _ := m["error"].(string); err != "" {
			return err
		}
	}
	return ""
}

func (a *agentImpl) recordSpanEvent(span trace.Span, e RunEvent) {
	if sc := span.SpanContext(); sc.IsValid() {
		e.TraceID = sc.TraceID().String()
		e.SpanID = sc.SpanID().String()
	}
	a.recordRunEvent(e)
}

func (a *agentImpl) recordRunEvent(e RunEvent) {
	if e.RunID == "" {
		return
	}
	b, _ := json.Marshal(e)
	key := fmt.Sprintf("runs/%s/%020d-%s", e.RunID, e.Time.UnixNano(), e.Kind)
	_ = a.stateStore().Write(&store.Record{Key: key, Value: b})
}

// ListRunSummaries returns a deterministic summary of recorded runs for agentName.
func ListRunSummaries(s store.Store, agentName string) ([]RunSummary, error) {
	return ListRunSummariesWithOptions(s, agentName, RunListOptions{})
}

// ListRunSummariesWithOptions returns summaries of recorded runs for agentName,
// optionally filtered by status and limited to the most recently updated runs.
func ListRunSummariesWithOptions(s store.Store, agentName string, opts RunListOptions) ([]RunSummary, error) {
	st := store.Scope(s, "agent", agentName)
	keys, err := st.List(store.ListPrefix("runs/"))
	if err != nil {
		return nil, err
	}
	runs := map[string]bool{}
	for _, k := range keys {
		parts := strings.Split(k, "/")
		if len(parts) >= 2 && parts[1] != "" {
			runs[parts[1]] = true
		}
	}
	ids := make([]string, 0, len(runs))
	for id := range runs {
		ids = append(ids, id)
	}
	sort.Strings(ids)

	summaries := make([]RunSummary, 0, len(ids))
	for _, id := range ids {
		events, err := LoadRunEvents(s, agentName, id)
		if err != nil {
			return nil, err
		}
		if len(events) == 0 {
			continue
		}
		first := events[0]
		last := events[len(events)-1]
		summary := RunSummary{
			RunID:      id,
			Agent:      first.Agent,
			ParentID:   first.ParentID,
			TraceID:    first.TraceID,
			SpanID:     first.SpanID,
			StartedAt:  first.Time,
			UpdatedAt:  last.Time,
			DurationMS: last.Time.Sub(first.Time).Milliseconds(),
			Events:     len(events),
			Status:     runStatus(events),
			LastKind:   last.Kind,
			LastError:  last.Error,
		}
		for _, e := range events {
			if e.Agent != "" {
				summary.Agent = e.Agent
			}
			if e.ParentID != "" {
				summary.ParentID = e.ParentID
			}
			if e.TraceID != "" {
				summary.TraceID = e.TraceID
			}
			if e.SpanID != "" {
				summary.SpanID = e.SpanID
			}
			if e.Error != "" {
				summary.LastError = e.Error
			}
		}
		if opts.Status != "" && summary.Status != opts.Status {
			continue
		}
		if opts.TraceID != "" && !strings.HasPrefix(summary.TraceID, opts.TraceID) {
			continue
		}
		summaries = append(summaries, summary)
	}
	if opts.Limit > 0 {
		sort.SliceStable(summaries, func(i, j int) bool {
			return summaries[i].UpdatedAt.After(summaries[j].UpdatedAt)
		})
		if len(summaries) > opts.Limit {
			summaries = summaries[:opts.Limit]
		}
	}
	return summaries, nil
}

func runStatus(events []RunEvent) string {
	if len(events) == 0 {
		return ""
	}
	status := "running"
	for _, e := range events {
		if e.Error != "" {
			status = "error"
		}
		if e.Refused != "" && status != "error" {
			status = "refused"
		}
		switch e.Kind {
		case "error":
			status = "error"
		case "done":
			if status == "running" {
				status = "done"
			}
		}
	}
	return status
}

func LoadRunEvents(s store.Store, agentName, runID string) ([]RunEvent, error) {
	st := store.Scope(s, "agent", agentName)
	keys, err := st.List(store.ListPrefix("runs/" + runID + "/"))
	if err != nil {
		return nil, err
	}
	sort.Strings(keys)
	events := make([]RunEvent, 0, len(keys))
	for _, k := range keys {
		recs, err := st.Read(k)
		if err != nil || len(recs) == 0 {
			continue
		}
		var e RunEvent
		if json.Unmarshal(recs[0].Value, &e) == nil {
			events = append(events, e)
		}
	}
	return events, nil
}