Update architect priorities after failure hardening

agent/agent.go

@@ -139,10 +139,6 @@ func (a *agentImpl) String() string {
 }
 
 func (a *agentImpl) setup() {
-	a.setupWithToolHandler(nil)
-}
-
-func (a *agentImpl) setupWithToolHandler(handler ai.ToolHandler) {
 	var modelOpts []ai.Option
 	modelOpts = append(modelOpts, ai.WithAPIKey(a.opts.APIKey))
 	if a.opts.Model != "" {
@@ -150,19 +146,12 @@ func (a *agentImpl) setupWithToolHandler(handler ai.ToolHandler) {
 	}
 
 	a.tools = ai.NewTools(a.opts.Registry, ai.ToolClient(a.opts.Client))
-	if handler == nil {
-		handler = a.toolHandler()
-	}
-	modelOpts = append(modelOpts, ai.WithToolHandler(handler))
+	modelOpts = append(modelOpts, ai.WithToolHandler(a.toolHandler()))
 	a.model = ai.New(a.opts.Provider, modelOpts...)
 	if a.model != nil {
 		a.model = a.tracedModel(a.model)
 	}
 
-	if a.mem != nil {
-		return
-	}
-
 	// Memory is pluggable. Use the configured one, otherwise the default
 	// store-backed memory — except ephemeral sub-agents, which keep an
 	// isolated, non-persistent context.
@@ -236,18 +225,16 @@ func (a *agentImpl) ask(ctx context.Context, message, parentRunID string) (*Resp
 		a.setup()
 	}
 
-	return a.askLocked(ctx, uuid.New().String(), message, parentRunID, nil, true)
+	return a.askLocked(ctx, uuid.New().String(), message, parentRunID, nil)
 }
 
-func (a *agentImpl) askLocked(ctx context.Context, runID, message, parentRunID string, existing *flow.Run, addUserMessage bool) (*Response, error) {
+func (a *agentImpl) askLocked(ctx context.Context, runID, message, parentRunID string, existing *flow.Run) (*Response, error) {
 	toolList, err := a.discoverTools()
 	if err != nil {
 		return nil, fmt.Errorf("discover tools: %w", err)
 	}
 
-	if addUserMessage {
-		a.mem.Add("user", message)
-	}
+	a.mem.Add("user", message)
 	a.steps = 0
 	a.calls = map[string]int{}
 	a.pause = nil

agent/checkpoint.go

@@ -95,7 +95,7 @@ func (a *agentImpl) resume(ctx context.Context, runID string) (*Response, error)
 	if a.model == nil {
 		a.setup()
 	}
-	return a.askLocked(ctx, run.ID, message, parentID, &run, false)
+	return a.askLocked(ctx, run.ID, message, parentID, &run)
 }
 
 // ResumeInput resumes a checkpointed agent run that paused via the built-in
@@ -140,7 +140,7 @@ func (a *agentImpl) resumeInput(ctx context.Context, runID, input string) (*Resp
 	if a.model == nil {
 		a.setup()
 	}
-	return a.askLocked(ctx, run.ID, message, run.ParentID, &run, true)
+	return a.askLocked(ctx, run.ID, message, run.ParentID, &run)
 }
 
 func (a *agentImpl) pending(ctx context.Context) ([]flow.Run, error) {

agent/checkpoint_test.go

@@ -102,72 +102,6 @@ func TestResumeFailedCheckpointDoesNotReplayCompletedTool(t *testing.T) {
 	}
 }
 
-func TestResumeFailedCheckpointDoesNotDuplicateCompactedMemory(t *testing.T) {
-	ctx := context.Background()
-	st := store.NewMemoryStore()
-	cp := flow.StoreCheckpoint(st, "memory-resume-agent")
-	failRetry := true
-	var sawRecall bool
-	fakeGen = func(ctx context.Context, opts ai.Options, req *ai.Request) (*ai.Response, error) {
-		for _, msg := range req.Messages {
-			if text, ok := msg.Content.(string); ok && strings.Contains(text, "alpha code is 42") {
-				sawRecall = true
-			}
-		}
-		if strings.Contains(req.Prompt, "use alpha code") && failRetry {
-			failRetry = false
-			return nil, errors.New("model connection dropped")
-		}
-		return &ai.Response{Reply: "ok"}, nil
-	}
-	defer func() { fakeGen = nil }()
-
-	a := newTestAgent(Name("memory-resume-agent"), WithStore(st), WithCheckpoint(cp), CompactMemory(4, 1), MemoryRecallLimit(2))
-	for _, msg := range []string{"alpha code is 42", "beta note", "gamma note"} {
-		if _, err := a.Ask(ctx, msg); err != nil {
-			t.Fatalf("Ask(%q): %v", msg, err)
-		}
-	}
-
-	_, err := a.Ask(ctx, "use alpha code now")
-	if err == nil {
-		t.Fatal("Ask succeeded, want simulated provider failure")
-	}
-	if got := countMemoryContent(a.mem.Messages(), "use alpha code now"); got != 1 {
-		t.Fatalf("failed Ask stored prompt %d times, want 1", got)
-	}
-
-	runs, err := Pending(ctx, a)
-	if err != nil {
-		t.Fatalf("Pending: %v", err)
-	}
-	if len(runs) != 1 {
-		t.Fatalf("Pending returned %d runs, want 1", len(runs))
-	}
-	if _, err := Resume(ctx, a, runs[0].ID); err != nil {
-		t.Fatalf("Resume: %v", err)
-	}
-	if got := countMemoryContent(a.mem.Messages(), "use alpha code now"); got != 1 {
-		t.Fatalf("resumed failed Ask stored prompt %d times, want no duplicate", got)
-	}
-	if !sawRecall {
-		t.Fatal("resume did not retrieve archived compacted memory")
-	}
-	if got := len(a.mem.Messages()); got > 4 {
-		t.Fatalf("compacted memory retained %d messages after resume, want <= 4", got)
-	}
-}
-
-func countMemoryContent(messages []ai.Message, needle string) int {
-	var count int
-	for _, msg := range messages {
-		if text, ok := msg.Content.(string); ok && strings.Contains(text, needle) {
-			count++
-		}
-	}
-	return count
-}
-
 func TestPendingReturnsUnfinishedAgentRuns(t *testing.T) {
 	ctx := context.Background()
 	cp := flow.StoreCheckpoint(store.NewStore(), "pending-agent")

agent/otel.go

@@ -31,8 +31,6 @@ const (
 	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"
@@ -40,22 +38,20 @@ const (
 )
 
 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"`
-	Attempt     int       `json:"attempt,omitempty"`
-	MaxAttempts int       `json:"max_attempts,omitempty"`
-	LatencyMS   int64     `json:"latency_ms,omitempty"`
-	Tokens      Usage     `json:"tokens,omitempty"`
-	Refused     string    `json:"refused,omitempty"`
-	Error       string    `json:"error,omitempty"`
+	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
@@ -148,7 +144,7 @@ func (m *tracedModel) Generate(ctx context.Context, req *ai.Request, opts ...ai.
 		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, Attempt: info.Attempt, MaxAttempts: info.MaxAttempts, LatencyMS: dur, Tokens: 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()
 		}
@@ -166,12 +162,6 @@ func (m *tracedModel) Generate(ctx context.Context, req *ai.Request, opts ...ai.
 	resp, err := m.Model.Generate(ctx, req, opts...)
 	dur := time.Since(start).Milliseconds()
 	attrs := []attribute.KeyValue{attribute.Int64(AttrLatencyMS, dur)}
-	if info.Attempt > 0 {
-		attrs = append(attrs, attribute.Int(AttrAttempt, info.Attempt))
-	}
-	if info.MaxAttempts > 0 {
-		attrs = append(attrs, attribute.Int(AttrMaxAttempts, info.MaxAttempts))
-	}
 	usage := ai.Usage{}
 	if resp != nil {
 		usage = resp.Usage
@@ -185,7 +175,7 @@ func (m *tracedModel) Generate(ctx context.Context, req *ai.Request, opts ...ai.
 		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, Attempt: info.Attempt, MaxAttempts: info.MaxAttempts, LatencyMS: dur, Tokens: 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()
 	}
@@ -284,12 +274,6 @@ func runEventAttributes(e RunEvent) []attribute.KeyValue {
 	if e.Model != "" {
 		attrs = append(attrs, attribute.String(AttrModel, e.Model))
 	}
-	if e.Attempt > 0 {
-		attrs = append(attrs, attribute.Int(AttrAttempt, e.Attempt))
-	}
-	if e.MaxAttempts > 0 {
-		attrs = append(attrs, attribute.Int(AttrMaxAttempts, e.MaxAttempts))
-	}
 	if e.LatencyMS > 0 {
 		attrs = append(attrs, attribute.Int64(AttrLatencyMS, e.LatencyMS))
 	}

agent/otel_test.go

@@ -3,7 +3,6 @@ package agent
 import (
 	"context"
 	"encoding/json"
-	"errors"
 	"fmt"
 	"strings"
 	"testing"
@@ -12,13 +11,10 @@ import (
 	"go-micro.dev/v6/ai"
 	"go-micro.dev/v6/store"
 	"go.opentelemetry.io/otel/attribute"
-	"go.opentelemetry.io/otel/codes"
 	"go.opentelemetry.io/otel/sdk/trace"
 	"go.opentelemetry.io/otel/sdk/trace/tracetest"
 )
 
-const codesError = codes.Error
-
 type otelTestModel struct{ opts ai.Options }
 
 func (m *otelTestModel) Init(opts ...ai.Option) error {
@@ -93,9 +89,6 @@ func TestAgentOpenTelemetrySpans(t *testing.T) {
 		if attrs[AttrRunID] != runID || attrs[AttrAgentName] != "runner" {
 			t.Fatalf("%s missing run correlation attributes: %#v", s.Name(), attrs)
 		}
-		if s.Name() == spanNameModelCall && (attrs[AttrAttempt] != "1" || attrs[AttrMaxAttempts] != "1") {
-			t.Fatalf("model span missing attempt attributes: %#v", attrs)
-		}
 	}
 	keys, err := store.Scope(st, "agent", "runner").List(store.ListPrefix("runs/"))
 	if err != nil {
@@ -132,77 +125,6 @@ func TestAgentOpenTelemetrySpans(t *testing.T) {
 	}
 }
 
-type failingOtelModel struct{ opts ai.Options }
-
-func (m *failingOtelModel) Init(opts ...ai.Option) error {
-	for _, o := range opts {
-		o(&m.opts)
-	}
-	return nil
-}
-func (m *failingOtelModel) Options() ai.Options { return m.opts }
-func (m *failingOtelModel) String() string      { return "otelfail" }
-func (m *failingOtelModel) Stream(context.Context, *ai.Request, ...ai.GenerateOption) (ai.Stream, error) {
-	return nil, nil
-}
-func (m *failingOtelModel) Generate(context.Context, *ai.Request, ...ai.GenerateOption) (*ai.Response, error) {
-	return nil, errors.New("provider exploded")
-}
-
-func init() {
-	ai.Register("otelfail", func(opts ...ai.Option) ai.Model { return &failingOtelModel{opts: ai.NewOptions(opts...)} })
-}
-
-func TestAgentOpenTelemetrySpansModelFailure(t *testing.T) {
-	exp := tracetest.NewInMemoryExporter()
-	tp := trace.NewTracerProvider(trace.WithSyncer(exp))
-	st := store.NewMemoryStore()
-	a := New(Name("failing-runner"), Provider("otelfail"), WithStore(st), TraceProvider(tp))
-	if _, err := a.Ask(context.Background(), "hello"); err == nil {
-		t.Fatal("Ask succeeded, want provider error")
-	}
-
-	spans := exp.GetSpans().Snapshots()
-	var sawRunError, sawModelError bool
-	for _, s := range spans {
-		attrs := spanAttributes(s.Attributes())
-		switch s.Name() {
-		case spanNameRun:
-			if attrs[AttrAgentName] == "failing-runner" && s.Status().Code == codesError {
-				sawRunError = true
-			}
-		case spanNameModelCall:
-			if attrs[AttrAgentName] == "failing-runner" && attrs[AttrAttempt] == "1" && s.Status().Code == codesError {
-				sawModelError = true
-			}
-		}
-	}
-	if !sawRunError || !sawModelError {
-		t.Fatalf("missing error spans: run=%v model=%v spans=%d", sawRunError, sawModelError, len(spans))
-	}
-
-	summaries, err := ListRunSummaries(st, "failing-runner")
-	if err != nil {
-		t.Fatal(err)
-	}
-	if len(summaries) != 1 || summaries[0].Status != "error" || summaries[0].LastError == "" {
-		t.Fatalf("unexpected failure summary: %#v", summaries)
-	}
-	events, err := LoadRunEvents(st, "failing-runner", summaries[0].RunID)
-	if err != nil {
-		t.Fatal(err)
-	}
-	var sawModelEvent bool
-	for _, event := range events {
-		if event.Kind == "model" && event.Attempt == 1 && event.MaxAttempts == 1 && event.Error != "" {
-			sawModelEvent = true
-		}
-	}
-	if !sawModelEvent {
-		t.Fatalf("missing failed model event with attempt metadata: %#v", events)
-	}
-}
-
 func spanEventHasRunInfo(events []trace.Event, name, runID, agentName string) bool {
 	for _, event := range events {
 		if event.Name != name {

agent/stream.go

@@ -1,237 +0,0 @@
-package agent
-
-import (
-	"context"
-	"encoding/json"
-	"errors"
-	"io"
-	"strings"
-	"sync"
-
-	"github.com/google/uuid"
-
-	"go-micro.dev/v6/ai"
-)
-
-// StreamEventType identifies an event emitted by a tool-aware agent stream.
-type StreamEventType string
-
-const (
-	// StreamEventToolStart is emitted immediately before a tool call runs.
-	StreamEventToolStart StreamEventType = "tool_start"
-	// StreamEventToolEnd is emitted after a tool call returns or is refused.
-	StreamEventToolEnd StreamEventType = "tool_end"
-	// StreamEventToken carries a chunk of the final answer.
-	StreamEventToken StreamEventType = "token"
-	// StreamEventDone carries the completed agent response.
-	StreamEventDone StreamEventType = "done"
-)
-
-// StreamEvent is one event from StreamAsk.
-type StreamEvent struct {
-	Type     StreamEventType
-	Token    string
-	ToolCall ai.ToolCall
-	Result   ai.ToolResult
-	Response *Response
-}
-
-// AgentStream is a stream of tool execution events followed by final-answer chunks.
-type AgentStream interface {
-	Recv() (*StreamEvent, error)
-	Close() error
-}
-
-// StreamAsk runs an agent Ask turn with tool start/end events and streams the final answer.
-// It is additive for callers that hold the public Agent interface; concrete agents also
-// expose the same method directly.
-func StreamAsk(ctx context.Context, ag Agent, message string) (AgentStream, error) {
-	streamer, ok := ag.(interface {
-		StreamAsk(context.Context, string) (AgentStream, error)
-	})
-	if !ok {
-		return nil, errors.New("agent: StreamAsk unsupported by implementation")
-	}
-	return streamer.StreamAsk(ctx, message)
-}
-
-// ResumeStreamAsk resumes a checkpointed agent run and emits the same event
-// shape as StreamAsk. Completed runs are streamed from the persisted response;
-// unfinished runs continue from their checkpoint and emit tool events for any
-// work that still needs to run. Tool calls already recorded as done in the
-// checkpoint are reused by the agent checkpoint wrapper and are not re-executed.
-func ResumeStreamAsk(ctx context.Context, ag Agent, runID string) (AgentStream, error) {
-	a, ok := ag.(*agentImpl)
-	if !ok {
-		return nil, errors.New("agent: ResumeStreamAsk unsupported by implementation")
-	}
-	return a.resumeStreamAsk(ctx, runID)
-}
-
-// StreamAsk runs tools like Ask, emits ToolStart/ToolEnd events as they execute,
-// then emits chunks of the final answer followed by a Done event.
-func (a *agentImpl) StreamAsk(ctx context.Context, message string) (AgentStream, error) {
-	events := make(chan *StreamEvent, 16)
-	done := make(chan struct{})
-	s := &agentStream{events: events, done: done}
-
-	go func() {
-		defer close(events)
-		defer close(done)
-		resp, err := a.askWithStreamEvents(ctx, message, events)
-		if err != nil {
-			s.setErr(err)
-			return
-		}
-		for _, tok := range splitStreamTokens(resp.Reply) {
-			if !sendStreamEvent(ctx, events, &StreamEvent{Type: StreamEventToken, Token: tok}) {
-				return
-			}
-		}
-		_ = sendStreamEvent(ctx, events, &StreamEvent{Type: StreamEventDone, Response: resp})
-	}()
-	return s, nil
-}
-
-func (a *agentImpl) resumeStreamAsk(ctx context.Context, runID string) (AgentStream, error) {
-	events := make(chan *StreamEvent, 16)
-	done := make(chan struct{})
-	s := &agentStream{events: events, done: done}
-
-	go func() {
-		defer close(events)
-		defer close(done)
-		resp, err := a.resumeWithStreamEvents(ctx, runID, events)
-		if err != nil {
-			s.setErr(err)
-			return
-		}
-		for _, tok := range splitStreamTokens(resp.Reply) {
-			if !sendStreamEvent(ctx, events, &StreamEvent{Type: StreamEventToken, Token: tok}) {
-				return
-			}
-		}
-		_ = sendStreamEvent(ctx, events, &StreamEvent{Type: StreamEventDone, Response: resp})
-	}()
-	return s, nil
-}
-
-func (a *agentImpl) askWithStreamEvents(ctx context.Context, message string, events chan<- *StreamEvent) (*Response, error) {
-	a.mu.Lock()
-	defer a.mu.Unlock()
-
-	if a.tools == nil {
-		a.tools = ai.NewTools(a.opts.Registry, ai.ToolClient(a.opts.Client))
-	}
-	base := a.toolHandler()
-	handler := func(ctx context.Context, call ai.ToolCall) ai.ToolResult {
-		_ = sendStreamEvent(ctx, events, &StreamEvent{Type: StreamEventToolStart, ToolCall: call})
-		result := base(ctx, call)
-		_ = sendStreamEvent(ctx, events, &StreamEvent{Type: StreamEventToolEnd, ToolCall: call, Result: result})
-		return result
-	}
-	a.setupWithToolHandler(handler)
-	defer a.setupWithToolHandler(nil)
-	return a.askLocked(ctx, uuid.New().String(), message, a.parentRunID, nil, true)
-}
-
-func (a *agentImpl) resumeWithStreamEvents(ctx context.Context, runID string, events chan<- *StreamEvent) (*Response, error) {
-	if a.opts.Checkpoint == nil {
-		return nil, errors.New("agent: ResumeStreamAsk requires a checkpoint")
-	}
-	run, ok, err := a.opts.Checkpoint.Load(ctx, runID)
-	if err != nil {
-		return nil, err
-	}
-	if !ok {
-		return nil, errors.New("agent: checkpointed run not found")
-	}
-	if run.Status == "done" {
-		var resp Response
-		if err := json.Unmarshal(run.State.Data, &resp); err != nil {
-			return nil, err
-		}
-		return &resp, nil
-	}
-
-	a.mu.Lock()
-	defer a.mu.Unlock()
-	if a.tools == nil {
-		a.tools = ai.NewTools(a.opts.Registry, ai.ToolClient(a.opts.Client))
-	}
-	base := a.toolHandler()
-	handler := func(ctx context.Context, call ai.ToolCall) ai.ToolResult {
-		_ = sendStreamEvent(ctx, events, &StreamEvent{Type: StreamEventToolStart, ToolCall: call})
-		result := base(ctx, call)
-		_ = sendStreamEvent(ctx, events, &StreamEvent{Type: StreamEventToolEnd, ToolCall: call, Result: result})
-		return result
-	}
-	a.setupWithToolHandler(handler)
-	defer a.setupWithToolHandler(nil)
-	if run.Status == "paused" {
-		if run.State.Stage == agentInputStep {
-			return nil, errors.New("agent: checkpointed run is input-required; resume with ResumeInput")
-		}
-		run.Status = "running"
-		run.State.Stage = agentAskStep
-	}
-	return a.askLocked(ctx, run.ID, string(run.State.Data), run.ParentID, &run, false)
-}
-
-type agentStream struct {
-	events <-chan *StreamEvent
-	done   <-chan struct{}
-	mu     sync.Mutex
-	err    error
-}
-
-func (s *agentStream) Recv() (*StreamEvent, error) {
-	ev, ok := <-s.events
-	if ok {
-		return ev, nil
-	}
-	s.mu.Lock()
-	defer s.mu.Unlock()
-	if s.err != nil {
-		return nil, s.err
-	}
-	return nil, io.EOF
-}
-
-func (s *agentStream) Close() error {
-	<-s.done
-	return nil
-}
-
-func (s *agentStream) setErr(err error) {
-	s.mu.Lock()
-	defer s.mu.Unlock()
-	s.err = err
-}
-
-func sendStreamEvent(ctx context.Context, events chan<- *StreamEvent, ev *StreamEvent) bool {
-	select {
-	case events <- ev:
-		return true
-	case <-ctx.Done():
-		return false
-	}
-}
-
-func splitStreamTokens(reply string) []string {
-	if reply == "" {
-		return nil
-	}
-	parts := strings.Fields(reply)
-	if len(parts) == 0 {
-		return []string{reply}
-	}
-	out := make([]string, 0, len(parts))
-	for i, part := range parts {
-		if i > 0 {
-			part = " " + part
-		}
-		out = append(out, part)
-	}
-	return out
-}

agent/stream_test.go

@@ -1,175 +0,0 @@
-package agent
-
-import (
-	"context"
-	"errors"
-	"io"
-	"testing"
-
-	"go-micro.dev/v6/ai"
-	"go-micro.dev/v6/flow"
-	"go-micro.dev/v6/store"
-)
-
-func TestStreamAskEmitsToolEventsAndFinalTokens(t *testing.T) {
-	calls := 0
-	fakeGen = func(ctx context.Context, opts ai.Options, req *ai.Request) (*ai.Response, error) {
-		if opts.ToolHandler == nil {
-			t.Fatal("StreamAsk must configure a tool handler")
-		}
-		calls++
-		result := opts.ToolHandler(ctx, ai.ToolCall{ID: "call-1", Name: "echo", Input: map[string]any{"text": "hello"}})
-		return &ai.Response{
-			Reply:     "planning",
-			Answer:    "final answer",
-			ToolCalls: []ai.ToolCall{{ID: "call-1", Name: "echo", Input: map[string]any{"text": "hello"}, Result: result.Content}},
-		}, nil
-	}
-	defer func() { fakeGen = nil }()
-
-	a := newTestAgent(Name("streamer"), WithTool("echo", "echo text", nil, func(ctx context.Context, input map[string]any) (string, error) {
-		return input["text"].(string), nil
-	}))
-	stream, err := a.StreamAsk(context.Background(), "say hello")
-	if err != nil {
-		t.Fatalf("StreamAsk: %v", err)
-	}
-
-	var types []StreamEventType
-	var tokens string
-	var done *Response
-	for {
-		event, err := stream.Recv()
-		if errors.Is(err, io.EOF) {
-			break
-		}
-		if err != nil {
-			t.Fatalf("Recv: %v", err)
-		}
-		types = append(types, event.Type)
-		if event.Type == StreamEventToken {
-			tokens += event.Token
-		}
-		if event.Type == StreamEventDone {
-			done = event.Response
-		}
-	}
-
-	want := []StreamEventType{StreamEventToolStart, StreamEventToolEnd, StreamEventToken, StreamEventToken, StreamEventToken, StreamEventDone}
-	if len(types) != len(want) {
-		t.Fatalf("event types = %v, want %v", types, want)
-	}
-	for i := range want {
-		if types[i] != want[i] {
-			t.Fatalf("event types = %v, want %v", types, want)
-		}
-	}
-	if tokens != "planning final answer" {
-		t.Fatalf("tokens = %q", tokens)
-	}
-	if done == nil || done.Reply != "planning\n\nfinal answer" {
-		t.Fatalf("done response = %#v", done)
-	}
-	if calls != 1 {
-		t.Fatalf("Generate calls = %d, want 1", calls)
-	}
-}
-
-func TestStreamAskHelperRejectsUnsupportedAgent(t *testing.T) {
-	_, err := StreamAsk(context.Background(), unsupportedAgent{}, "hello")
-	if err == nil {
-		t.Fatal("StreamAsk helper should reject unsupported implementations")
-	}
-}
-
-func TestResumeStreamAskDoesNotReplayCompletedTool(t *testing.T) {
-	ctx := context.Background()
-	cp := flow.StoreCheckpoint(store.NewStore(), "stream-resume-agent")
-	toolRuns := 0
-	first := true
-	fakeGen = func(ctx context.Context, opts ai.Options, req *ai.Request) (*ai.Response, error) {
-		if opts.ToolHandler != nil {
-			res := opts.ToolHandler(ctx, ai.ToolCall{ID: "call-1", Name: "charge", Input: map[string]any{"order": "42"}})
-			if res.Content != "charged" {
-				t.Fatalf("tool result = %q, want charged", res.Content)
-			}
-		}
-		if first {
-			first = false
-			return nil, errors.New("stream disconnected after tool")
-		}
-		return &ai.Response{Reply: "finished from streamed checkpoint"}, nil
-	}
-	defer func() { fakeGen = nil }()
-
-	a := newTestAgent(Name("stream-resume-agent"), WithCheckpoint(cp),
-		WithTool("charge", "charge once", nil, func(context.Context, map[string]any) (string, error) {
-			toolRuns++
-			return "charged", nil
-		}))
-	stream, err := a.StreamAsk(ctx, "charge order 42")
-	if err != nil {
-		t.Fatalf("StreamAsk: %v", err)
-	}
-	for {
-		_, err := stream.Recv()
-		if errors.Is(err, io.EOF) {
-			break
-		}
-		if err != nil {
-			break
-		}
-	}
-	if toolRuns != 1 {
-		t.Fatalf("tool executions after failed StreamAsk = %d, want 1", toolRuns)
-	}
-	runs, err := Pending(ctx, a)
-	if err != nil {
-		t.Fatalf("Pending: %v", err)
-	}
-	if len(runs) != 1 {
-		t.Fatalf("Pending returned %d runs, want 1", len(runs))
-	}
-
-	resumed, err := ResumeStreamAsk(ctx, a, runs[0].ID)
-	if err != nil {
-		t.Fatalf("ResumeStreamAsk: %v", err)
-	}
-	var toolEvents int
-	var done *Response
-	for {
-		event, err := resumed.Recv()
-		if errors.Is(err, io.EOF) {
-			break
-		}
-		if err != nil {
-			t.Fatalf("resumed Recv: %v", err)
-		}
-		if event.Type == StreamEventToolStart || event.Type == StreamEventToolEnd {
-			toolEvents++
-		}
-		if event.Type == StreamEventDone {
-			done = event.Response
-		}
-	}
-	if toolRuns != 1 {
-		t.Fatalf("tool executions after ResumeStreamAsk = %d, want completed tool was not replayed", toolRuns)
-	}
-	if toolEvents != 2 {
-		t.Fatalf("resumed tool events = %d, want start/end for replayed checkpoint result", toolEvents)
-	}
-	if done == nil || done.Reply != "finished from streamed checkpoint" || done.RunID != runs[0].ID {
-		t.Fatalf("done response = %#v", done)
-	}
-}
-
-type unsupportedAgent struct{}
-
-func (unsupportedAgent) Name() string                                      { return "unsupported" }
-func (unsupportedAgent) Init(...Option)                                    {}
-func (unsupportedAgent) Options() Options                                  { return Options{} }
-func (unsupportedAgent) Ask(context.Context, string) (*Response, error)    { return nil, nil }
-func (unsupportedAgent) Stream(context.Context, string) (ai.Stream, error) { return nil, nil }
-func (unsupportedAgent) Run() error                                        { return nil }
-func (unsupportedAgent) Stop() error                                       { return nil }
-func (unsupportedAgent) String() string                                    { return "unsupported" }

ai/atlascloud/atlascloud.go

@@ -43,7 +43,6 @@ func init() {
 	ai.RegisterVideo("atlascloud", func(opts ...ai.Option) ai.VideoModel {
 		return NewProvider(opts...)
 	})
-	ai.RegisterStream("atlascloud")
 }
 
 // Provider implements the ai.Model interface for Atlas Cloud.

ai/atlascloud/atlascloud_test.go

@@ -2,11 +2,7 @@ package atlascloud
 
 import (
 	"context"
-	"encoding/json"
 	"errors"
-	"io"
-	"net/http"
-	"net/http/httptest"
 	"testing"
 
 	"go-micro.dev/v6/ai"
@@ -85,58 +81,16 @@ func TestProvider_Generate_NoAPIKey(t *testing.T) {
 	}
 }
 
-func TestProvider_Stream(t *testing.T) {
-	var sawStream, sawIncludeUsage bool
-	ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
-		if r.URL.Path != "/v1/chat/completions" {
-			t.Errorf("path = %s, want /v1/chat/completions", r.URL.Path)
-		}
-		var body map[string]any
-		if err := json.NewDecoder(r.Body).Decode(&body); err != nil {
-			t.Fatalf("decode request: %v", err)
-		}
-		sawStream, _ = body["stream"].(bool)
-		if so, ok := body["stream_options"].(map[string]any); ok {
-			sawIncludeUsage, _ = so["include_usage"].(bool)
-		}
-		w.Header().Set("Content-Type", "text/event-stream")
-		_, _ = w.Write([]byte("data: {\"choices\":[{\"delta\":{\"content\":\"hel\"}}]}\n\n"))
-		_, _ = w.Write([]byte("data: {\"choices\":[{\"delta\":{\"content\":\"lo\"}}]}\n\n"))
-		_, _ = w.Write([]byte("data: {\"choices\":[],\"usage\":{\"prompt_tokens\":7,\"completion_tokens\":2,\"total_tokens\":9}}\n\n"))
-		_, _ = w.Write([]byte("data: [DONE]\n\n"))
-	}))
-	defer ts.Close()
+func TestProvider_Stream_NotImplemented(t *testing.T) {
+	p := NewProvider()
 
-	p := NewProvider(ai.WithAPIKey("test-key"), ai.WithBaseURL(ts.URL))
-	stream, err := p.Stream(context.Background(), &ai.Request{Prompt: "Hello"})
-	if err != nil {
-		t.Fatalf("Stream returned error: %v", err)
-	}
-	defer stream.Close()
-	if !sawStream {
-		t.Fatal("stream request did not set stream=true")
-	}
-	if !sawIncludeUsage {
-		t.Fatal("stream request did not set stream_options.include_usage=true")
+	req := &ai.Request{
+		Prompt: "Hello",
 	}
 
-	first, err := stream.Recv()
-	if err != nil || first.Reply != "hel" {
-		t.Fatalf("first chunk = %#v, %v; want hel", first, err)
-	}
-	second, err := stream.Recv()
-	if err != nil || second.Reply != "lo" {
-		t.Fatalf("second chunk = %#v, %v; want lo", second, err)
-	}
-	usage, err := stream.Recv()
-	if err != nil {
-		t.Fatalf("usage chunk error: %v", err)
-	}
-	if usage.Usage.TotalTokens != 9 || usage.Usage.InputTokens != 7 || usage.Usage.OutputTokens != 2 {
-		t.Fatalf("usage = %#v; want input=7 output=2 total=9", usage.Usage)
-	}
-	if _, err := stream.Recv(); !errors.Is(err, io.EOF) {
-		t.Fatalf("final error = %v, want EOF", err)
+	_, err := p.Stream(context.Background(), req)
+	if !errors.Is(err, ai.ErrStreamingUnsupported) {
+		t.Fatalf("Stream error = %v, want ErrStreamingUnsupported", err)
 	}
 }
 

ai/capabilities_test.go

@@ -34,7 +34,7 @@ func TestRegisteredProviders(t *testing.T) {
 	}
 
 	got = ai.RegisteredProviders("stream")
-	want = []string{"atlascloud", "groq", "mistral", "openai", "together"}
+	want = []string{"openai"}
 	if !reflect.DeepEqual(got, want) {
 		t.Fatalf("RegisteredProviders(stream) = %#v, want %#v", got, want)
 	}
@@ -44,12 +44,12 @@ func TestCapabilityRows(t *testing.T) {
 	got := ai.CapabilityRows()
 	want := []ai.CapabilityRow{
 		{Provider: "anthropic", Capabilities: ai.Capabilities{Model: true}},
-		{Provider: "atlascloud", Capabilities: ai.Capabilities{Model: true, Image: true, Video: true, Stream: true}},
+		{Provider: "atlascloud", Capabilities: ai.Capabilities{Model: true, Image: true, Video: true}},
 		{Provider: "gemini", Capabilities: ai.Capabilities{Model: true}},
-		{Provider: "groq", Capabilities: ai.Capabilities{Model: true, Stream: true}},
-		{Provider: "mistral", Capabilities: ai.Capabilities{Model: true, Stream: true}},
+		{Provider: "groq", Capabilities: ai.Capabilities{Model: true}},
+		{Provider: "mistral", Capabilities: ai.Capabilities{Model: true}},
 		{Provider: "openai", Capabilities: ai.Capabilities{Model: true, Image: true, Stream: true}},
-		{Provider: "together", Capabilities: ai.Capabilities{Model: true, Stream: true}},
+		{Provider: "together", Capabilities: ai.Capabilities{Model: true}},
 	}
 	if !reflect.DeepEqual(got, want) {
 		t.Fatalf("CapabilityRows() = %#v, want %#v", got, want)
@@ -72,7 +72,7 @@ func TestCapabilityMatrix(t *testing.T) {
 	if caps := ai.ProviderCapabilities("openai"); caps != (ai.Capabilities{Model: true, Image: true, Stream: true}) {
 		t.Fatalf("ProviderCapabilities(openai) = %#v", caps)
 	}
-	if caps := ai.ProviderCapabilities("atlascloud"); caps != (ai.Capabilities{Model: true, Image: true, Video: true, Stream: true}) {
+	if caps := ai.ProviderCapabilities("atlascloud"); caps != (ai.Capabilities{Model: true, Image: true, Video: true}) {
 		t.Fatalf("ProviderCapabilities(atlascloud) = %#v", caps)
 	}
 	if caps := ai.ProviderCapabilities("missing"); caps != (ai.Capabilities{}) {
@@ -88,7 +88,7 @@ func TestRegisterStream(t *testing.T) {
 	}
 
 	got := ai.RegisteredProviders("stream")
-	want := []string{"atlascloud", "groq", "mistral", "openai", "test-stream", "together"}
+	want := []string{"openai", "test-stream"}
 	if !reflect.DeepEqual(got, want) {
 		t.Fatalf("RegisteredProviders(stream) = %#v, want %#v", got, want)
 	}

ai/groq/groq.go

@@ -22,14 +22,12 @@ import (
 	"strings"
 
 	"go-micro.dev/v6/ai"
-	"go-micro.dev/v6/ai/internal/openaiapi"
 )
 
 func init() {
 	ai.Register("groq", func(opts ...ai.Option) ai.Model {
 		return NewProvider(opts...)
 	})
-	ai.RegisterStream("groq")
 }
 
 type Provider struct {
@@ -121,7 +119,7 @@ func (p *Provider) Generate(ctx context.Context, req *ai.Request, opts ...ai.Gen
 }
 
 func (p *Provider) Stream(ctx context.Context, req *ai.Request, opts ...ai.GenerateOption) (ai.Stream, error) {
-	return openaiapi.Stream(ctx, p.opts, req, "/v1/chat/completions")
+	return nil, fmt.Errorf("%w: groq provider", ai.ErrStreamingUnsupported)
 }
 
 func (p *Provider) callAPI(ctx context.Context, req map[string]any) (*ai.Response, map[string]any, error) {

ai/groq/groq_test.go

@@ -2,11 +2,7 @@ package groq
 
 import (
 	"context"
-	"encoding/json"
 	"errors"
-	"io"
-	"net/http"
-	"net/http/httptest"
 	"testing"
 
 	"go-micro.dev/v6/ai"
@@ -44,44 +40,9 @@ func TestProvider_Generate_NoAPIKey(t *testing.T) {
 	}
 }
 
-func TestProvider_Stream(t *testing.T) {
-	var sawStream bool
-	ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
-		if r.URL.Path != "/v1/chat/completions" {
-			t.Fatalf("path = %s, want /v1/chat/completions", r.URL.Path)
-		}
-		var body map[string]any
-		if err := json.NewDecoder(r.Body).Decode(&body); err != nil {
-			t.Fatalf("decode request: %v", err)
-		}
-		sawStream, _ = body["stream"].(bool)
-		w.Header().Set("Content-Type", "text/event-stream")
-		_, _ = w.Write([]byte("data: {\"choices\":[{\"delta\":{\"content\":\"hel\"}}]}\n\n"))
-		_, _ = w.Write([]byte("data: {\"choices\":[{\"delta\":{\"content\":\"lo\"}}]}\n\n"))
-		_, _ = w.Write([]byte("data: [DONE]\n\n"))
-	}))
-	defer ts.Close()
-
-	p := NewProvider(ai.WithAPIKey("test-key"), ai.WithBaseURL(ts.URL))
-	stream, err := p.Stream(context.Background(), &ai.Request{Prompt: "Hello"})
-	if err != nil {
-		t.Fatalf("Stream returned error: %v", err)
-	}
-	defer stream.Close()
-	if !sawStream {
-		t.Fatal("stream request did not set stream=true")
-	}
-
-	first, err := stream.Recv()
-	if err != nil || first.Reply != "hel" {
-		t.Fatalf("first chunk = %#v, %v; want hel", first, err)
-	}
-	second, err := stream.Recv()
-	if err != nil || second.Reply != "lo" {
-		t.Fatalf("second chunk = %#v, %v; want lo", second, err)
-	}
-	if _, err := stream.Recv(); !errors.Is(err, io.EOF) {
-		t.Fatalf("final error = %v, want EOF", err)
+func TestProvider_Stream_NotImplemented(t *testing.T) {
+	if _, err := NewProvider().Stream(context.Background(), &ai.Request{Prompt: "hi"}); !errors.Is(err, ai.ErrStreamingUnsupported) {
+		t.Fatalf("Stream error = %v, want ErrStreamingUnsupported", err)
 	}
 }
 

ai/internal/openaiapi/stream.go

@@ -1,117 +0,0 @@
-package openaiapi
-
-import (
-	"bufio"
-	"bytes"
-	"context"
-	"encoding/json"
-	"fmt"
-	"io"
-	"net/http"
-	"strings"
-
-	"go-micro.dev/v6/ai"
-)
-
-// Stream opens an OpenAI-compatible chat completions SSE stream.
-func Stream(ctx context.Context, opts ai.Options, req *ai.Request, basePath string) (ai.Stream, error) {
-	messages := []map[string]any{{"role": "system", "content": req.SystemPrompt}}
-	for _, m := range req.Messages {
-		messages = append(messages, map[string]any{"role": m.Role, "content": m.Content})
-	}
-	if req.Prompt != "" {
-		messages = append(messages, map[string]any{"role": "user", "content": req.Prompt})
-	}
-	apiReq := map[string]any{
-		"model":          opts.Model,
-		"messages":       messages,
-		"stream":         true,
-		"stream_options": map[string]any{"include_usage": true},
-	}
-	if opts.MaxTokens > 0 {
-		apiReq["max_tokens"] = opts.MaxTokens
-	}
-	reqBody, err := json.Marshal(apiReq)
-	if err != nil {
-		return nil, fmt.Errorf("failed to marshal stream request: %w", err)
-	}
-	apiURL := strings.TrimRight(opts.BaseURL, "/") + basePath
-	httpReq, err := http.NewRequestWithContext(ctx, http.MethodPost, apiURL, bytes.NewReader(reqBody))
-	if err != nil {
-		return nil, fmt.Errorf("failed to create stream request: %w", err)
-	}
-	httpReq.Header.Set("Content-Type", "application/json")
-	httpReq.Header.Set("Accept", "text/event-stream")
-	httpReq.Header.Set("Authorization", "Bearer "+opts.APIKey)
-
-	httpResp, err := http.DefaultClient.Do(httpReq)
-	if err != nil {
-		return nil, fmt.Errorf("stream API request failed: %w", err)
-	}
-	if httpResp.StatusCode != http.StatusOK {
-		defer httpResp.Body.Close()
-		respBody, _ := io.ReadAll(httpResp.Body)
-		return nil, fmt.Errorf("stream API error (%s): %s", httpResp.Status, string(respBody))
-	}
-	return &StreamReader{body: httpResp.Body, scanner: bufio.NewScanner(httpResp.Body)}, nil
-}
-
-// StreamReader reads OpenAI-compatible server-sent event chunks.
-type StreamReader struct {
-	body    io.ReadCloser
-	scanner *bufio.Scanner
-	closed  bool
-}
-
-func (s *StreamReader) Recv() (*ai.Response, error) {
-	for s.scanner.Scan() {
-		line := strings.TrimSpace(s.scanner.Text())
-		if line == "" || strings.HasPrefix(line, ":") {
-			continue
-		}
-		if !strings.HasPrefix(line, "data:") {
-			continue
-		}
-		data := strings.TrimSpace(strings.TrimPrefix(line, "data:"))
-		if data == "[DONE]" {
-			return nil, io.EOF
-		}
-		var chunk struct {
-			Choices []struct {
-				Delta struct {
-					Content string `json:"content"`
-				} `json:"delta"`
-			} `json:"choices"`
-			Usage *struct {
-				PromptTokens     int `json:"prompt_tokens"`
-				CompletionTokens int `json:"completion_tokens"`
-				TotalTokens      int `json:"total_tokens"`
-			} `json:"usage"`
-		}
-		if err := json.Unmarshal([]byte(data), &chunk); err != nil {
-			return nil, fmt.Errorf("failed to parse stream chunk: %w", err)
-		}
-		if len(chunk.Choices) > 0 && chunk.Choices[0].Delta.Content != "" {
-			return &ai.Response{Reply: chunk.Choices[0].Delta.Content}, nil
-		}
-		if chunk.Usage != nil {
-			return &ai.Response{Usage: ai.Usage{
-				InputTokens:  chunk.Usage.PromptTokens,
-				OutputTokens: chunk.Usage.CompletionTokens,
-				TotalTokens:  chunk.Usage.TotalTokens,
-			}}, nil
-		}
-	}
-	if err := s.scanner.Err(); err != nil {
-		return nil, err
-	}
-	return nil, io.EOF
-}
-
-func (s *StreamReader) Close() error {
-	if s.closed {
-		return nil
-	}
-	s.closed = true
-	return s.body.Close()
-}

ai/mistral/mistral.go

@@ -22,14 +22,12 @@ import (
 	"strings"
 
 	"go-micro.dev/v6/ai"
-	"go-micro.dev/v6/ai/internal/openaiapi"
 )
 
 func init() {
 	ai.Register("mistral", func(opts ...ai.Option) ai.Model {
 		return NewProvider(opts...)
 	})
-	ai.RegisterStream("mistral")
 }
 
 type Provider struct {
@@ -121,7 +119,7 @@ func (p *Provider) Generate(ctx context.Context, req *ai.Request, opts ...ai.Gen
 }
 
 func (p *Provider) Stream(ctx context.Context, req *ai.Request, opts ...ai.GenerateOption) (ai.Stream, error) {
-	return openaiapi.Stream(ctx, p.opts, req, "/v1/chat/completions")
+	return nil, fmt.Errorf("%w: mistral provider", ai.ErrStreamingUnsupported)
 }
 
 func (p *Provider) callAPI(ctx context.Context, req map[string]any) (*ai.Response, map[string]any, error) {

ai/mistral/mistral_test.go

@@ -2,11 +2,7 @@ package mistral
 
 import (
 	"context"
-	"encoding/json"
 	"errors"
-	"io"
-	"net/http"
-	"net/http/httptest"
 	"testing"
 
 	"go-micro.dev/v6/ai"
@@ -44,44 +40,9 @@ func TestProvider_Generate_NoAPIKey(t *testing.T) {
 	}
 }
 
-func TestProvider_Stream(t *testing.T) {
-	var sawStream bool
-	ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
-		if r.URL.Path != "/v1/chat/completions" {
-			t.Fatalf("path = %s, want /v1/chat/completions", r.URL.Path)
-		}
-		var body map[string]any
-		if err := json.NewDecoder(r.Body).Decode(&body); err != nil {
-			t.Fatalf("decode request: %v", err)
-		}
-		sawStream, _ = body["stream"].(bool)
-		w.Header().Set("Content-Type", "text/event-stream")
-		_, _ = w.Write([]byte("data: {\"choices\":[{\"delta\":{\"content\":\"hel\"}}]}\n\n"))
-		_, _ = w.Write([]byte("data: {\"choices\":[{\"delta\":{\"content\":\"lo\"}}]}\n\n"))
-		_, _ = w.Write([]byte("data: [DONE]\n\n"))
-	}))
-	defer ts.Close()
-
-	p := NewProvider(ai.WithAPIKey("test-key"), ai.WithBaseURL(ts.URL))
-	stream, err := p.Stream(context.Background(), &ai.Request{Prompt: "Hello"})
-	if err != nil {
-		t.Fatalf("Stream returned error: %v", err)
-	}
-	defer stream.Close()
-	if !sawStream {
-		t.Fatal("stream request did not set stream=true")
-	}
-
-	first, err := stream.Recv()
-	if err != nil || first.Reply != "hel" {
-		t.Fatalf("first chunk = %#v, %v; want hel", first, err)
-	}
-	second, err := stream.Recv()
-	if err != nil || second.Reply != "lo" {
-		t.Fatalf("second chunk = %#v, %v; want lo", second, err)
-	}
-	if _, err := stream.Recv(); !errors.Is(err, io.EOF) {
-		t.Fatalf("final error = %v, want EOF", err)
+func TestProvider_Stream_NotImplemented(t *testing.T) {
+	if _, err := NewProvider().Stream(context.Background(), &ai.Request{Prompt: "hi"}); !errors.Is(err, ai.ErrStreamingUnsupported) {
+		t.Fatalf("Stream error = %v, want ErrStreamingUnsupported", err)
 	}
 }
 

ai/together/together.go

@@ -22,14 +22,12 @@ import (
 	"strings"
 
 	"go-micro.dev/v6/ai"
-	"go-micro.dev/v6/ai/internal/openaiapi"
 )
 
 func init() {
 	ai.Register("together", func(opts ...ai.Option) ai.Model {
 		return NewProvider(opts...)
 	})
-	ai.RegisterStream("together")
 }
 
 type Provider struct {
@@ -121,7 +119,7 @@ func (p *Provider) Generate(ctx context.Context, req *ai.Request, opts ...ai.Gen
 }
 
 func (p *Provider) Stream(ctx context.Context, req *ai.Request, opts ...ai.GenerateOption) (ai.Stream, error) {
-	return openaiapi.Stream(ctx, p.opts, req, "/v1/chat/completions")
+	return nil, fmt.Errorf("%w: together provider", ai.ErrStreamingUnsupported)
 }
 
 func (p *Provider) callAPI(ctx context.Context, req map[string]any) (*ai.Response, map[string]any, error) {

ai/together/together_test.go

@@ -2,11 +2,7 @@ package together
 
 import (
 	"context"
-	"encoding/json"
 	"errors"
-	"io"
-	"net/http"
-	"net/http/httptest"
 	"testing"
 
 	"go-micro.dev/v6/ai"
@@ -44,44 +40,9 @@ func TestProvider_Generate_NoAPIKey(t *testing.T) {
 	}
 }
 
-func TestProvider_Stream(t *testing.T) {
-	var sawStream bool
-	ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
-		if r.URL.Path != "/v1/chat/completions" {
-			t.Fatalf("path = %s, want /v1/chat/completions", r.URL.Path)
-		}
-		var body map[string]any
-		if err := json.NewDecoder(r.Body).Decode(&body); err != nil {
-			t.Fatalf("decode request: %v", err)
-		}
-		sawStream, _ = body["stream"].(bool)
-		w.Header().Set("Content-Type", "text/event-stream")
-		_, _ = w.Write([]byte("data: {\"choices\":[{\"delta\":{\"content\":\"hel\"}}]}\n\n"))
-		_, _ = w.Write([]byte("data: {\"choices\":[{\"delta\":{\"content\":\"lo\"}}]}\n\n"))
-		_, _ = w.Write([]byte("data: [DONE]\n\n"))
-	}))
-	defer ts.Close()
-
-	p := NewProvider(ai.WithAPIKey("test-key"), ai.WithBaseURL(ts.URL))
-	stream, err := p.Stream(context.Background(), &ai.Request{Prompt: "Hello"})
-	if err != nil {
-		t.Fatalf("Stream returned error: %v", err)
-	}
-	defer stream.Close()
-	if !sawStream {
-		t.Fatal("stream request did not set stream=true")
-	}
-
-	first, err := stream.Recv()
-	if err != nil || first.Reply != "hel" {
-		t.Fatalf("first chunk = %#v, %v; want hel", first, err)
-	}
-	second, err := stream.Recv()
-	if err != nil || second.Reply != "lo" {
-		t.Fatalf("second chunk = %#v, %v; want lo", second, err)
-	}
-	if _, err := stream.Recv(); !errors.Is(err, io.EOF) {
-		t.Fatalf("final error = %v, want EOF", err)
+func TestProvider_Stream_NotImplemented(t *testing.T) {
+	if _, err := NewProvider().Stream(context.Background(), &ai.Request{Prompt: "hi"}); !errors.Is(err, ai.ErrStreamingUnsupported) {
+		t.Fatalf("Stream error = %v, want ErrStreamingUnsupported", err)
 	}
 }
 

cmd/micro/inspect/inspect.go

@@ -1,169 +0,0 @@
-// Package inspect registers the 'micro inspect' CLI command.
-package inspect
-
-import (
-	"context"
-	"encoding/json"
-	"fmt"
-	"io"
-	"os"
-
-	"github.com/urfave/cli/v2"
-	goagent "go-micro.dev/v6/agent"
-	"go-micro.dev/v6/cmd"
-	aiflow "go-micro.dev/v6/flow"
-	"go-micro.dev/v6/store"
-)
-
-func init() {
-	cmd.Register(&cli.Command{
-		Name:  "inspect",
-		Usage: "Inspect recent agent and workflow activity",
-		Description: `Inspect is the CLI checkpoint in the local scaffold → run → chat → inspect loop.
-It reads durable local run history, so it works after the agent or flow has stopped.`,
-		Subcommands: []*cli.Command{
-			{
-				Name:      "agent",
-				Usage:     "Show recent recorded runs for an agent",
-				ArgsUsage: "[agent]",
-				Flags:     inspectAgentFlags(),
-				Action:    inspectAgent,
-			},
-			{
-				Name:      "flow",
-				Usage:     "Show durable run history for a flow",
-				ArgsUsage: "[flow]",
-				Flags:     inspectFlowFlags(),
-				Action:    inspectFlow,
-			},
-		},
-	})
-}
-
-func inspectAgentFlags() []cli.Flag {
-	return []cli.Flag{
-		&cli.BoolFlag{Name: "json", Usage: "Print run summaries as JSON for automation"},
-		&cli.StringFlag{Name: "status", Usage: "Only show runs with this status (running, done, error, refused)"},
-		&cli.StringFlag{Name: "trace", Usage: "Only show runs whose trace id matches this full id or prefix"},
-		&cli.IntFlag{Name: "limit", Usage: "Show the most recently updated N runs"},
-	}
-}
-
-func inspectFlowFlags() []cli.Flag {
-	return []cli.Flag{
-		&cli.BoolFlag{Name: "json", Usage: "Print durable run history as JSON for automation"},
-		&cli.BoolFlag{Name: "pending", Usage: "Only show runs that have not completed"},
-		&cli.StringFlag{Name: "status", Usage: "Only show runs with this status (running, done, failed)"},
-		&cli.IntFlag{Name: "limit", Usage: "Show the most recently updated N runs"},
-		&cli.StringFlag{Name: "stage", Usage: "Only show runs currently checkpointed at this stage"},
-	}
-}
-
-func inspectAgent(c *cli.Context) error {
-	name := c.Args().First()
-	if name == "" {
-		return fmt.Errorf("agent name required: micro inspect agent <name>")
-	}
-	opts := goagent.RunListOptions{Status: c.String("status"), TraceID: c.String("trace"), Limit: c.Int("limit")}
-	runs, err := goagent.ListRunSummariesWithOptions(store.DefaultStore, name, opts)
-	if err != nil {
-		return err
-	}
-	return writeAgentInspection(os.Stdout, name, runs, c.Bool("json"))
-}
-
-func writeAgentInspection(w io.Writer, name string, runs []goagent.RunSummary, asJSON bool) error {
-	if asJSON {
-		enc := json.NewEncoder(w)
-		enc.SetIndent("", "  ")
-		return enc.Encode(runs)
-	}
-	if len(runs) == 0 {
-		fmt.Fprintf(w, "  No agent runs recorded for %q. After chatting, try: micro inspect agent %s\n", name, name)
-		return nil
-	}
-	fmt.Fprintf(w, "  Agent %q runs\n", name)
-	for _, run := range runs {
-		fmt.Fprintf(w, "  %s  status=%s  events=%d  last=%s", run.RunID, run.Status, run.Events, run.LastKind)
-		if run.LastError != "" {
-			fmt.Fprintf(w, "  error=%q", run.LastError)
-		}
-		if run.TraceID != "" {
-			fmt.Fprintf(w, "  trace=%s", shortID(run.TraceID))
-		}
-		fmt.Fprintln(w)
-	}
-	return nil
-}
-
-func inspectFlow(c *cli.Context) error {
-	name := c.Args().First()
-	if name == "" {
-		return fmt.Errorf("flow name required: micro inspect flow <name>")
-	}
-	runs, err := aiflow.StoreCheckpoint(nil, name).List(context.Background())
-	if err != nil {
-		return err
-	}
-	runs = filterFlowInspection(runs, c.Bool("pending"), c.String("status"), c.String("stage"), c.Int("limit"))
-	return writeFlowInspection(os.Stdout, name, runs, c.Bool("json"), c.Bool("pending"))
-}
-
-func filterFlowInspection(runs []aiflow.Run, pending bool, status, stage string, limit int) []aiflow.Run {
-	filtered := make([]aiflow.Run, 0, len(runs))
-	for _, run := range runs {
-		if pending && run.Status == "done" {
-			continue
-		}
-		if status != "" && run.Status != status {
-			continue
-		}
-		if stage != "" && run.State.Stage != stage {
-			continue
-		}
-		filtered = append(filtered, run)
-	}
-	if limit > 0 && len(filtered) > limit {
-		return filtered[len(filtered)-limit:]
-	}
-	return filtered
-}
-
-func writeFlowInspection(w io.Writer, name string, runs []aiflow.Run, asJSON, pending bool) error {
-	if asJSON {
-		enc := json.NewEncoder(w)
-		enc.SetIndent("", "  ")
-		return enc.Encode(runs)
-	}
-	if len(runs) == 0 {
-		if pending {
-			fmt.Fprintf(w, "  No pending flow runs recorded for %q.\n", name)
-			return nil
-		}
-		fmt.Fprintf(w, "  No flow runs recorded for %q. After executing a durable flow, try: micro inspect flow %s\n", name, name)
-		return nil
-	}
-	fmt.Fprintf(w, "  Flow %q runs\n", name)
-	for _, run := range runs {
-		stage := run.State.Stage
-		if stage == "" {
-			stage = "-"
-		}
-		fmt.Fprintf(w, "  %s  status=%s  stage=%s  steps=%d", shortID(run.ID), run.Status, stage, len(run.Steps))
-		for _, step := range run.Steps {
-			if step.Error != "" {
-				fmt.Fprintf(w, "  error=%q", step.Error)
-				break
-			}
-		}
-		fmt.Fprintln(w)
-	}
-	return nil
-}
-
-func shortID(id string) string {
-	if len(id) <= 12 {
-		return id
-	}
-	return id[:12]
-}

cmd/micro/inspect/inspect_test.go

@@ -1,64 +0,0 @@
-package inspect
-
-import (
-	"bytes"
-	"encoding/json"
-	"strings"
-	"testing"
-
-	goagent "go-micro.dev/v6/agent"
-	aiflow "go-micro.dev/v6/flow"
-)
-
-func TestWriteAgentInspectionIncludesActionableBreadcrumbs(t *testing.T) {
-	runs := []goagent.RunSummary{{RunID: "run-1", Status: "error", Events: 4, LastKind: "tool", LastError: "boom", TraceID: "1234567890abcdef"}}
-	var out bytes.Buffer
-	if err := writeAgentInspection(&out, "support", runs, false); err != nil {
-		t.Fatal(err)
-	}
-	got := out.String()
-	for _, want := range []string{"Agent \"support\" runs", "run-1", "status=error", "events=4", "last=tool", `error="boom"`, "trace=1234567890ab"} {
-		if !strings.Contains(got, want) {
-			t.Fatalf("output missing %q:\n%s", want, got)
-		}
-	}
-}
-
-func TestWriteAgentInspectionEmptyStateNamesInspectCommand(t *testing.T) {
-	var out bytes.Buffer
-	if err := writeAgentInspection(&out, "support", nil, false); err != nil {
-		t.Fatal(err)
-	}
-	if got := out.String(); !strings.Contains(got, "micro inspect agent support") {
-		t.Fatalf("empty state missing next step: %q", got)
-	}
-}
-
-func TestWriteFlowInspectionIncludesFailedStepBreadcrumb(t *testing.T) {
-	runs := []aiflow.Run{{ID: "1234567890abcdef", Status: "failed", State: aiflow.State{Stage: "charge"}, Steps: []aiflow.StepRecord{{Name: "charge", Status: "failed", Error: "card declined"}}}}
-	var out bytes.Buffer
-	if err := writeFlowInspection(&out, "checkout", runs, false, false); err != nil {
-		t.Fatal(err)
-	}
-	got := out.String()
-	for _, want := range []string{"Flow \"checkout\" runs", "1234567890ab", "status=failed", "stage=charge", "steps=1", `error="card declined"`} {
-		if !strings.Contains(got, want) {
-			t.Fatalf("output missing %q:\n%s", want, got)
-		}
-	}
-}
-
-func TestWriteFlowInspectionJSON(t *testing.T) {
-	runs := []aiflow.Run{{ID: "run-1", Flow: "checkout", Status: "done"}}
-	var out bytes.Buffer
-	if err := writeFlowInspection(&out, "checkout", runs, true, false); err != nil {
-		t.Fatal(err)
-	}
-	var got []aiflow.Run
-	if err := json.Unmarshal(out.Bytes(), &got); err != nil {
-		t.Fatalf("invalid JSON: %v\n%s", err, out.String())
-	}
-	if len(got) != 1 || got[0].ID != "run-1" || got[0].Status != "done" {
-		t.Fatalf("decoded runs = %+v", got)
-	}
-}

cmd/micro/main.go

@@ -12,7 +12,6 @@ import (
 	_ "go-micro.dev/v6/cmd/micro/cli/build"
 	_ "go-micro.dev/v6/cmd/micro/cli/deploy"
 	_ "go-micro.dev/v6/cmd/micro/flow"
-	_ "go-micro.dev/v6/cmd/micro/inspect"
 	_ "go-micro.dev/v6/cmd/micro/mcp"
 	_ "go-micro.dev/v6/cmd/micro/resource"
 	_ "go-micro.dev/v6/cmd/micro/run"

cmd/micro/zero_to_hero_cli_test.go

@@ -19,7 +19,7 @@ func TestZeroToHeroCLIBoundaries(t *testing.T) {
 		}
 	}
 
-	for _, want := range []string{"run", "chat", "flow", "inspect"} {
+	for _, want := range []string{"run", "chat", "flow"} {
 		if !commands[want] {
 			t.Fatalf("missing %q command", want)
 		}
@@ -27,7 +27,4 @@ func TestZeroToHeroCLIBoundaries(t *testing.T) {
 	if !subcommands["flow"]["runs"] {
 		t.Fatal("missing inspect boundary: flow runs")
 	}
-	if !subcommands["inspect"]["agent"] || !subcommands["inspect"]["flow"] {
-		t.Fatal("missing inspect boundary: inspect agent/flow")
-	}
 }

events/memory.go

@@ -19,11 +19,7 @@ func NewStream(opts ...Option) (Stream, error) {
 	for _, o := range opts {
 		o(&options)
 	}
-	st := options.Store
-	if st == nil {
-		st = store.NewMemoryStore()
-	}
-	return &mem{store: st}, nil
+	return &mem{store: store.NewMemoryStore()}, nil
 }
 
 type subscriber struct {

events/options.go

@@ -1,25 +1,11 @@
 package events
 
-import (
-	"time"
+import "time"
 
-	"go-micro.dev/v6/store"
-)
-
-type Options struct {
-	// Store persists published events for durability and replay. If nil, an
-	// in-memory store is used and events do not survive a restart.
-	Store store.Store
-}
+type Options struct{}
 
 type Option func(o *Options)
 
-// WithStore backs the stream with a durable store (e.g. the file store), so
-// published events persist and can be replayed across restarts.
-func WithStore(s store.Store) Option {
-	return func(o *Options) { o.Store = s }
-}
-
 type StoreOptions struct {
 	TTL    time.Duration
 	Backup Backup

gateway/mcp/httpjsonrpc.go

@@ -1,129 +0,0 @@
-package mcp
-
-import (
-	"encoding/json"
-	"net/http"
-	"strings"
-)
-
-// HandlerOption configures NewHandler.
-type HandlerOption func(*handlerOptions)
-
-type handlerOptions struct {
-	serverName, serverVersion, protocolVersion string
-}
-
-// WithServerInfo sets the name/version advertised in the initialize response.
-func WithServerInfo(name, version string) HandlerOption {
-	return func(o *handlerOptions) { o.serverName, o.serverVersion = name, version }
-}
-
-// WithProtocolVersion sets the MCP protocol version advertised in initialize.
-func WithProtocolVersion(v string) HandlerOption {
-	return func(o *handlerOptions) { o.protocolVersion = v }
-}
-
-// NewHandler returns an http.Handler serving the MCP protocol over HTTP as
-// JSON-RPC 2.0 (initialize, ping, notifications/*, tools/list, tools/call),
-// backed by the resolver. Mount it on your own server (e.g. POST /mcp): the
-// gateway provides the protocol; you keep your routes, middleware and any
-// human-facing docs page.
-func NewHandler(r Resolver, opts ...HandlerOption) http.Handler {
-	o := handlerOptions{serverName: "go-micro-mcp", serverVersion: "1.0.0", protocolVersion: "2024-11-05"}
-	for _, fn := range opts {
-		fn(&o)
-	}
-	return http.HandlerFunc(func(w http.ResponseWriter, req *http.Request) {
-		if req.Method != http.MethodPost {
-			http.Error(w, "method not allowed", http.StatusMethodNotAllowed)
-			return
-		}
-		var rpc struct {
-			JSONRPC string          `json:"jsonrpc"`
-			ID      json.RawMessage `json:"id"`
-			Method  string          `json:"method"`
-			Params  json.RawMessage `json:"params"`
-		}
-		if err := json.NewDecoder(req.Body).Decode(&rpc); err != nil {
-			writeRPCError(w, nil, ParseError, "Parse error", err.Error())
-			return
-		}
-
-		// Notifications (and any id-less request) expect no response body.
-		if strings.HasPrefix(rpc.Method, "notifications/") || len(rpc.ID) == 0 {
-			w.WriteHeader(http.StatusNoContent)
-			return
-		}
-
-		ctx := req.Context()
-		switch rpc.Method {
-		case "initialize":
-			writeRPCResult(w, rpc.ID, map[string]interface{}{
-				"protocolVersion": o.protocolVersion,
-				"capabilities":    map[string]interface{}{"tools": map[string]interface{}{}},
-				"serverInfo":      map[string]interface{}{"name": o.serverName, "version": o.serverVersion},
-			})
-		case "ping":
-			writeRPCResult(w, rpc.ID, map[string]interface{}{})
-		case "tools/list":
-			tools, err := r.List(ctx)
-			if err != nil {
-				writeRPCError(w, rpc.ID, InternalError, "Failed to list tools", err.Error())
-				return
-			}
-			list := make([]map[string]interface{}, 0, len(tools))
-			for _, t := range tools {
-				list = append(list, map[string]interface{}{
-					"name": t.Name, "description": t.Description, "inputSchema": t.InputSchema,
-				})
-			}
-			writeRPCResult(w, rpc.ID, map[string]interface{}{"tools": list})
-		case "tools/call":
-			var p struct {
-				Name      string                 `json:"name"`
-				Arguments map[string]interface{} `json:"arguments"`
-			}
-			if err := json.Unmarshal(rpc.Params, &p); err != nil {
-				writeRPCError(w, rpc.ID, InvalidParams, "Invalid params", err.Error())
-				return
-			}
-			res, err := r.Call(ctx, p.Name, p.Arguments)
-			if err != nil {
-				// Protocol/pre-check failure -> JSON-RPC error. An *RPCError
-				// carries a specific code; anything else is InternalError.
-				if rpcErr, ok := err.(*RPCError); ok {
-					writeRPCError(w, rpc.ID, rpcErr.Code, rpcErr.Message, rpcErr.Data)
-				} else {
-					writeRPCError(w, rpc.ID, InternalError, "Tool call failed", err.Error())
-				}
-				return
-			}
-			result := map[string]interface{}{
-				"content": []map[string]interface{}{{"type": "text", "text": res.Text}},
-			}
-			if res.IsError {
-				result["isError"] = true
-			}
-			writeRPCResult(w, rpc.ID, result)
-		default:
-			writeRPCError(w, rpc.ID, MethodNotFound, "Method not found", rpc.Method)
-		}
-	})
-}
-
-func writeRPCResult(w http.ResponseWriter, id json.RawMessage, result interface{}) {
-	w.Header().Set("Content-Type", "application/json")
-	_ = json.NewEncoder(w).Encode(map[string]interface{}{"jsonrpc": "2.0", "id": rawOrNull(id), "result": result})
-}
-
-func writeRPCError(w http.ResponseWriter, id json.RawMessage, code int, msg string, data interface{}) {
-	w.Header().Set("Content-Type", "application/json")
-	_ = json.NewEncoder(w).Encode(map[string]interface{}{"jsonrpc": "2.0", "id": rawOrNull(id), "error": map[string]interface{}{"code": code, "message": msg, "data": data}})
-}
-
-func rawOrNull(id json.RawMessage) interface{} {
-	if len(id) == 0 {
-		return nil
-	}
-	return id
-}

gateway/mcp/resolver.go

@@ -1,131 +0,0 @@
-package mcp
-
-import (
-	"context"
-	"sync"
-
-	"go-micro.dev/v6/ai"
-	"go-micro.dev/v6/client"
-	"go-micro.dev/v6/registry"
-)
-
-// CallResult is the outcome of a successful tool dispatch. A tool that ran but
-// produced an error sets IsError — per the MCP spec this is returned as a
-// tools/call result with isError:true, not a JSON-RPC protocol error.
-type CallResult struct {
-	Text    string
-	IsError bool
-}
-
-// Error lets the package's RPCError (see stdio.go) be returned by a resolver
-// to signal a protocol/pre-check failure with a specific JSON-RPC code; the
-// handler maps it straight to the JSON-RPC error.
-func (e *RPCError) Error() string { return e.Message }
-
-// ToolFunc executes a manually-registered tool. Return a *CallResult for tool
-// outcomes (set IsError for tool-level failures); return a non-nil error — an
-// *RPCError for a specific code — for protocol/pre-check failures.
-type ToolFunc func(ctx context.Context, args map[string]any) (*CallResult, error)
-
-// Resolver supplies the gateway's tools and executes calls. Swapping the
-// resolver changes where tools come from without touching the MCP protocol or
-// transport:
-//
-//   - NewManualResolver:   tools you register explicitly (full product control,
-//     including tools that are not go-micro services, executed via your own
-//     logic — auth, metering, …).
-//   - NewRegistryResolver: tools auto-discovered from registered services.
-//
-// The built-in store/broker tools are intentionally NOT exposed by any
-// resolver — they remain a development convenience on the legacy Serve() path.
-type Resolver interface {
-	// List returns the current tool catalog.
-	List(ctx context.Context) ([]Tool, error)
-	// Call executes a tool by name with JSON arguments.
-	Call(ctx context.Context, name string, args map[string]any) (*CallResult, error)
-}
-
-// ManualResolver exposes an explicitly-registered set of tools.
-type ManualResolver struct {
-	mu    sync.RWMutex
-	order []Tool
-	funcs map[string]ToolFunc
-}
-
-// NewManualResolver returns an empty manual resolver.
-func NewManualResolver() *ManualResolver {
-	return &ManualResolver{funcs: map[string]ToolFunc{}}
-}
-
-// Add registers (or replaces) a tool and its handler. Returns the resolver for
-// chaining.
-func (m *ManualResolver) Add(t Tool, fn ToolFunc) *ManualResolver {
-	m.mu.Lock()
-	defer m.mu.Unlock()
-	if _, ok := m.funcs[t.Name]; ok {
-		for i := range m.order {
-			if m.order[i].Name == t.Name {
-				m.order[i] = t
-			}
-		}
-	} else {
-		m.order = append(m.order, t)
-	}
-	m.funcs[t.Name] = fn
-	return m
-}
-
-// List returns the registered tools.
-func (m *ManualResolver) List(_ context.Context) ([]Tool, error) {
-	m.mu.RLock()
-	defer m.mu.RUnlock()
-	out := make([]Tool, len(m.order))
-	copy(out, m.order)
-	return out, nil
-}
-
-// Call runs the handler registered for name.
-func (m *ManualResolver) Call(ctx context.Context, name string, args map[string]any) (*CallResult, error) {
-	m.mu.RLock()
-	fn, ok := m.funcs[name]
-	m.mu.RUnlock()
-	if !ok {
-		return nil, &RPCError{Code: InvalidParams, Message: "Tool not found: " + name, Data: name}
-	}
-	return fn(ctx, args)
-}
-
-// RegistryResolver auto-discovers tools from registered go-micro services and
-// executes them over RPC. It exposes only services — never the internal
-// store/broker tools.
-type RegistryResolver struct {
-	tools *ai.Tools
-}
-
-// NewRegistryResolver discovers services from reg and calls them with cl.
-func NewRegistryResolver(reg registry.Registry, cl client.Client) *RegistryResolver {
-	return &RegistryResolver{tools: ai.NewTools(reg, ai.ToolClient(cl))}
-}
-
-// List discovers the current service tools.
-func (r *RegistryResolver) List(_ context.Context) ([]Tool, error) {
-	discovered, err := r.tools.Discover()
-	if err != nil {
-		return nil, err
-	}
-	out := make([]Tool, 0, len(discovered))
-	for _, t := range discovered {
-		out = append(out, Tool{
-			Name:        t.Name,
-			Description: t.Description,
-			InputSchema: map[string]interface{}{"type": "object", "properties": t.Properties},
-		})
-	}
-	return out, nil
-}
-
-// Call executes a discovered service tool.
-func (r *RegistryResolver) Call(ctx context.Context, name string, args map[string]any) (*CallResult, error) {
-	res := r.tools.Handler()(ctx, ai.ToolCall{ID: "1", Name: name, Input: args})
-	return &CallResult{Text: res.Content}, nil
-}

gateway/mcp/resolver_test.go

@@ -1,64 +0,0 @@
-package mcp
-
-import (
-	"context"
-	"encoding/json"
-	"net/http"
-	"net/http/httptest"
-	"strings"
-	"testing"
-)
-
-func TestManualResolverHandler(t *testing.T) {
-	res := NewManualResolver().
-		Add(Tool{Name: "echo", Description: "echoes text"},
-			func(_ context.Context, args map[string]interface{}) (*CallResult, error) {
-				s, _ := args["text"].(string)
-				return &CallResult{Text: "you said: " + s}, nil
-			}).
-		Add(Tool{Name: "boom", Description: "errors"},
-			func(_ context.Context, _ map[string]interface{}) (*CallResult, error) {
-				return &CallResult{Text: "kaboom", IsError: true}, nil
-			}).
-		Add(Tool{Name: "blocked", Description: "coded error"},
-			func(_ context.Context, _ map[string]interface{}) (*CallResult, error) {
-				return nil, &RPCError{Code: -32000, Message: "insufficient credits"}
-			})
-
-	ts := httptest.NewServer(NewHandler(res))
-	defer ts.Close()
-	rpc := func(body string) (int, map[string]interface{}) {
-		resp, err := http.Post(ts.URL, "application/json", strings.NewReader(body))
-		if err != nil {
-			t.Fatalf("post rpc: %v", err)
-		}
-		defer resp.Body.Close()
-		var out map[string]interface{}
-		json.NewDecoder(resp.Body).Decode(&out)
-		return resp.StatusCode, out
-	}
-
-	if _, out := rpc(`{"jsonrpc":"2.0","id":1,"method":"tools/list"}`); len(out["result"].(map[string]interface{})["tools"].([]interface{})) != 3 {
-		t.Fatalf("tools/list: %v", out)
-	}
-	// tool result
-	_, out := rpc(`{"jsonrpc":"2.0","id":2,"method":"tools/call","params":{"name":"echo","arguments":{"text":"hi"}}}`)
-	if out["result"].(map[string]interface{})["content"].([]interface{})[0].(map[string]interface{})["text"] != "you said: hi" {
-		t.Fatalf("echo: %v", out)
-	}
-	// tool-level error -> isError result, NOT protocol error
-	_, out = rpc(`{"jsonrpc":"2.0","id":3,"method":"tools/call","params":{"name":"boom","arguments":{}}}`)
-	if out["error"] != nil || out["result"].(map[string]interface{})["isError"] != true {
-		t.Fatalf("boom should be isError result: %v", out)
-	}
-	// coded protocol error -> JSON-RPC error with the code
-	_, out = rpc(`{"jsonrpc":"2.0","id":4,"method":"tools/call","params":{"name":"blocked","arguments":{}}}`)
-	if out["error"] == nil || int(out["error"].(map[string]interface{})["code"].(float64)) != -32000 {
-		t.Fatalf("blocked should be -32000: %v", out)
-	}
-	// notification -> 204, no body
-	code, _ := rpc(`{"jsonrpc":"2.0","method":"notifications/initialized"}`)
-	if code != http.StatusNoContent {
-		t.Fatalf("notification status = %d, want 204", code)
-	}
-}

internal/docs/AGENT_DESIGN.md

@@ -122,19 +122,6 @@ database "agent", table "{name}":
   history    — conversation history
 ```
 
-## Durable Ask / StreamAsk runs
-
-Agents can opt into the same checkpoint backend used by flows with
-`micro.AgentWithCheckpoint(...)`. When enabled, each `Ask` or `StreamAsk` run is
-persisted with its input, terminal status, response, and tool-call records. If
-the process or transport drops after a tool has completed but before the model
-returns a final answer, restart the agent with the same checkpoint store and
-call `micro.AgentResume(ctx, ag, runID)` or
-`micro.AgentResumeStreamAsk(ctx, ag, runID)`.
-Completed tool calls are served from the checkpoint instead of being executed
-again, while guardrails such as `MaxSteps`, loop detection, approval pauses, and
-`request_input` pauses continue to apply to the resumed run.
-
 ## Built-in Capabilities
 
 Beyond its scoped service tools, every agent gets two built-in tools. They are not service endpoints — they are capabilities the agent has over itself and over other agents. They are plain tools wired into the agent's tool handler; there is no separate harness, loop engine, or graph. The LLM calls them exactly like any other tool.

internal/docs/PRIORITIES.md

@@ -19,11 +19,10 @@ redirect the loop; direction always wins.
 items the loop can auto-merge): brand/positioning copy, breaking public-API
 changes, architectural rewrites. Those go to the human.
 
-## Work queue (ranked)
+## Developer experience (ranked)
 
-1. **CI-verify the 0→1 and 0→hero getting-started flows** ([#3354](https://github.com/micro/go-micro/issues/3354)) — the highest-value Now-phase hardening gap: the README and roadmap promise scaffold → run → call and a working multi-agent path, but the autonomous loop can still regress that developer contract unless it is executable in default CI. Define the smoke/e2e path with mock/local providers where possible, document the commands, and keep live-provider/network steps gated.
-2. **Add human-in-the-loop pause and resume for agent workflows** ([#3329](https://github.com/micro/go-micro/issues/3329)) — after memory compaction, tool-aware streaming, and durable `Ask`/`StreamAsk` checkpoints shipped, close the next operability gap for scheduled/looping work: durable pending-input states that let humans approve or provide context and then resume the same service/agent/workflow runtime.
-3. **gateway/a2a: multiple typed skills per agent card** ([#3342](https://github.com/micro/go-micro/issues/3342)) — the A2A gateway now has the full task lifecycle (send/stream/get/cancel/resubscribe, push config, multi-turn, input-required); the remaining gap is skill granularity. `Card` only ever advertises one synthetic "chat" skill with services flattened into tags. Let an agent advertise N typed skills and route per skill, so domain-routing agents expose their real capabilities over A2A through the gateway instead of a hand-rolled handler.
+1. **Expose run inspection in the CLI inner loop** ([#3297](https://github.com/micro/go-micro/issues/3297)) — the canon promises scaffold → run → chat → inspect → deploy, and recent work added run timelines, trace correlation, and a `micro runs` foothold; with conformance scheduling, failure/backoff hardening, the deploy contract, and the support reference now shipped, this is the highest-value remaining DX seam in the Now/ongoing roadmap until local agent/flow activity is documented and CI-tested as an actionable inspect step.
+2. **Add durable agent run checkpoint and resume** ([#3306](https://github.com/micro/go-micro/issues/3306)) — once the remaining inspection seam is closed, the highest-value Next-roadmap item is making agent loops resumable like flows so long-running work can survive restarts without unsafe replay or hidden state loss.
 
 _Seeded by Claude Code from the roadmap + open issues; thereafter maintained by the
 architecture-review pass._

internal/harness/zero-to-hero-ci/README.md

@@ -8,9 +8,8 @@ scripted so CI can run it on every push without external services or model keys.
 
 1. **Run** — `micro run` remains available as the local development entry point.
 2. **Chat** — `micro chat` remains available as the interactive agent entry point.
-3. **Inspect** — `micro inspect agent <name>` and `micro inspect flow <name>`
-   remain available as the local run-history inspection step, with `micro flow
-   runs` preserving durable workflow history inspection.
+3. **Inspect** — `micro flow runs` remains available for durable workflow run
+   history inspection.
 
 After the CLI boundary smoke checks, the script runs the deterministic harnesses
 that boot real services, agents, workflows, store-backed run history, and A2A

internal/website/docs/guides/agent-harness.md

@@ -36,46 +36,13 @@ your stack — the harness *is* the stack.
 | Discovery & RPC | Registry + client; agents and services find and call each other | Shipped |
 | Interop | MCP (tools), A2A (agents), x402 (paid tools) | Shipped |
 | Resilience | Per-call timeout with context propagation; opt-in retry/backoff (`ModelRetry`) across the loop | Shipped |
-| Durable runs | Checkpoint and resume an agent run with the same checkpoint backend flows use | Shipped |
+| Durable runs | Checkpoint and resume an agent run (flows already do) | In progress |
 | Observability | `RunInfo` → OpenTelemetry spans for runs, model calls, tools, delegation, and failures; persisted run history | Shipped |
 | Streaming | `ai.Stream` through chat, agent, and A2A | In progress |
 
 The "in progress" rows are exactly the roadmap's [Now and Next](/docs/roadmap.html),
 and the work is happening in the open.
 
-## Durable agent runs
-
-Agents can persist their execution history to the same `Checkpoint` backend as
-flows. A checkpointed `Ask` records the run id, original prompt, model result,
-and completed tool calls. If the process restarts after a tool succeeds but
-before the model finishes, `AgentResume` continues the same run and returns the
-recorded tool result instead of re-running the side effect. If a run already
-completed, resume returns the persisted response without calling the model.
-
-```go
-agent := micro.NewAgent("conductor",
-    micro.AgentProvider("anthropic"),
-    micro.AgentWithCheckpoint(checkpoint),
-)
-
-resp, err := agent.Ask(ctx, "charge order 42 and send a receipt")
-if err != nil {
-    // On startup, or after a transient failure, discover unfinished work:
-    pending, _ := micro.AgentPending(ctx, agent)
-    for _, run := range pending {
-        _, _ = micro.AgentResume(ctx, agent, run.ID)
-    }
-}
-_ = resp
-```
-
-For human-in-the-loop runs that pause through the built-in `request_input` tool,
-resume with the operator's response:
-
-```go
-_, err := micro.AgentResumeInput(ctx, agent, runID, "Deploy to us-east-1")
-```
-
 ## Observing agent runs
 
 Pass an OpenTelemetry tracer provider when you construct an agent to turn the

internal/website/docs/guides/agents-and-workflows.md

@@ -83,44 +83,6 @@ a := micro.NewAgent("conductor",
 a.Ask(ctx, "Plan the launch, create the tasks, and have comms notify the owner.")
 ```
 
-### Long-running memory
-
-Agents use store-backed conversation memory by default, scoped under the agent's
-name. That makes short restarts boring: the next `Ask` reloads the retained
-history from the same store backend you already use for services and flows.
-Long-running agents can also keep model context bounded without losing useful
-prior context:
-
-```go
-a := micro.NewAgent("conductor",
-    micro.AgentServices("task"),
-    micro.AgentProvider("anthropic"),
-    micro.AgentCompactMemory(40, 12),      // max active messages, recent messages kept verbatim
-    micro.AgentMemoryRecallLimit(5),       // archived turns recalled per Ask
-)
-```
-
-`AgentCompactMemory(maxMessages, keepRecent)` switches the default memory to a
-deterministic compactor. Once active history grows past `maxMessages`, older
-turns move into the durable archive, a provider-neutral summary is injected into
-active context, and the newest `keepRecent` messages stay verbatim. On future
-asks, archived turns whose text matches the current request are recalled ahead of
-the active context. The built-in retrieval is intentionally simple and
-credential-free for CI; teams that need embeddings or a vector database can still
-provide their own `AgentMemory` implementation.
-
-This is harness memory, not prompt-layer orchestration: services remain the
-capabilities, agents remain the dynamic decision makers, and flows remain the
-durable predefined paths. Compaction only keeps a scheduled or looping agent from
-turning every past turn into model context while still letting it remember facts
-that matter to the current service → agent → workflow run.
-
-Checkpointed agent runs and compacted memory share the same store-backed shape.
-If a provider call fails after the prompt has been recorded, `agent.Resume` uses
-the checkpointed run id and does not append that same user turn a second time;
-completed tool results and recalled archived memory remain available for the
-retry.
-
 ## The patterns — most are already here
 
 Anthropic lists five workflow patterns. Go Micro implements the two richest ones natively, as services and tools, and the rest are ordinary compositions:

internal/website/docs/guides/ai-provider-guide.md

@@ -39,12 +39,12 @@ The built-in providers currently register these capability interfaces:
 | Provider | Chat/text (`ai.Model`) | Image (`ai.ImageModel`) | Video (`ai.VideoModel`) | Streaming (`ai.Stream`) |
 | --- | --- | --- | --- | --- |
 | `anthropic` | Yes | No | No | No |
-| `atlascloud` | Yes | Yes | Yes | Yes |
+| `atlascloud` | Yes | Yes | Yes | No |
 | `gemini` | Yes | No | No | No |
-| `groq` | Yes | No | No | Yes |
-| `mistral` | Yes | No | No | Yes |
+| `groq` | Yes | No | No | No |
+| `mistral` | Yes | No | No | No |
 | `openai` | Yes | Yes | No | Yes |
-| `together` | Yes | No | No | Yes |
+| `together` | Yes | No | No | No |
 
 ## Step 1: Implement the `ai.Model` Interface
 

internal/website/docs/roadmap.md

@@ -36,13 +36,14 @@ The priority is that what exists works everywhere, under real conditions.
 
 ## Next — agentic depth
 
+- **Durable agent loop.** Flows resume; the agent's own loop does not yet. Reuse `Checkpoint` so a long-running agent survives a restart and continues.
 - **Streaming.** Broaden provider-backed `ai.Stream` coverage and keep chat plus A2A `message/stream` working end to end for real chat and long-task UX.
 - **Agent observability.** Wire the new `RunInfo` into OpenTelemetry spans so a run — steps, tool calls, delegation — is traceable. This is also what anyone running it in production will need.
 
 ## Later
 
 - **Memory management** — summarization and retrieval (RAG) beyond a fixed buffer.
-- **Human-in-the-loop** — broaden pause/resume UX around `input-required` runs and approvals.
+- **Human-in-the-loop** — pause and resume mid-run (`input-required`), beyond the binary `ApproveTool` gate.
 - **A2A** — richer live-stream reconnection (`tasks/resubscribe`) and `input-required` handoffs.
 
 ## Developer experience (ongoing)

micro.go

@@ -23,21 +23,12 @@ type Service = service.Service
 // Agent is the interface for an AI agent that manages services.
 type Agent = agent.Agent
 
-// AgentResponse is what an agent returns from Ask or a resumed run.
-type AgentResponse = agent.Response
-
-// AgentStream is a stream of tool execution events followed by final-answer chunks.
-type AgentStream = agent.AgentStream
-
 // AgentOption configures an Agent.
 type AgentOption = agent.Option
 
 // Flow is an event-driven LLM orchestration unit.
 type Flow = flow.Flow
 
-// FlowRun is a checkpointed flow or agent run record.
-type FlowRun = flow.Run
-
 // FlowOption configures a Flow.
 type FlowOption = flow.Option
 
@@ -173,32 +164,6 @@ func AgentWrapTool(w ...ai.ToolWrapper) AgentOption {
 // tool calls, delegation, and failures.
 func AgentTraceProvider(tp trace.TracerProvider) AgentOption { return agent.TraceProvider(tp) }
 
-// AgentWithCheckpoint sets the durability backend for agent Ask runs.
-// It uses the same Checkpoint interface as flows so services, agents,
-// and workflows can share one execution history backend.
-func AgentWithCheckpoint(c Checkpoint) AgentOption { return agent.WithCheckpoint(c) }
-
-// AgentPending returns checkpointed agent runs that have not completed.
-// Use it at process startup to discover agent work that should be resumed.
-func AgentPending(ctx context.Context, a Agent) ([]FlowRun, error) { return agent.Pending(ctx, a) }
-
-// AgentResume resumes a checkpointed agent run by id. Completed runs return
-// the persisted response without calling the model or replaying tool calls.
-func AgentResume(ctx context.Context, a Agent, runID string) (*AgentResponse, error) {
-	return agent.Resume(ctx, a, runID)
-}
-
-// AgentResumeInput resumes a checkpointed agent run waiting for human input.
-func AgentResumeInput(ctx context.Context, a Agent, runID, input string) (*AgentResponse, error) {
-	return agent.ResumeInput(ctx, a, runID, input)
-}
-
-// AgentResumeStreamAsk resumes a checkpointed agent run by id and streams the
-// resulting tool events and final answer.
-func AgentResumeStreamAsk(ctx context.Context, a Agent, runID string) (AgentStream, error) {
-	return agent.ResumeStreamAsk(ctx, a, runID)
-}
-
 // NewFlow creates an event-driven LLM orchestration unit.
 //
 //	f := micro.NewFlow("onboard-user",