Add configurable provider conformance dispatch

agent/a2a_stream_test.go

@@ -1,102 +0,0 @@
-package agent
-
-import (
-	"bytes"
-	"context"
-	"encoding/json"
-	"net/http"
-	"net/http/httptest"
-	"strings"
-	"testing"
-
-	"go-micro.dev/v6/ai"
-	"go-micro.dev/v6/gateway/a2a"
-)
-
-func TestA2AStreamUsesAgentChatPathWithTools(t *testing.T) {
-	var sawTool bool
-	fakeGen = func(ctx context.Context, opts ai.Options, req *ai.Request) (*ai.Response, error) {
-		if opts.ToolHandler == nil {
-			t.Fatal("model was not wired with agent tool handler")
-		}
-		result := opts.ToolHandler(ctx, ai.ToolCall{
-			ID:    "call-1",
-			Name:  "echo",
-			Input: map[string]any{"value": "a2a-stream"},
-		})
-		if !strings.Contains(result.Content, "a2a-stream-ok") {
-			t.Fatalf("tool result = %q, want marker", result.Content)
-		}
-		return &ai.Response{Answer: "streamed " + result.Content}, nil
-	}
-	defer func() { fakeGen = nil }()
-
-	a := newTestAgent(Name("stream-agent"), WithTool("echo", "echo text", nil, func(ctx context.Context, input map[string]any) (string, error) {
-		sawTool = true
-		if info, ok := ai.RunInfoFrom(ctx); !ok || info.RunID == "" || info.Agent != "stream-agent" {
-			t.Fatalf("RunInfo = %+v ok=%v, want stream-agent run", info, ok)
-		}
-		if input["value"] != "a2a-stream" {
-			t.Fatalf("tool input = %+v, want a2a-stream", input)
-		}
-		return "a2a-stream-ok", nil
-	}))
-	h := a2a.NewAgentStreamHandler(
-		a2a.Card("stream-agent", "http://example.invalid/stream-agent", "", nil),
-		func(ctx context.Context, text string) (string, error) {
-			resp, err := a.Ask(ctx, text)
-			if err != nil {
-				return "", err
-			}
-			return resp.Reply, nil
-		},
-		a.streamAskAI,
-	)
-
-	body := []byte(`{"jsonrpc":"2.0","id":1,"method":"message/stream","params":{"message":{"role":"user","parts":[{"kind":"text","text":"run stream tool"}],"kind":"message"}}}`)
-	req := httptest.NewRequest(http.MethodPost, "/", bytes.NewReader(body))
-	rr := httptest.NewRecorder()
-	h.ServeHTTP(rr, req)
-
-	if !sawTool {
-		t.Fatal("A2A stream did not execute the agent tool path")
-	}
-	if ct := rr.Result().Header.Get("Content-Type"); !strings.HasPrefix(ct, "text/event-stream") {
-		t.Fatalf("content-type = %q, want text/event-stream", ct)
-	}
-	if !strings.Contains(rr.Body.String(), "a2a-stream-ok") {
-		t.Fatalf("stream body missing tool marker: %s", rr.Body.String())
-	}
-
-	var final struct {
-		Result struct {
-			Status struct {
-				State string `json:"state"`
-			} `json:"status"`
-			Artifacts []struct {
-				Parts []struct {
-					Text string `json:"text"`
-				} `json:"parts"`
-			} `json:"artifacts"`
-		} `json:"result"`
-		Error any `json:"error"`
-	}
-	for _, line := range strings.Split(strings.TrimSpace(rr.Body.String()), "\n") {
-		line = strings.TrimSpace(strings.TrimPrefix(strings.TrimSpace(line), "data: "))
-		if line == "" {
-			continue
-		}
-		if err := json.Unmarshal([]byte(line), &final); err != nil {
-			t.Fatalf("decode event %q: %v", line, err)
-		}
-	}
-	if final.Error != nil {
-		t.Fatalf("final event error: %+v", final.Error)
-	}
-	if final.Result.Status.State != "completed" {
-		t.Fatalf("final state = %q, want completed", final.Result.Status.State)
-	}
-	if len(final.Result.Artifacts) != 1 || len(final.Result.Artifacts[0].Parts) != 1 || !strings.Contains(final.Result.Artifacts[0].Parts[0].Text, "a2a-stream-ok") {
-		t.Fatalf("final artifacts = %+v, want tool marker", final.Result.Artifacts)
-	}
-}

agent/agent.go

@@ -19,7 +19,6 @@ import (
 	"net/http"
 	"strings"
 	"sync"
-	"time"
 
 	"github.com/google/uuid"
 	pb "go-micro.dev/v6/agent/proto"
@@ -178,7 +177,7 @@ func (a *agentImpl) setupWithToolHandler(handler ai.ToolHandler) {
 	case a.ephemeral:
 		a.mem = NewInMemory(a.opts.HistoryLimit)
 	case a.opts.MemoryCompaction.MaxMessages > 0:
-		a.mem = NewCompactingMemoryWithOptions(a.stateStore(), "history", a.opts.MemoryCompaction)
+		a.mem = NewCompactingMemory(a.stateStore(), "history", a.opts.MemoryCompaction.MaxMessages, a.opts.MemoryCompaction.KeepRecent)
 	default:
 		a.mem = NewMemory(a.stateStore(), "history", a.opts.HistoryLimit)
 	}
@@ -272,9 +271,6 @@ func (a *agentImpl) askLocked(ctx context.Context, runID, message, parentRunID s
 		return nil, err
 	}
 	ctx, endRun := a.startRun(ctx, message)
-	if existing != nil {
-		a.recordTimelineEvent(ctx, RunEvent{Time: time.Now(), RunID: runID, ParentID: parentRunID, Agent: a.opts.Name, Kind: "resume", Name: run.State.Stage})
-	}
 	defer func() { endRun(err) }()
 
 	messages := a.mem.Messages()
@@ -423,7 +419,7 @@ func (a *agentImpl) Run() error {
 				return "", err
 			}
 			return resp.Reply, nil
-		}, a.streamAskAI)
+		}, a.Stream)
 		go func() {
 			if err := http.ListenAndServe(a.opts.A2AAddress, handler); err != nil {
 				fmt.Printf("agent %s A2A server: %v\n", a.opts.Name, err)

agent/checkpoint.go

@@ -49,12 +49,6 @@ func (a *agentImpl) saveRun(ctx context.Context, run flow.Run) error {
 	if err := a.opts.Checkpoint.Save(ctx, run); err != nil {
 		return fmt.Errorf("agent %s checkpoint save: %w", a.opts.Name, err)
 	}
-	if info, ok := ai.RunInfoFrom(ctx); ok {
-		a.recordTimelineEvent(ctx, RunEvent{
-			Time: time.Now(), RunID: info.RunID, ParentID: info.ParentID, Agent: info.Agent,
-			Kind: "checkpoint", Name: run.State.Stage, Status: run.Status,
-		})
-	}
 	return nil
 }
 

agent/memory.go

@@ -24,12 +24,6 @@ type Memory interface {
 	Clear()
 }
 
-// MemorySummaryFunc turns older conversation messages into a compact
-// replacement message for active context. It is called while the default
-// memory is locked, so implementations should be deterministic and avoid
-// calling back into the same memory instance.
-type MemorySummaryFunc func([]ai.Message) ai.Message
-
 // MemoryCompaction configures deterministic, store-backed context compaction
 // for the default memory implementation. When the retained conversation grows
 // past MaxMessages, older turns are collapsed into a summary message while the
@@ -37,7 +31,6 @@ type MemorySummaryFunc func([]ai.Message) ai.Message
 type MemoryCompaction struct {
 	MaxMessages int
 	KeepRecent  int
-	Summarize   MemorySummaryFunc
 }
 
 // MemoryRecall is implemented by memory backends that can retrieve durable
@@ -61,14 +54,6 @@ func NewMemory(s store.Store, key string, limit int) Memory {
 // turns into a deterministic summary when the conversation exceeds maxMessages,
 // and lets callers recall relevant prior turns with Recall.
 func NewCompactingMemory(s store.Store, key string, maxMessages, keepRecent int) Memory {
-	return NewCompactingMemoryWithOptions(s, key, MemoryCompaction{MaxMessages: maxMessages, KeepRecent: keepRecent})
-}
-
-// NewCompactingMemoryWithOptions returns store-backed memory configured with
-// explicit compaction options, including an optional summarization hook.
-func NewCompactingMemoryWithOptions(s store.Store, key string, compaction MemoryCompaction) Memory {
-	maxMessages := compaction.MaxMessages
-	keepRecent := compaction.KeepRecent
 	if keepRecent <= 0 {
 		keepRecent = maxMessages / 2
 	}
@@ -84,7 +69,6 @@ func NewCompactingMemoryWithOptions(s store.Store, key string, compaction Memory
 		compaction: MemoryCompaction{
 			MaxMessages: maxMessages,
 			KeepRecent:  keepRecent,
-			Summarize:   compaction.Summarize,
 		},
 	}
 	m.load()
@@ -229,13 +213,9 @@ func (m *storeMemory) compact() {
 	older := msgs[:cut]
 	recent := msgs[cut:]
 	m.archive = append(m.archive, older...)
-	summarize := m.compaction.Summarize
-	if summarize == nil {
-		summarize = defaultMemorySummary
-	}
-	summary := summarize(older)
-	if summary.Role == "" {
-		summary.Role = "system"
+	summary := ai.Message{
+		Role:    "system",
+		Content: fmt.Sprintf("Conversation memory summary: %s", summarizeMessages(older)),
 	}
 	m.hist.Reset()
 	m.hist.Add(summary.Role, summary.Content)
@@ -244,13 +224,6 @@ func (m *storeMemory) compact() {
 	}
 }
 
-func defaultMemorySummary(msgs []ai.Message) ai.Message {
-	return ai.Message{
-		Role:    "system",
-		Content: fmt.Sprintf("Conversation memory summary: %s", summarizeMessages(msgs)),
-	}
-}
-
 func summarizeMessages(msgs []ai.Message) string {
 	var b strings.Builder
 	for i, msg := range msgs {

agent/memory_test.go

@@ -3,11 +3,9 @@ package agent
 import (
 	"context"
 	"errors"
-	"strconv"
 	"strings"
 	"testing"
 
-	"go-micro.dev/v6/ai"
 	"go-micro.dev/v6/registry"
 	"go-micro.dev/v6/store"
 )
@@ -104,36 +102,6 @@ func TestCompactingMemoryArchivePersistsAndReloads(t *testing.T) {
 	}
 }
 
-func TestCompactingMemoryUsesCustomSummarizerAndReloadsRecall(t *testing.T) {
-	st := store.NewMemoryStore()
-	m := NewCompactingMemoryWithOptions(st, "agent/custom/history", MemoryCompaction{
-		MaxMessages: 3,
-		KeepRecent:  1,
-		Summarize: func(msgs []ai.Message) ai.Message {
-			return ai.Message{Role: "system", Content: "custom summary count=" + strconv.Itoa(len(msgs))}
-		},
-	})
-	m.Add("user", "alpha budget is 42")
-	m.Add("assistant", "noted")
-	m.Add("user", "beta budget is 7")
-	m.Add("assistant", "noted")
-
-	msgs := m.Messages()
-	if len(msgs) == 0 || msgs[0].Content != "custom summary count=3" {
-		t.Fatalf("summary = %#v, want custom summarizer output", msgs)
-	}
-
-	reloaded := NewCompactingMemoryWithOptions(st, "agent/custom/history", MemoryCompaction{MaxMessages: 3, KeepRecent: 1})
-	recall := reloaded.(MemoryRecall)
-	recalled := recall.Recall("alpha budget", 1)
-	if len(recalled) != 1 {
-		t.Fatalf("recalled %d messages, want 1", len(recalled))
-	}
-	if got := recalled[0].Content.(string); !strings.Contains(got, "alpha budget is 42") {
-		t.Fatalf("reloaded recall = %q, want alpha budget", got)
-	}
-}
-
 // A custom tool is offered to the model and dispatched to its handler.
 func TestWithToolExposedAndDispatched(t *testing.T) {
 	var got map[string]any

agent/options.go

@@ -246,22 +246,13 @@ func WithMemory(m Memory) Option {
 // turns are injected into matching future asks.
 func CompactMemory(maxMessages, keepRecent int) Option {
 	return func(o *Options) {
-		o.MemoryCompaction.MaxMessages = maxMessages
-		o.MemoryCompaction.KeepRecent = keepRecent
+		o.MemoryCompaction = MemoryCompaction{MaxMessages: maxMessages, KeepRecent: keepRecent}
 		if o.MemoryRecallLimit == 0 {
 			o.MemoryRecallLimit = 5
 		}
 	}
 }
 
-// MemorySummarizer sets the deterministic summarization hook used by the
-// default compacting memory. It is optional; without it, compacted memory uses
-// a provider-neutral text summary. The hook receives the older messages being
-// removed from active context and returns the replacement summary message.
-func MemorySummarizer(fn MemorySummaryFunc) Option {
-	return func(o *Options) { o.MemoryCompaction.Summarize = fn }
-}
-
 // MemoryRecallLimit sets how many archived turns a memory backend may inject
 // into a model request for the current Ask. Use 0 to disable retrieval.
 func MemoryRecallLimit(n int) Option {

agent/otel.go

@@ -22,25 +22,23 @@ const (
 	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"
-	AttrAttempt          = "agent.model.attempt"
-	AttrMaxAttempts      = "agent.model.max_attempts"
-	AttrToolName         = "agent.tool.name"
-	AttrDelegate         = "agent.delegate"
-	AttrGuardrailBlock   = "agent.guardrail.block"
-	AttrRefusal          = "agent.refusal"
-	AttrInputChars       = "agent.input.chars"
-	AttrErrorKind        = "agent.error.kind"
-	AttrCheckpointStatus = "agent.checkpoint.status"
-	AttrCheckpointStage  = "agent.checkpoint.stage"
+	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"
+	AttrAttempt        = "agent.model.attempt"
+	AttrMaxAttempts    = "agent.model.max_attempts"
+	AttrToolName       = "agent.tool.name"
+	AttrDelegate       = "agent.delegate"
+	AttrGuardrailBlock = "agent.guardrail.block"
+	AttrRefusal        = "agent.refusal"
+	AttrInputChars     = "agent.input.chars"
+	AttrErrorKind      = "agent.error.kind"
 )
 
 type RunEvent struct {
@@ -59,7 +57,6 @@ type RunEvent struct {
 	LatencyMS   int64     `json:"latency_ms,omitempty"`
 	Tokens      Usage     `json:"tokens,omitempty"`
 	Refused     string    `json:"refused,omitempty"`
-	Status      string    `json:"status,omitempty"`
 	Error       string    `json:"error,omitempty"`
 	ErrorKind   string    `json:"error_kind,omitempty"`
 	InputChars  int       `json:"input_chars,omitempty"`
@@ -292,15 +289,6 @@ func classifyToolError(err string) string {
 	}
 }
 
-func (a *agentImpl) recordTimelineEvent(ctx context.Context, e RunEvent) {
-	span := trace.SpanFromContext(ctx)
-	if span.SpanContext().IsValid() {
-		a.recordSpanEvent(span, e)
-		return
-	}
-	a.recordRunEvent(e)
-}
-
 func (a *agentImpl) recordSpanEvent(span trace.Span, e RunEvent) {
 	if sc := span.SpanContext(); sc.IsValid() {
 		e.TraceID = sc.TraceID().String()
@@ -349,14 +337,6 @@ func runEventAttributes(e RunEvent) []attribute.KeyValue {
 	if e.ErrorKind != "" {
 		attrs = append(attrs, attribute.String(AttrErrorKind, e.ErrorKind))
 	}
-	if e.Kind == "checkpoint" {
-		if e.Status != "" {
-			attrs = append(attrs, attribute.String(AttrCheckpointStatus, e.Status))
-		}
-		if e.Name != "" {
-			attrs = append(attrs, attribute.String(AttrCheckpointStage, e.Name))
-		}
-	}
 	return attrs
 }
 

agent/otel_test.go

@@ -10,7 +10,6 @@ import (
 	"time"
 
 	"go-micro.dev/v6/ai"
-	"go-micro.dev/v6/flow"
 	"go-micro.dev/v6/store"
 	"go.opentelemetry.io/otel/attribute"
 	"go.opentelemetry.io/otel/codes"
@@ -389,74 +388,6 @@ func TestAgentRunTimelineRecordsModelAndToolWithoutTraceProvider(t *testing.T) {
 	}
 }
 
-func TestAgentCheckpointAndResumeTimelineEvents(t *testing.T) {
-	exp := tracetest.NewInMemoryExporter()
-	tp := trace.NewTracerProvider(trace.WithSyncer(exp))
-	st := store.NewMemoryStore()
-	cp := flow.StoreCheckpoint(st, "resume-otel-agent")
-	first := true
-	fakeGen = func(ctx context.Context, opts ai.Options, req *ai.Request) (*ai.Response, error) {
-		if first {
-			first = false
-			return nil, errors.New("temporary provider failure")
-		}
-		return &ai.Response{Reply: "resumed"}, nil
-	}
-	defer func() { fakeGen = nil }()
-
-	a := newTestAgent(Name("resume-otel-agent"), WithStore(st), WithCheckpoint(cp), TraceProvider(tp))
-	_, err := a.Ask(context.Background(), "resume me")
-	if err == nil {
-		t.Fatal("Ask succeeded, want simulated failure")
-	}
-
-	runs, err := cp.List(context.Background())
-	if err != nil {
-		t.Fatal(err)
-	}
-	if len(runs) != 1 {
-		t.Fatalf("checkpointed runs = %d, want 1", len(runs))
-	}
-	resp, err := Resume(context.Background(), a, runs[0].ID)
-	if err != nil {
-		t.Fatalf("Resume: %v", err)
-	}
-	if resp.Reply != "resumed" {
-		t.Fatalf("reply = %q, want resumed", resp.Reply)
-	}
-
-	events, err := LoadRunEvents(st, "resume-otel-agent", runs[0].ID)
-	if err != nil {
-		t.Fatal(err)
-	}
-	seen := map[string]bool{"checkpoint": false, "resume": false}
-	for _, e := range events {
-		if _, ok := seen[e.Kind]; ok {
-			seen[e.Kind] = true
-		}
-	}
-	for kind, ok := range seen {
-		if !ok {
-			t.Fatalf("missing %s event in timeline: %#v", kind, events)
-		}
-	}
-
-	var resumeSpanEvent bool
-	for _, s := range exp.GetSpans().Snapshots() {
-		if s.Name() != spanNameRun {
-			continue
-		}
-		for _, e := range s.Events() {
-			if e.Name == "agent.resume" {
-				resumeSpanEvent = true
-			}
-		}
-	}
-	if !resumeSpanEvent {
-		t.Fatal("run span missing agent.resume event")
-	}
-}
-
 func TestLoadRunEventsSortsTimelineKeys(t *testing.T) {
 	st := store.NewMemoryStore()
 	scoped := store.Scope(st, "agent", "runner")

agent/stream.go

@@ -181,43 +181,6 @@ func (a *agentImpl) resumeWithStreamEvents(ctx context.Context, runID string, ev
 	return a.askLocked(ctx, run.ID, string(run.State.Data), run.ParentID, &run, false)
 }
 
-type agentStreamAdapter struct {
-	stream AgentStream
-}
-
-func (s *agentStreamAdapter) Recv() (*ai.Response, error) {
-	for {
-		event, err := s.stream.Recv()
-		if err != nil {
-			return nil, err
-		}
-		if event == nil {
-			continue
-		}
-		switch event.Type {
-		case StreamEventToken:
-			if event.Token == "" {
-				continue
-			}
-			return &ai.Response{Reply: event.Token}, nil
-		case StreamEventDone:
-			return nil, io.EOF
-		}
-	}
-}
-
-func (s *agentStreamAdapter) Close() error {
-	return s.stream.Close()
-}
-
-func (a *agentImpl) streamAskAI(ctx context.Context, message string) (ai.Stream, error) {
-	stream, err := a.StreamAsk(ctx, message)
-	if err != nil {
-		return nil, err
-	}
-	return &agentStreamAdapter{stream: stream}, nil
-}
-
 type agentStream struct {
 	events <-chan *StreamEvent
 	done   <-chan struct{}

flow/flow.go

@@ -76,7 +76,6 @@ type Result struct {
 	Answer    string    `json:"answer,omitempty"`
 	ToolCalls []string  `json:"tool_calls,omitempty"`
 	Error     string    `json:"error,omitempty"`
-	ErrorKind string    `json:"error_kind,omitempty"`
 	Timestamp time.Time `json:"timestamp"`
 	Duration  float64   `json:"duration_seconds"`
 }
@@ -247,7 +246,6 @@ func (f *Flow) Execute(ctx context.Context, data string) error {
 		result.Duration = time.Since(start).Seconds()
 		if err != nil {
 			result.Error = err.Error()
-			result.ErrorKind = string(ai.ClassifyError(err))
 			f.record(result)
 			return err
 		}
@@ -263,7 +261,6 @@ func (f *Flow) Execute(ctx context.Context, data string) error {
 	if err != nil {
 		result.Duration = time.Since(start).Seconds()
 		result.Error = err.Error()
-		result.ErrorKind = string(ai.ClassifyError(err))
 		f.record(result)
 		return fmt.Errorf("discover tools: %w", err)
 	}
@@ -277,7 +274,6 @@ func (f *Flow) Execute(ctx context.Context, data string) error {
 
 	if err != nil {
 		result.Error = err.Error()
-		result.ErrorKind = string(ai.ClassifyError(err))
 		f.record(result)
 		return err
 	}

flow/otel.go

@@ -23,7 +23,6 @@ const (
 	AttrFlowStatus    = "flow.status"
 	AttrFlowAttempts  = "flow.step.attempts"
 	AttrFlowLatencyMS = "flow.latency_ms"
-	AttrFlowErrorKind = "flow.error.kind"
 )
 
 func (f *Flow) tracer() trace.Tracer {
@@ -48,7 +47,6 @@ func (f *Flow) startRunSpan(ctx context.Context, run Run) (context.Context, func
 		)
 		if err != nil {
 			span.RecordError(err)
-			span.SetAttributes(attribute.String(AttrFlowErrorKind, string(ai.ClassifyError(err))))
 			span.SetStatus(codes.Error, err.Error())
 		} else {
 			span.SetStatus(codes.Ok, "")
@@ -76,7 +74,6 @@ func (f *Flow) runStepSpan(ctx context.Context, step Step, in State) (State, int
 	)
 	if err != nil {
 		span.RecordError(err)
-		span.SetAttributes(attribute.String(AttrFlowErrorKind, string(ai.ClassifyError(err))))
 		span.SetStatus(codes.Error, err.Error())
 	} else {
 		span.SetStatus(codes.Ok, "")

flow/steps.go

@@ -63,12 +63,11 @@ type Step struct {
 
 // StepRecord is the recorded outcome of one step within a run.
 type StepRecord struct {
-	Name      string `json:"name"`
-	Status    string `json:"status"` // pending | in_progress | done | failed
-	Attempts  int    `json:"attempts"`
-	Result    string `json:"result,omitempty"`
-	Error     string `json:"error,omitempty"`
-	ErrorKind string `json:"error_kind,omitempty"`
+	Name     string `json:"name"`
+	Status   string `json:"status"` // pending | in_progress | done | failed
+	Attempts int    `json:"attempts"`
+	Result   string `json:"result,omitempty"`
+	Error    string `json:"error,omitempty"`
 }
 
 // Run is the persisted record of one flow execution — what a Checkpoint
@@ -432,7 +431,6 @@ func (f *Flow) runFrom(ctx context.Context, run Run) (Run, error) {
 			spanErr = err
 			run.Steps[i].Status = "failed"
 			run.Steps[i].Error = err.Error()
-			run.Steps[i].ErrorKind = string(ai.ClassifyError(err))
 			run.Status = "failed"
 			if saveErr := f.save(ctx, run); saveErr != nil {
 				spanErr = saveErr
@@ -557,7 +555,6 @@ func resultFromRun(trigger string, run Run) Result {
 		r.ToolCalls = append(r.ToolCalls, s.Name+":"+s.Status)
 		if s.Error != "" {
 			r.Error = s.Error
-			r.ErrorKind = s.ErrorKind
 		}
 	}
 	if run.Status == "done" {

flow/steps_test.go

@@ -548,37 +548,3 @@ func TestStateSetScan(t *testing.T) {
 		t.Errorf("round-trip failed: %+v", got)
 	}
 }
-
-func TestFlowFailureRecordsErrorKind(t *testing.T) {
-	cp := StoreCheckpoint(store.NewMemoryStore(), "failure-kind")
-	f := New("failure-kind",
-		WithCheckpoint(cp),
-		Steps(Step{Name: "limited", Run: func(_ context.Context, in State) (State, error) {
-			return in, errors.New("rate limit exceeded")
-		}}),
-	)
-
-	err := f.Execute(context.Background(), "payload")
-	if err == nil {
-		t.Fatal("Execute error = nil, want failure")
-	}
-
-	runs, listErr := cp.List(context.Background())
-	if listErr != nil {
-		t.Fatalf("List: %v", listErr)
-	}
-	if len(runs) != 1 {
-		t.Fatalf("runs = %d, want 1", len(runs))
-	}
-	if got := runs[0].Steps[0].ErrorKind; got != string(ai.ErrorKindRateLimited) {
-		t.Fatalf("step error kind = %q, want %q", got, ai.ErrorKindRateLimited)
-	}
-
-	results := f.Results()
-	if len(results) != 1 {
-		t.Fatalf("results = %d, want 1", len(results))
-	}
-	if got := results[0].ErrorKind; got != string(ai.ErrorKindRateLimited) {
-		t.Fatalf("result error kind = %q, want %q", got, ai.ErrorKindRateLimited)
-	}
-}

internal/docs/PRIORITIES.md

@@ -21,9 +21,11 @@ changes, architectural rewrites. Those go to the human.
 
 ## Work queue (ranked)
 
-1. **Support A2A task resubscribe and input-required handoffs** ([#3474](https://github.com/micro/go-micro/issues/3474)) — the earlier Now/Next hardening seams are closed for this pass: cross-provider dispatch, failure classification, checkpoint/resume trace coverage, RunInfo spans, end-to-end A2A streaming, and memory summarization/retrieval hooks have shipped. The highest-value remaining roadmap gap is live-operation continuity over A2A: remote agents must be able to reconnect to task streams and handle explicit `input-required` pauses so Go Micro agents remain dependable neighbours over open protocols.
+1. **Add scheduled cross-provider 0→hero conformance** ([#3454](https://github.com/micro/go-micro/issues/3454)) — the previous top Now/Next items shipped: verification/grader loops landed in #3443, run-trace optimization analysis landed in #3447, and durable agent resume was completed by #3452, closing #3449. With the core loop primitives now in place, the highest-value Now-phase gap is trust: the same scaffold → run → chat/tool → workflow path must stay true across supported providers on a schedule, with keyed runs gated and no-secret CI still useful. This keeps the services → agents → workflows lifecycle cohesive instead of letting provider behavior drift behind green unit tests.
 
-2. **Add a CI-verifiable agent verification loop** ([#3481](https://github.com/micro/go-micro/issues/3481)) — blog/32 makes the next strategic frontier explicit: scheduled, looping, work-performing agents need a verification/grader loop around the agent loop. This ranks just behind the open A2A continuity issue because it is an internal operational-harness primitive rather than a named roadmap item, but it is the next cohesive step after RunInfo spans, failure semantics, and memory: grade outputs, route feedback through existing retry/supervision paths, and make dependable agent work observable without drifting into a graph DSL.
+2. **Emit OpenTelemetry spans for agent RunInfo timelines** ([#3455](https://github.com/micro/go-micro/issues/3455)) — flows now have trace-oriented optimization feedback, and durable agent runs can resume, but the agent side still needs first-class operability in production traces. Translating `RunInfo` / run timeline events into spans closes a Next-phase observability seam across agent runs, tool calls, model calls, retries, failures, and checkpoint/resume events without changing public APIs.
+
+3. **Complete end-to-end chat and A2A streaming coverage** ([#3456](https://github.com/micro/go-micro/issues/3456)) — provider streaming conformance and A2A fallback work recently shipped, but the mission is one runtime where agents can operate as services, which means streaming must be dependable through the whole path: provider tokens → chat / `Agent.Chat` → A2A. This remains behind conformance and observability because it is a Next-phase depth item, but it is the next user-visible seam in the developer inner loop and interop story.
 
 _Seeded by Claude Code from the roadmap + open issues; thereafter maintained by the
 architecture-review pass._

internal/website/blog/32.md

@@ -1,89 +0,0 @@
----
-layout: blog
-title: "An Agent Is a Service: Where Agent Frameworks Are Going"
-permalink: /blog/32
-description: "A field guide to the agent-framework landscape — from LangChain and the first wave, through the two layers of a harness and the rise of loop engineering, to where the frameworks diverge. And why Go Micro's answer is that an agent is a service."
----
-
-# An Agent Is a Service: Where Agent Frameworks Are Going
-
-*June 30, 2026 • By the Go Micro Team*
-
-There are now a lot of ways to build an agent. LangChain and LangGraph, LlamaIndex, CrewAI, Microsoft's AutoGen, Google's ADK, the model labs' own SDKs, and — most recently in our own backyard — [tRPC-Agent-Go](https://github.com/trpc-group/trpc-agent-go) from Tencent. They are not all solving the same problem, and the places they differ tell you a lot about where this is heading.
-
-This is a field guide to that landscape, and an honest account of where Go Micro sits in it.
-
-## The first wave: a model in a loop
-
-The first wave of agent frameworks solved one thing: get a model to call tools in a loop until a task is done. LangChain, more than any other project, defined that category in 2022 — chains, then agents, then graphs. LlamaIndex came at it from the data and retrieval side. CrewAI and AutoGen leaned into multi-agent orchestration — crews and conversations of role-played agents. The model labs shipped their own agent SDKs so you could stay close to the metal.
-
-That first problem — model, tools, a loop — is now largely commoditized. Every SDK does it, and they mostly do it well. Which means the interesting question has moved. It is no longer "how do I get a model to use a tool." It is everything that happens *around* the loop once the agent has to do real work: connect to real systems, hold state across restarts, recover from failure, be observed, be scheduled, and be reached by other agents. That is the part that decides whether an agent makes it out of a demo.
-
-LangChain itself is the clearest evidence. The framework was the distribution; the value moved to *operating* agents — which is why their commercial product is LangSmith (observability, evaluation, monitoring), not the framework. The lesson the pioneer taught is that the framework gets you to a running agent, and the hard, durable, valuable problems are in operating it.
-
-## "Agent = Model + Harness" — but a harness has two layers
-
-LangChain has a good framing for this: an agent is a model plus a *harness* — the runtime around the model that makes it useful. The framing is right. What is usually left implicit is that "harness" has two distinct layers, and almost all the frameworks live in the first one.
-
-**The intra-agent harness** is the runtime around a *single model*: the system prompt, the tool definitions, context management and compaction, the sandbox, self-verification, and the continuation loop that keeps the model going until it is done. LangChain and LangGraph, deepagents, Claude Code, and the model labs' SDKs are excellent at this. It is real, hard work, and it is most of what people mean when they say "agent framework."
-
-**The operational harness** is the distributed substrate an agent *operates inside*: services exposed as typed tools, discovery and RPC, durable and resumable runs, observability, scheduling, and the protocols agents use to reach each other. This is the layer where a single agent stops being a script and becomes part of a system — where many agents, many services, and many workflows have to compose without falling over.
-
-The first layer produces an agent. The second is where that agent has to live. Most frameworks build the first and leave the second to you — you bring your own services, your own discovery, your own durability, your own deployment. That is the gap that matters now, because the moment you have more than one agent or one service, the operational harness *is* the product.
-
-## The loop is the new frontier
-
-If the first wave was "a model in a loop," the direction now is what LangChain has started calling [loop engineering](https://www.langchain.com/blog/the-art-of-loop-engineering): stacking loops around the agent. It is a useful map. There is the **agent loop** (model calls tools until done), the **verification loop** (a grader checks the output against a rubric and sends failures back with feedback), the **event-driven loop** (the agent is triggered by webhooks, schedules, or messages instead of a human typing), and the **hill-climbing loop** (production traces feed back to improve the prompts, tools, and graders over time).
-
-Notice that only the first of those four is the intra-agent harness. The other three — verification, event-driven triggers, learning from traces — are the operational harness. The frontier is moving from "answer a prompt" to **scheduled, looping, work-performing agents**: agents that run on a cadence, do real work, check their own output, and get better. That is exactly the layer that is underbuilt, and it is the layer that decides whether agents are dependable.
-
-## Where the frameworks are going
-
-Survey the field and a shape emerges. LangChain and LangGraph pair graph-based orchestration with LangSmith for operations, funded to build the team that operates the platform. CrewAI and AutoGen are converging on multi-agent orchestration patterns. Google's ADK is a strong code-first framework with first-class evaluation, tuned for Gemini and Google Cloud. tRPC-Agent-Go brings a production-grade Go agent SDK — LLM, Chain, Parallel, Cycle, and Graph agents; tools; MCP and A2A; memory and RAG; evaluation; agent self-evolution; OpenTelemetry — maintained by Tencent's tRPC group and validated inside Tencent.
-
-They differ in the details, but most share two structural choices. They are an **agent SDK you run alongside your services** — the agents are a layer, and your service tier lives somewhere else and is called into. And they are **graph-centric** — you compose agents and tools into graphs and conditional workflows. That is a coherent, well-trodden approach, and for a lot of teams it is exactly right.
-
-Go Micro starts somewhere else.
-
-## Where Go Micro fits: an agent is a service
-
-Go Micro's position is a single claim: **an agent is a service.** Not a layer bolted onto a service tier — the same runtime.
-
-The reasoning is straightforward. The moment an agent has to discover services, call them, hold state, and recover from failure, it *is* a distributed system. That is precisely the problem a service framework already solves. So instead of building an agent SDK that sits next to your services, Go Micro makes agents and services the same primitives:
-
-- **Every service endpoint is automatically an AI-callable tool**, derived from registry metadata. You do not wire tools into a graph; you write a service and it is already a tool, reachable over MCP.
-- **An agent is a service.** It registers, is discovered, load-balances, exposes an `Agent.Chat` RPC, keeps store-backed memory, and is reachable over A2A — the same lifecycle as anything else you run.
-- **Workflows are durable code paths, not a graph DSL.** Use a `flow` of checkpointed steps where the path is known; dispatch to an agent where it is not. The deterministic parts are plain, resumable Go; the dynamic parts are agents.
-
-The premise is that the line between "your services" and "your agents" is accidental complexity. Remove it, and there is less to wire, less to keep in sync, and a much shorter path from a service to an agent that uses it. The operational harness — discovery, RPC, pub/sub, durable runs, observability, deployment — is not something you assemble around the framework. It *is* the framework.
-
-This is also why Go Micro is deliberately not a graph DSL. Graphs are expressive, and for some teams that visual, declarative model is the draw. But a graph is one more thing to learn and maintain next to your services. "It is just services and durable flows" is a smaller surface to hold in your head, and it composes with everything a service already does.
-
-## A concrete contrast: tRPC-Agent-Go
-
-Because it is the closest neighbour — a serious, production Go framework — tRPC-Agent-Go makes the fork concrete. It is an agent SDK that runs alongside your tRPC services, organised around graph, chain, parallel, and cycle agents. Go Micro is one runtime where the agent *is* the service and orchestration is durable flows.
-
-We will be honest about where they are ahead: tRPC-Agent-Go ships a first-class evaluation framework, agent self-evolution, AG-UI streaming, and RAG today. Go Micro has the trace foundation (OpenTelemetry run timelines, `micro runs`) and has the verification/grader loop and richer memory on the roadmap — but if you need those right now, they are further along there, with a large team behind them. Pretending the checklists match would help no one.
-
-What Go Micro offers in return is the thing an SDK-alongside-your-services cannot: services that become tools with zero glue, agents that are first-class services, and one set of primitives — service, agent, flow — instead of a service stack plus an agent layer plus a graph runtime.
-
-## The direction we're building
-
-If scheduled, looping, work-performing agents are where this goes, then the operational harness is the thing to get right, and loops are the organising idea. Go Micro already has the agent loop, durable event-driven flows, and the trace foundation for learning. The verification loop — grade a step's output against a rubric and route failures back with feedback — is the next primitive, building on the supervised loop and retry machinery already there. Durable agent runs, streaming end to end, and richer observability are on the same line. The aim is not to win a feature checklist; it is to be the runtime where an operating agent is dependable.
-
-There is one more piece of evidence we find hard to argue with: Go Micro is increasingly built by its own loop — an autonomous improvement loop running in CI, opening and merging its own changes against a thesis. An agent harness, operated by agents, building itself. If it is good enough to do that, it is good enough to operate yours.
-
-## Open protocols, different homes
-
-None of this is winner-take-all, and it should not be. Every serious framework here speaks **MCP** for tools and **A2A** for agents. A Go Micro agent and a tRPC-Agent-Go agent can call each other; either can consume the other's tools; an ADK or LangGraph agent can plug into a Go Micro runtime over A2A, and the reverse. The protocols are the commons.
-
-So the real question is not which framework wins. It is where your agents should *live*. The answer that Go Micro is built around is that when an agent has to operate inside a real system, it is a distributed system — and the simplest place to build it is the runtime where your services already live.
-
----
-
-*Go Micro is an open source agent harness and service framework for Go. [Star us on GitHub](https://github.com/micro/go-micro).*
-
-<div class="post-nav">
-  <div><a href="/blog/31">&larr; How Go Micro Builds Itself</a></div>
-  <div><a href="/blog/">All Posts</a></div>
-</div>

internal/website/blog/index.html

@@ -11,13 +11,6 @@ permalink: /blog/
 
 <div class="posts">
 
-  <article style="margin-bottom: 2rem; padding-bottom: 1.5rem; border-bottom: 1px solid #e5e5e5;">
-    <h2 style="margin: 0 0 0.5rem;"><a href="/blog/32">An Agent Is a Service: Where Agent Frameworks Are Going</a></h2>
-    <p class="meta" style="color: #666; font-size: 0.85rem;">June 30, 2026</p>
-    <p>A field guide to the agent-framework landscape — LangChain and the first wave, the two layers of a harness, the rise of loop engineering, and where the frameworks (LangGraph, ADK, CrewAI, AutoGen, tRPC-Agent-Go) diverge. And why Go Micro's answer is that an agent is a service.</p>
-    <a href="/blog/32">Read more &rarr;</a>
-  </article>
-
   <article style="margin-bottom: 2rem; padding-bottom: 1.5rem; border-bottom: 1px solid #e5e5e5;">
     <h2 style="margin: 0 0 0.5rem;"><a href="/blog/31">How Go Micro Builds Itself</a></h2>
     <p class="meta" style="color: #666; font-size: 0.85rem;">June 25, 2026</p>

internal/website/docs/guides/comparison.md

@@ -227,53 +227,6 @@ and an ADK agent (in any language) can call each other over A2A, and either can
 consume the other's MCP tools. A common pattern is to run Go Micro as the service
 mesh / runtime and let ADK (or any A2A agent) plug into it.
 
-## vs tRPC-Agent-Go
-
-[tRPC-Agent-Go](https://github.com/trpc-group/trpc-agent-go) (maintained by tRPC-Group,
-validated inside Tencent) is a production-grade Go framework for agent systems:
-LLM / Chain / Parallel / Cycle / Graph agents, function tools, MCP, A2A, AG-UI, Redis
-memory and RAG, evaluation, agent self-evolution, and OpenTelemetry. It's a serious,
-well-resourced project.
-
-They overlap heavily on agents but take a different approach. tRPC-Agent-Go is an **agent
-SDK you run alongside your services** — you compose agents and tools into graphs and
-conditional workflows, and your microservices (tRPC) live separately and are called
-into. Go Micro starts from the premise that **an agent is a service** — one runtime
-where every endpoint is automatically a tool, an agent registers and is discovered and
-load-balanced like anything else, and workflows are durable code paths rather than a
-graph DSL. The premise is that the line between "your services" and "your agents" is
-accidental complexity; remove it and there's less to wire and keep in sync.
-
-| | Go Micro | tRPC-Agent-Go |
-|---|----------|---------------|
-| **Primary unit** | A harnessed service (an agent is a service with an LLM inside) | An agent |
-| **Orchestration** | Durable `flow` steps + `Loop` — plain code paths | Graph / Chain / Parallel / Cycle agents (graph DSL) |
-| **Services as tools** | Every endpoint is automatically an MCP tool | Function tools + MCP, wired explicitly |
-| **Service runtime** | Built in — agents *are* services (registry, RPC, load balancing, pub/sub) | Runs alongside your existing service stack (tRPC) |
-| **MCP / A2A** | Both, generated from the registry | Both |
-| **Evaluation / self-evolution** | Verification loop on the roadmap; not yet first-class | First-class today |
-| **Memory / RAG** | Store-backed memory (Postgres, NATS KV, file); RAG on the roadmap | In-memory / Redis memory; RAG today |
-| **Observability** | OpenTelemetry run timelines, `micro runs` | OpenTelemetry, Langfuse examples |
-| **Backing** | Independent, community | tRPC-Group / Tencent |
-
-### When to choose tRPC-Agent-Go
-- You want a graph/workflow DSL for composing agents and tools
-- You're on tRPC, or want to add agents alongside an existing service stack
-- You want first-class evaluation and self-evolution today, with a large team behind it
-
-### When to choose Go Micro
-- You want one runtime where services, agents, and flows are the same primitives —
-  registered, discoverable, and deployed the same way
-- You want your existing services to become agent tools with zero extra code
-- You prefer durable flows and plain code paths over a graph DSL, in a small,
-  independent framework you can hold in your head
-
-### They interoperate
-
-Both speak **MCP** and **A2A**, so a Go Micro agent and a tRPC-Agent-Go agent can call
-each other over A2A, and either can consume the other's MCP tools. You can run Go Micro
-as the service-and-agent runtime and still reach an agent built on tRPC-Agent-Go.
-
 ## Feature Deep Dive
 
 ### Service Discovery