docs(priorities): refresh architect queue

Co-authored-by: Codex <codex@openai.com>

.github/workflows/harness.yml

@@ -27,14 +27,8 @@ jobs:
           cache: true
       - name: Build
         run: go build ./...
-      - name: 0→1 scaffold contract
-        run: go test ./cmd/micro/cli/new -run TestZeroToOneContract -count=1
-      - name: Universe end-to-end (asserts; exits non-zero on failure)
-        run: go run ./internal/harness/universe
-      - name: Agent-flow harness
-        run: go run ./internal/harness/agent-flow
-      - name: 0→hero run/chat/inspect reference scenario
-        run: ./internal/harness/zero-to-hero-ci/run.sh
+      - name: 0→1 and 0→hero developer-flow harness
+        run: make harness
 
   harness-live:
     name: Provider harnesses (live LLM conformance)
@@ -50,17 +44,6 @@ jobs:
         with:
           go-version: stable
           cache: true
-      - name: Agent provider conformance matrix
-        env:
-          GO_MICRO_AGENT_CONFORMANCE_LIVE: "1"
-          ANTHROPIC_API_KEY: ${{ secrets.ANTHROPIC_API_KEY }}
-          OPENAI_API_KEY: ${{ secrets.OPENAI_API_KEY }}
-          GEMINI_API_KEY: ${{ secrets.GEMINI_API_KEY }}
-          GROQ_API_KEY: ${{ secrets.GROQ_API_KEY }}
-          MISTRAL_API_KEY: ${{ secrets.MISTRAL_API_KEY }}
-          TOGETHER_API_KEY: ${{ secrets.TOGETHER_API_KEY }}
-          ATLASCLOUD_API_KEY: ${{ secrets.ATLASCLOUD_API_KEY }}
-        run: go test ./agent -run TestAgentProviderConformanceMatrix -count=1 -v
       - name: Provider conformance against live models
         env:
           ANTHROPIC_API_KEY: ${{ secrets.ANTHROPIC_API_KEY }}

Makefile

@@ -18,7 +18,7 @@ help:
 	@echo "  make test-race     - Run tests with race detector"
 	@echo "  make test-coverage - Run tests with coverage"
 	@echo "  make lint          - Run linter"
-	@echo "  make harness       - Run deterministic end-to-end harnesses"
+	@echo "  make harness       - Run deterministic getting-started and end-to-end harnesses"
 	@echo "  make provider-conformance - Run harnesses against configured live providers"
 	@echo "  make fmt           - Format code"
 	@echo "  make install-tools - Install development tools"
@@ -42,12 +42,15 @@ test-coverage:
 	go tool cover -html=coverage.out -o coverage.html
 	@echo "Coverage report: coverage.html"
 
-# Run the end-to-end harnesses (deterministic, mock LLM — no API key).
-# The universe harness exits non-zero on assertion failure.
+# Run the documented getting-started contracts plus the deterministic
+# services → agents → workflows harnesses (mock LLM — no API key).
+# This mirrors the default CI path so local dogfooding catches scaffold,
+# run/chat/inspect, and 0→hero regressions before a PR is opened.
 harness:
-	go run ./internal/harness/universe
+	go test ./cmd/micro/cli/new -run TestZeroToOne -count=1
+	./internal/harness/zero-to-hero-ci/run.sh
 	go run ./internal/harness/agent-flow
-	go run ./internal/harness/plan-delegate # 0→hero: services + agents + flow + plan/delegate
+	go run ./internal/harness/provider-conformance -providers mock
 
 # Run the same harnesses against every configured live provider. Providers
 # without API keys are skipped; configured providers must pass.

README.md

@@ -64,6 +64,14 @@ curl -X POST http://localhost:8080/api/helloworld/Helloworld.Call \
   -H 'Content-Type: application/json' -d '{"name":"World"}'
 ```
 
+This scaffold → run → call path is covered by the no-secret CI harness. To run
+the same local contract (including the 0→hero services → agents → workflows path,
+chat/inspect CLI boundaries, and deploy dry-run), use:
+
+```bash
+make harness
+```
+
 ### Generate from a prompt — with an LLM key
 
 Set a provider key, describe what you want, and the AI designs services, writes handlers, compiles, and starts them:

agent/agent.go

@@ -139,19 +139,35 @@ 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 != "" {
 		modelOpts = append(modelOpts, ai.WithModel(a.opts.Model))
 	}
 
-	a.tools = ai.NewTools(a.opts.Registry, ai.ToolClient(a.opts.Client))
-	modelOpts = append(modelOpts, ai.WithToolHandler(a.toolHandler()))
+	// Reuse the existing tools instance: its name map is populated by
+	// discoverTools, and rebuilding it here would orphan a base handler that
+	// already captured the old instance (breaking StreamAsk tool resolution).
+	if a.tools == nil {
+		a.tools = ai.NewTools(a.opts.Registry, ai.ToolClient(a.opts.Client))
+	}
+	if handler == nil {
+		handler = a.toolHandler()
+	}
+	modelOpts = append(modelOpts, ai.WithToolHandler(handler))
 	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.
@@ -225,16 +241,18 @@ func (a *agentImpl) ask(ctx context.Context, message, parentRunID string) (*Resp
 		a.setup()
 	}
 
-	return a.askLocked(ctx, uuid.New().String(), message, parentRunID, nil)
+	return a.askLocked(ctx, uuid.New().String(), message, parentRunID, nil, true)
 }
 
-func (a *agentImpl) askLocked(ctx context.Context, runID, message, parentRunID string, existing *flow.Run) (*Response, error) {
+func (a *agentImpl) askLocked(ctx context.Context, runID, message, parentRunID string, existing *flow.Run, addUserMessage bool) (*Response, error) {
 	toolList, err := a.discoverTools()
 	if err != nil {
 		return nil, fmt.Errorf("discover tools: %w", err)
 	}
 
-	a.mem.Add("user", message)
+	if addUserMessage {
+		a.mem.Add("user", message)
+	}
 	a.steps = 0
 	a.calls = map[string]int{}
 	a.pause = nil

agent/builtin.go

@@ -113,13 +113,14 @@ func Builtins(opts ...Option) (tools []ai.Tool, handle func(name string, input m
 // prevents runaway recursion).
 func (a *agentImpl) toolHandler() ai.ToolHandler {
 	if a.ephemeral {
-		return a.tools.Handler()
+		return a.toolTimeoutWrap(a.tools.Handler())
 	}
 
 	// Innermost first: base, then guardrails (approve → loop → step →
 	// plan), then developer wrappers outermost. Wrapping reverses order,
 	// so the result runs plan → step → loop → approve → checkpoint → base.
 	h := a.baseHandler()
+	h = a.toolTimeoutWrap(h)
 	h = a.checkpointToolWrap(h)
 	h = a.approveWrap(h)
 	h = a.loopWrap(h)
@@ -148,6 +149,21 @@ func contextWrap(next ai.ToolHandler) ai.ToolHandler {
 	}
 }
 
+// toolTimeoutWrap gives each tool execution its own deadline while preserving
+// caller cancellation. Handlers still execute synchronously; tools that honor
+// context (custom tools, delegate RPC/A2A, and go-micro RPC clients) return
+// promptly with a bounded error result when the deadline expires.
+func (a *agentImpl) toolTimeoutWrap(next ai.ToolHandler) ai.ToolHandler {
+	return func(ctx context.Context, call ai.ToolCall) ai.ToolResult {
+		if a.opts.ToolTimeout <= 0 {
+			return next(ctx, call)
+		}
+		toolCtx, cancel := context.WithTimeout(ctx, a.opts.ToolTimeout)
+		defer cancel()
+		return next(toolCtx, call)
+	}
+}
+
 // baseHandler executes a tool call: a developer custom tool, the built-in
 // delegate, or an RPC to the service. It is the innermost handler.
 func (a *agentImpl) baseHandler() ai.ToolHandler {
@@ -319,6 +335,9 @@ func (a *agentImpl) handleDelegate(ctx context.Context, call ai.ToolCall) ai.Too
 		WithRegistry(a.opts.Registry),
 		WithClient(a.opts.Client),
 		WithStore(a.opts.Store),
+		ModelCallTimeout(a.opts.ModelTimeout),
+		ModelRetry(a.opts.ModelMaxAttempts, a.opts.ModelRetryBackoff),
+		ToolCallTimeout(a.opts.ToolTimeout),
 		TraceProvider(a.opts.TraceProvider),
 	)
 	// Record lineage so the sub-agent's tool calls carry this run as parent.

agent/checkpoint.go

@@ -88,6 +88,9 @@ func (a *agentImpl) resume(ctx context.Context, runID string) (*Response, error)
 		}
 		return &resp, nil
 	}
+	if terminalAgentRunStatus(run.Status) {
+		return nil, fmt.Errorf("agent run %s is terminal with status %q", runID, run.Status)
+	}
 	message := string(run.State.Data)
 	parentID := run.ParentID
 	a.mu.Lock()
@@ -95,7 +98,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)
+	return a.askLocked(ctx, run.ID, message, parentID, &run, false)
 }
 
 // ResumeInput resumes a checkpointed agent run that paused via the built-in
@@ -140,7 +143,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)
+	return a.askLocked(ctx, run.ID, message, run.ParentID, &run, true)
 }
 
 func (a *agentImpl) pending(ctx context.Context) ([]flow.Run, error) {
@@ -153,13 +156,22 @@ func (a *agentImpl) pending(ctx context.Context) ([]flow.Run, error) {
 	}
 	out := runs[:0]
 	for _, run := range runs {
-		if run.Flow == a.opts.Name && run.Status != "done" {
+		if run.Flow == a.opts.Name && !terminalAgentRunStatus(run.Status) {
 			out = append(out, run)
 		}
 	}
 	return out, nil
 }
 
+func terminalAgentRunStatus(status string) bool {
+	switch status {
+	case "done", "canceled", "expired":
+		return true
+	default:
+		return false
+	}
+}
+
 func (a *agentImpl) checkpointToolWrap(next ai.ToolHandler) ai.ToolHandler {
 	return func(ctx context.Context, call ai.ToolCall) ai.ToolResult {
 		if a.opts.Checkpoint == nil || a.currentRun == nil {

agent/checkpoint_test.go

@@ -13,7 +13,7 @@ import (
 
 func TestResumeCompletedCheckpointDoesNotReplayModel(t *testing.T) {
 	ctx := context.Background()
-	cp := flow.StoreCheckpoint(store.NewStore(), "durable-agent")
+	cp := flow.StoreCheckpoint(store.NewMemoryStore(), "durable-agent")
 	calls := 0
 	fakeGen = func(ctx context.Context, opts ai.Options, req *ai.Request) (*ai.Response, error) {
 		calls++
@@ -52,7 +52,7 @@ func TestResumeCompletedCheckpointDoesNotReplayModel(t *testing.T) {
 
 func TestResumeFailedCheckpointDoesNotReplayCompletedTool(t *testing.T) {
 	ctx := context.Background()
-	cp := flow.StoreCheckpoint(store.NewStore(), "tool-resume-agent")
+	cp := flow.StoreCheckpoint(store.NewMemoryStore(), "tool-resume-agent")
 	toolRuns := 0
 	first := true
 	fakeGen = func(ctx context.Context, opts ai.Options, req *ai.Request) (*ai.Response, error) {
@@ -102,9 +102,75 @@ 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")
+	cp := flow.StoreCheckpoint(store.NewMemoryStore(), "pending-agent")
 	run := flow.Run{ID: "run-1", Flow: "pending-agent", Status: "failed", State: flow.State{Stage: agentAskStep, Data: []byte("retry me")}}
 	if err := cp.Save(ctx, run); err != nil {
 		t.Fatalf("Save: %v", err)
@@ -119,9 +185,38 @@ func TestPendingReturnsUnfinishedAgentRuns(t *testing.T) {
 	}
 }
 
+func TestPendingSkipsTerminalCanceledAndExpiredAgentRuns(t *testing.T) {
+	ctx := context.Background()
+	cp := flow.StoreCheckpoint(store.NewMemoryStore(), "terminal-agent")
+	for _, run := range []flow.Run{
+		{ID: "active", Flow: "terminal-agent", Status: "failed", State: flow.State{Stage: agentAskStep, Data: []byte("retry me")}},
+		{ID: "done", Flow: "terminal-agent", Status: "done", State: flow.State{Stage: agentAskStep, Data: []byte("done")}},
+		{ID: "canceled", Flow: "terminal-agent", Status: "canceled", State: flow.State{Stage: agentAskStep, Data: []byte("canceled")}},
+		{ID: "expired", Flow: "terminal-agent", Status: "expired", State: flow.State{Stage: agentAskStep, Data: []byte("expired")}},
+	} {
+		if err := cp.Save(ctx, run); err != nil {
+			t.Fatalf("Save(%s): %v", run.ID, err)
+		}
+	}
+
+	a := newTestAgent(Name("terminal-agent"), WithCheckpoint(cp))
+	runs, err := Pending(ctx, a)
+	if err != nil {
+		t.Fatalf("Pending: %v", err)
+	}
+	if len(runs) != 1 || runs[0].ID != "active" {
+		t.Fatalf("Pending = %#v, want only active failed run", runs)
+	}
+	for _, id := range []string{"canceled", "expired"} {
+		if _, err := Resume(ctx, a, id); err == nil || !strings.Contains(err.Error(), "terminal") {
+			t.Fatalf("Resume(%s) err = %v, want terminal status error", id, err)
+		}
+	}
+}
+
 func TestHumanInputPauseResumesSameRunWithInput(t *testing.T) {
 	ctx := context.Background()
-	cp := flow.StoreCheckpoint(store.NewStore(), "input-agent")
+	cp := flow.StoreCheckpoint(store.NewMemoryStore(), "input-agent")
 	calls := 0
 	fakeGen = func(ctx context.Context, opts ai.Options, req *ai.Request) (*ai.Response, error) {
 		calls++
@@ -177,9 +272,54 @@ func TestHumanInputPauseResumesSameRunWithInput(t *testing.T) {
 	}
 }
 
+func TestHumanInputResumeHonorsCanceledContextAndLeavesRunPending(t *testing.T) {
+	ctx := context.Background()
+	cp := flow.StoreCheckpoint(store.NewMemoryStore(), "input-cancel-agent")
+	fakeGen = func(ctx context.Context, opts ai.Options, req *ai.Request) (*ai.Response, error) {
+		if opts.ToolHandler != nil {
+			opts.ToolHandler(ctx, ai.ToolCall{ID: "input-1", Name: toolHumanInput, Input: map[string]any{"prompt": "Approve deploy?"}})
+		}
+		return &ai.Response{Reply: "waiting"}, nil
+	}
+	defer func() { fakeGen = nil }()
+
+	a := newTestAgent(Name("input-cancel-agent"), WithCheckpoint(cp))
+	if _, err := a.Ask(ctx, "deploy the service"); err == nil {
+		t.Fatal("Ask succeeded, want input-required pause")
+	}
+	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: %#v", len(runs), runs)
+	}
+
+	canceled, cancel := context.WithCancel(ctx)
+	cancel()
+	if _, err := ResumeInput(canceled, a, runs[0].ID, "yes"); !errors.Is(err, context.Canceled) {
+		t.Fatalf("ResumeInput canceled err = %v, want context.Canceled", err)
+	}
+
+	loaded, ok, err := cp.Load(ctx, runs[0].ID)
+	if err != nil || !ok {
+		t.Fatalf("Load paused run ok=%v err=%v", ok, err)
+	}
+	if loaded.Status != "paused" || loaded.State.Stage != agentInputStep {
+		t.Fatalf("run status/stage after canceled resume = %s/%s, want paused/%s", loaded.Status, loaded.State.Stage, agentInputStep)
+	}
+	var pause inputPause
+	if err := loaded.State.Scan(&pause); err != nil {
+		t.Fatalf("Scan pause after canceled resume: %v", err)
+	}
+	if pause.OriginalMessage != "deploy the service" || pause.Prompt != "Approve deploy?" {
+		t.Fatalf("pause after canceled resume = %#v", pause)
+	}
+}
+
 func TestApprovalDenialPausesCheckpointedRunAndResumeContinues(t *testing.T) {
 	ctx := context.Background()
-	cp := flow.StoreCheckpoint(store.NewStore(), "approval-agent")
+	cp := flow.StoreCheckpoint(store.NewMemoryStore(), "approval-agent")
 	calls := 0
 	fakeGen = func(ctx context.Context, opts ai.Options, req *ai.Request) (*ai.Response, error) {
 		calls++

agent/conformance_test.go

@@ -33,14 +33,29 @@ func TestAgentProviderConformanceMatrix(t *testing.T) {
 		{name: "together", key: "TOGETHER_API_KEY", model: "GO_MICRO_CONFORMANCE_TOGETHER_MODEL", live: true},
 	}
 
+	selected := selectedConformanceProviders(os.Getenv("GO_MICRO_AGENT_CONFORMANCE_PROVIDERS"))
 	for _, provider := range providers {
 		provider := provider
+		if len(selected) > 0 && !selected[provider.name] {
+			continue
+		}
 		t.Run(provider.name, func(t *testing.T) {
 			runAgentConformanceScenario(t, provider)
 		})
 	}
 }
 
+func selectedConformanceProviders(csv string) map[string]bool {
+	out := map[string]bool{}
+	for _, part := range strings.Split(csv, ",") {
+		part = strings.TrimSpace(part)
+		if part != "" {
+			out[part] = true
+		}
+	}
+	return out
+}
+
 func runAgentConformanceScenario(t *testing.T, provider conformanceProvider) {
 	t.Helper()
 	if provider.live {

agent/options.go

@@ -56,6 +56,10 @@ type Options struct {
 	// ModelRetryBackoff is the base delay between transient provider failures
 	// (grows exponentially per attempt when retries are enabled).
 	ModelRetryBackoff time.Duration
+	// ToolTimeout bounds each tool execution (0 disables). The timeout is
+	// applied before custom tools, delegate, and service RPC calls so context
+	// deadlines propagate consistently through the agent loop.
+	ToolTimeout time.Duration
 
 	// Memory is the agent's conversation memory. Nil = the default
 	// store-backed memory (durable across restarts).
@@ -91,6 +95,11 @@ type Options struct {
 	// and tool calls. Nil disables instrumentation.
 	TraceProvider trace.TracerProvider
 
+	// TraceInputs controls whether agent observability records include raw
+	// user messages. It is false by default so spans and persisted run
+	// timelines carry correlation and shape without leaking prompts.
+	TraceInputs bool
+
 	// tools are developer-registered custom tools (see WithTool).
 	tools []customTool
 	// wrappers are developer-registered tool-execution wrappers
@@ -107,6 +116,7 @@ func newOptions(opts ...Option) Options {
 		ModelTimeout:      30 * time.Second,
 		ModelMaxAttempts:  1, // retries opt-in via ModelRetry (see field doc)
 		ModelRetryBackoff: 100 * time.Millisecond,
+		ToolTimeout:       30 * time.Second,
 		// On by default and lenient: identical repeated calls are a
 		// no-progress loop, never useful. Set LoopLimit(0) to disable.
 		LoopLimit: 3,
@@ -199,6 +209,14 @@ func ModelCallTimeout(d time.Duration) Option {
 	return func(o *Options) { o.ModelTimeout = d }
 }
 
+// ToolCallTimeout sets the timeout for each tool execution. It bounds custom
+// tools, built-in delegate calls, and service RPC tools with the same context
+// deadline so mid-run cancellation and slow tools produce safe error results
+// instead of unbounded agent runs. Set 0 to disable.
+func ToolCallTimeout(d time.Duration) Option {
+	return func(o *Options) { o.ToolTimeout = d }
+}
+
 // ModelRetry sets the provider retry budget and backoff for transient failures.
 func ModelRetry(maxAttempts int, backoff time.Duration) Option {
 	return func(o *Options) {
@@ -295,3 +313,11 @@ func WithTool(name, description string, properties map[string]any, handler ToolF
 func TraceProvider(tp trace.TracerProvider) Option {
 	return func(o *Options) { o.TraceProvider = tp }
 }
+
+// TraceInputs opts in to recording raw user messages on agent run events.
+// By default inputs are redacted from OpenTelemetry spans and persisted run
+// timelines; use this only when the observability backend is approved to store
+// prompt content.
+func TraceInputs(enabled bool) Option {
+	return func(o *Options) { o.TraceInputs = enabled }
+}

agent/otel.go

@@ -31,27 +31,35 @@ 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"
 	AttrRefusal        = "agent.refusal"
+	AttrInputChars     = "agent.input.chars"
+	AttrErrorKind      = "agent.error.kind"
 )
 
 type RunEvent struct {
-	Time      time.Time `json:"time"`
-	RunID     string    `json:"run_id"`
-	ParentID  string    `json:"parent_id,omitempty"`
-	TraceID   string    `json:"trace_id,omitempty"`
-	SpanID    string    `json:"span_id,omitempty"`
-	Agent     string    `json:"agent"`
-	Kind      string    `json:"kind"`
-	Name      string    `json:"name,omitempty"`
-	Provider  string    `json:"provider,omitempty"`
-	Model     string    `json:"model,omitempty"`
-	LatencyMS int64     `json:"latency_ms,omitempty"`
-	Tokens    Usage     `json:"tokens,omitempty"`
-	Refused   string    `json:"refused,omitempty"`
-	Error     string    `json:"error,omitempty"`
+	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"`
+	ErrorKind   string    `json:"error_kind,omitempty"`
+	InputChars  int       `json:"input_chars,omitempty"`
 }
 
 type Usage = ai.Usage
@@ -60,7 +68,8 @@ type Usage = ai.Usage
 // Zero values preserve the full deterministic run list.
 type RunListOptions struct {
 	// Status, when set, keeps only runs with the matching status
-	// (for example "running", "done", "error", or "refused").
+	// (for example "running", "done", "canceled", "timeout",
+	// "rate_limited", "error", or "refused").
 	Status string
 	// TraceID, when set, keeps only runs correlated with this trace id.
 	// A prefix is accepted so operators can paste the shortened trace id
@@ -73,18 +82,19 @@ type RunListOptions struct {
 
 // RunSummary is a compact index entry for a recorded agent run.
 type RunSummary struct {
-	RunID      string    `json:"run_id"`
-	Agent      string    `json:"agent"`
-	ParentID   string    `json:"parent_id,omitempty"`
-	TraceID    string    `json:"trace_id,omitempty"`
-	SpanID     string    `json:"span_id,omitempty"`
-	StartedAt  time.Time `json:"started_at"`
-	UpdatedAt  time.Time `json:"updated_at"`
-	DurationMS int64     `json:"duration_ms,omitempty"`
-	Events     int       `json:"events"`
-	Status     string    `json:"status,omitempty"`
-	LastKind   string    `json:"last_kind,omitempty"`
-	LastError  string    `json:"last_error,omitempty"`
+	RunID         string    `json:"run_id"`
+	Agent         string    `json:"agent"`
+	ParentID      string    `json:"parent_id,omitempty"`
+	TraceID       string    `json:"trace_id,omitempty"`
+	SpanID        string    `json:"span_id,omitempty"`
+	StartedAt     time.Time `json:"started_at"`
+	UpdatedAt     time.Time `json:"updated_at"`
+	DurationMS    int64     `json:"duration_ms,omitempty"`
+	Events        int       `json:"events"`
+	Status        string    `json:"status,omitempty"`
+	LastKind      string    `json:"last_kind,omitempty"`
+	LastError     string    `json:"last_error,omitempty"`
+	LastErrorKind string    `json:"last_error_kind,omitempty"`
 }
 
 func (a *agentImpl) tracer() trace.Tracer {
@@ -94,13 +104,17 @@ func (a *agentImpl) tracer() trace.Tracer {
 func (a *agentImpl) startRun(ctx context.Context, message string) (context.Context, func(error)) {
 	info, _ := ai.RunInfoFrom(ctx)
 	start := time.Now()
+	runEvent := RunEvent{Time: start, RunID: info.RunID, ParentID: info.ParentID, Agent: info.Agent, Kind: "run", InputChars: len(message)}
+	if a.opts.TraceInputs {
+		runEvent.Name = message
+	}
 
 	if a.opts.TraceProvider == nil {
-		a.recordRunEvent(RunEvent{Time: start, RunID: info.RunID, ParentID: info.ParentID, Agent: info.Agent, Kind: "run", Name: message})
+		a.recordRunEvent(runEvent)
 		return ctx, func(err error) {
 			latency := time.Since(start).Milliseconds()
 			if err != nil {
-				a.recordRunEvent(RunEvent{Time: time.Now(), RunID: info.RunID, ParentID: info.ParentID, Agent: info.Agent, Kind: "error", LatencyMS: latency, Error: err.Error()})
+				a.recordRunEvent(RunEvent{Time: time.Now(), RunID: info.RunID, ParentID: info.ParentID, Agent: info.Agent, Kind: "error", LatencyMS: latency, Error: err.Error(), ErrorKind: string(ai.ClassifyError(err))})
 				return
 			}
 			a.recordRunEvent(RunEvent{Time: time.Now(), RunID: info.RunID, ParentID: info.ParentID, Agent: info.Agent, Kind: "done", LatencyMS: latency})
@@ -109,14 +123,15 @@ func (a *agentImpl) startRun(ctx context.Context, message string) (context.Conte
 
 	ctx, span := a.tracer().Start(ctx, spanNameRun, trace.WithSpanKind(trace.SpanKindInternal), trace.WithAttributes(
 		attribute.String(AttrRunID, info.RunID), attribute.String(AttrParentRunID, info.ParentID), attribute.String(AttrAgentName, info.Agent)))
-	a.recordSpanEvent(span, RunEvent{Time: start, RunID: info.RunID, ParentID: info.ParentID, Agent: info.Agent, Kind: "run", Name: message})
+	a.recordSpanEvent(span, runEvent)
 	return ctx, func(err error) {
 		latency := time.Since(start).Milliseconds()
 		span.SetAttributes(attribute.Int64(AttrLatencyMS, latency))
 		if err != nil {
+			span.SetAttributes(attribute.String(AttrErrorKind, string(ai.ClassifyError(err))))
 			span.RecordError(err)
 			span.SetStatus(codes.Error, err.Error())
-			a.recordSpanEvent(span, RunEvent{Time: time.Now(), RunID: info.RunID, ParentID: info.ParentID, Agent: info.Agent, Kind: "error", LatencyMS: latency, Error: err.Error()})
+			a.recordSpanEvent(span, RunEvent{Time: time.Now(), RunID: info.RunID, ParentID: info.ParentID, Agent: info.Agent, Kind: "error", LatencyMS: latency, Error: err.Error(), ErrorKind: string(ai.ClassifyError(err))})
 		} else {
 			span.SetStatus(codes.Ok, "")
 			a.recordSpanEvent(span, RunEvent{Time: time.Now(), RunID: info.RunID, ParentID: info.ParentID, Agent: info.Agent, Kind: "done", LatencyMS: latency})
@@ -144,9 +159,10 @@ 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, LatencyMS: dur, Tokens: 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}
 		if err != nil {
 			e.Error = err.Error()
+			e.ErrorKind = string(ai.ClassifyError(err))
 		}
 		m.a.recordRunEvent(e)
 		return resp, err
@@ -162,6 +178,12 @@ 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
@@ -169,15 +191,17 @@ func (m *tracedModel) Generate(ctx context.Context, req *ai.Request, opts ...ai.
 	}
 	span.SetAttributes(attrs...)
 	if err != nil {
+		span.SetAttributes(attribute.String(AttrErrorKind, string(ai.ClassifyError(err))))
 		span.RecordError(err)
 		span.SetStatus(codes.Error, err.Error())
 	} else {
 		span.SetStatus(codes.Ok, "")
 	}
 	span.End()
-	e := RunEvent{Time: time.Now(), RunID: info.RunID, ParentID: info.ParentID, Agent: info.Agent, Kind: "model", Provider: provider, Model: model, LatencyMS: dur, Tokens: usage}
+	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}
 	if err != nil {
 		e.Error = err.Error()
+		e.ErrorKind = string(ai.ClassifyError(err))
 	}
 	m.a.recordSpanEvent(span, e)
 	return resp, err
@@ -204,7 +228,8 @@ func (a *agentImpl) traceTool(next ai.ToolHandler) ai.ToolHandler {
 		if a.opts.TraceProvider == nil {
 			res := next(ctx, call)
 			dur := time.Since(start).Milliseconds()
-			a.recordRunEvent(RunEvent{Time: time.Now(), RunID: info.RunID, ParentID: info.ParentID, Agent: info.Agent, Kind: "tool", Name: call.Name, LatencyMS: dur, Refused: res.Refused, Error: resultError(res)})
+			resErr := resultError(res)
+			a.recordRunEvent(RunEvent{Time: time.Now(), RunID: info.RunID, ParentID: info.ParentID, Agent: info.Agent, Kind: "tool", Name: call.Name, LatencyMS: dur, Refused: res.Refused, Error: resErr, ErrorKind: classifyToolError(resErr)})
 			return res
 		}
 
@@ -221,8 +246,11 @@ func (a *agentImpl) traceTool(next ai.ToolHandler) ai.ToolHandler {
 		if res.Refused != "" {
 			attrs = append(attrs, attribute.Bool(AttrGuardrailBlock, true), attribute.String(AttrRefusal, res.Refused))
 		}
-		span.SetAttributes(attrs...)
 		resErr := resultError(res)
+		if kind := classifyToolError(resErr); kind != "" {
+			attrs = append(attrs, attribute.String(AttrErrorKind, kind))
+		}
+		span.SetAttributes(attrs...)
 		if res.Refused != "" {
 			span.SetStatus(codes.Error, res.Refused)
 		} else if resErr != "" {
@@ -231,7 +259,7 @@ func (a *agentImpl) traceTool(next ai.ToolHandler) ai.ToolHandler {
 			span.SetStatus(codes.Ok, "")
 		}
 		span.End()
-		a.recordSpanEvent(span, RunEvent{Time: time.Now(), RunID: info.RunID, ParentID: info.ParentID, Agent: info.Agent, Kind: "tool", Name: call.Name, LatencyMS: dur, Refused: res.Refused, Error: resErr})
+		a.recordSpanEvent(span, RunEvent{Time: time.Now(), RunID: info.RunID, ParentID: info.ParentID, Agent: info.Agent, Kind: "tool", Name: call.Name, LatencyMS: dur, Refused: res.Refused, Error: resErr, ErrorKind: classifyToolError(resErr)})
 		return res
 	}
 }
@@ -248,6 +276,19 @@ func resultError(res ai.ToolResult) string {
 	return ""
 }
 
+func classifyToolError(err string) string {
+	switch {
+	case err == "":
+		return ""
+	case strings.Contains(strings.ToLower(err), "context canceled"):
+		return string(ai.ErrorKindCanceled)
+	case strings.Contains(strings.ToLower(err), "deadline exceeded"):
+		return string(ai.ErrorKindTimeout)
+	default:
+		return string(ai.ErrorKindProvider)
+	}
+}
+
 func (a *agentImpl) recordSpanEvent(span trace.Span, e RunEvent) {
 	if sc := span.SpanContext(); sc.IsValid() {
 		e.TraceID = sc.TraceID().String()
@@ -274,9 +315,18 @@ 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))
 	}
+	if e.InputChars > 0 {
+		attrs = append(attrs, attribute.Int(AttrInputChars, e.InputChars))
+	}
 	attrs = appendUsage(attrs, e.Tokens)
 	if e.Refused != "" {
 		attrs = append(attrs, attribute.Bool(AttrGuardrailBlock, true), attribute.String(AttrRefusal, e.Refused))
@@ -284,6 +334,9 @@ func runEventAttributes(e RunEvent) []attribute.KeyValue {
 	if e.Error != "" {
 		attrs = append(attrs, attribute.String("agent.error", e.Error))
 	}
+	if e.ErrorKind != "" {
+		attrs = append(attrs, attribute.String(AttrErrorKind, e.ErrorKind))
+	}
 	return attrs
 }
 
@@ -363,6 +416,9 @@ func ListRunSummariesWithOptions(s store.Store, agentName string, opts RunListOp
 			if e.Error != "" {
 				summary.LastError = e.Error
 			}
+			if e.ErrorKind != "" {
+				summary.LastErrorKind = e.ErrorKind
+			}
 		}
 		if opts.Status != "" && summary.Status != opts.Status {
 			continue
@@ -389,24 +445,32 @@ func runStatus(events []RunEvent) string {
 	}
 	status := "running"
 	for _, e := range events {
-		if e.Error != "" {
-			status = "error"
-		}
-		if e.Refused != "" && status != "error" {
+		if e.Refused != "" && status == "running" {
 			status = "refused"
 		}
-		switch e.Kind {
-		case "error":
-			status = "error"
-		case "done":
-			if status == "running" {
-				status = "done"
-			}
+		if e.Error != "" || e.Kind == "error" {
+			status = runErrorStatus(e.ErrorKind)
+		}
+		if e.Kind == "done" && status == "running" {
+			status = "done"
 		}
 	}
 	return status
 }
 
+func runErrorStatus(kind string) string {
+	switch ai.ErrorKind(kind) {
+	case ai.ErrorKindCanceled:
+		return "canceled"
+	case ai.ErrorKindTimeout:
+		return "timeout"
+	case ai.ErrorKindRateLimited:
+		return "rate_limited"
+	default:
+		return "error"
+	}
+}
+
 func LoadRunEvents(s store.Store, agentName, runID string) ([]RunEvent, error) {
 	st := store.Scope(s, "agent", agentName)
 	keys, err := st.List(store.ListPrefix("runs/" + runID + "/"))

agent/otel_test.go

@@ -3,6 +3,7 @@ package agent
 import (
 	"context"
 	"encoding/json"
+	"errors"
 	"fmt"
 	"strings"
 	"testing"
@@ -11,10 +12,13 @@ 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 {
@@ -89,6 +93,9 @@ 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 {
@@ -125,6 +132,146 @@ func TestAgentOpenTelemetrySpans(t *testing.T) {
 	}
 }
 
+func TestAgentRunObservabilityRedactsInputByDefault(t *testing.T) {
+	secret := "deploy production with token sk-secret"
+	exp := tracetest.NewInMemoryExporter()
+	tp := trace.NewTracerProvider(trace.WithSyncer(exp))
+	st := store.NewMemoryStore()
+	a := New(Name("redactor"), Provider("oteltest"), WithStore(st), TraceProvider(tp))
+	if _, err := a.Ask(context.Background(), secret); err != nil {
+		t.Fatal(err)
+	}
+
+	spans := exp.GetSpans().Snapshots()
+	var sawInputChars bool
+	for _, s := range spans {
+		for _, event := range s.Events() {
+			attrs := spanAttributes(event.Attributes)
+			if attrs["agent.event.name"] == secret {
+				t.Fatalf("span event leaked raw input: %#v", event)
+			}
+			if attrs[AttrInputChars] == fmt.Sprint(len(secret)) {
+				sawInputChars = true
+			}
+		}
+	}
+	if !sawInputChars {
+		t.Fatal("run event missing redacted input length attribute")
+	}
+
+	summaries, err := ListRunSummaries(st, "redactor")
+	if err != nil {
+		t.Fatal(err)
+	}
+	events, err := LoadRunEvents(st, "redactor", summaries[0].RunID)
+	if err != nil {
+		t.Fatal(err)
+	}
+	for _, event := range events {
+		if event.Name == secret {
+			t.Fatalf("persisted run event leaked raw input: %#v", event)
+		}
+		if event.Kind == "run" && event.InputChars != len(secret) {
+			t.Fatalf("run event InputChars = %d, want %d", event.InputChars, len(secret))
+		}
+	}
+}
+
+func TestAgentTraceInputsOptInRecordsInput(t *testing.T) {
+	message := "operator-approved diagnostic prompt"
+	st := store.NewMemoryStore()
+	a := New(Name("input-opt-in"), Provider("oteltest"), WithStore(st), TraceInputs(true))
+	if _, err := a.Ask(context.Background(), message); err != nil {
+		t.Fatal(err)
+	}
+
+	summaries, err := ListRunSummaries(st, "input-opt-in")
+	if err != nil {
+		t.Fatal(err)
+	}
+	events, err := LoadRunEvents(st, "input-opt-in", summaries[0].RunID)
+	if err != nil {
+		t.Fatal(err)
+	}
+	for _, event := range events {
+		if event.Kind == "run" && event.Name == message {
+			return
+		}
+	}
+	t.Fatalf("opt-in run event did not record message: %#v", events)
+}
+
+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" && attrs[AttrErrorKind] == string(ai.ErrorKindUnknown) && 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 != "" && event.ErrorKind == string(ai.ErrorKindUnknown) {
+			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 {
@@ -282,7 +429,7 @@ func TestListRunSummaries(t *testing.T) {
 		{Time: time.Unix(0, 1), RunID: "run-a", Agent: "runner", TraceID: "trace-a", SpanID: "span-a", Kind: "run", Name: "first"},
 		{Time: time.Unix(0, 2), RunID: "run-a", Agent: "runner", Kind: "tool", Name: "probe"},
 		{Time: time.Unix(0, 3), RunID: "run-b", Agent: "runner", ParentID: "parent", Kind: "run", Name: "second"},
-		{Time: time.Unix(0, 4), RunID: "run-b", Agent: "runner", ParentID: "parent", Kind: "error", Error: "boom"},
+		{Time: time.Unix(0, 4), RunID: "run-b", Agent: "runner", ParentID: "parent", Kind: "error", Error: "context deadline exceeded", ErrorKind: string(ai.ErrorKindTimeout)},
 	}
 	for _, e := range events {
 		b, err := json.Marshal(e)
@@ -305,11 +452,35 @@ func TestListRunSummaries(t *testing.T) {
 	if got[0].RunID != "run-a" || got[0].TraceID != "trace-a" || got[0].SpanID != "span-a" || got[0].Events != 2 || got[0].Status != "running" || got[0].DurationMS != 0 || got[0].LastKind != "tool" || !got[0].UpdatedAt.Equal(time.Unix(0, 2)) {
 		t.Fatalf("unexpected run-a summary: %#v", got[0])
 	}
-	if got[1].RunID != "run-b" || got[1].ParentID != "parent" || got[1].Events != 2 || got[1].Status != "error" || got[1].DurationMS != 0 || got[1].LastKind != "error" || got[1].LastError != "boom" {
+	if got[1].RunID != "run-b" || got[1].ParentID != "parent" || got[1].Events != 2 || got[1].Status != "timeout" || got[1].DurationMS != 0 || got[1].LastKind != "error" || got[1].LastError != "context deadline exceeded" || got[1].LastErrorKind != string(ai.ErrorKindTimeout) {
 		t.Fatalf("unexpected run-b summary: %#v", got[1])
 	}
 }
 
+func TestRunStatusClassifiesOperationalErrorKinds(t *testing.T) {
+	tests := []struct {
+		name string
+		kind ai.ErrorKind
+		want string
+	}{
+		{name: "canceled", kind: ai.ErrorKindCanceled, want: "canceled"},
+		{name: "timeout", kind: ai.ErrorKindTimeout, want: "timeout"},
+		{name: "rate limited", kind: ai.ErrorKindRateLimited, want: "rate_limited"},
+		{name: "provider", kind: ai.ErrorKindProvider, want: "error"},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			got := runStatus([]RunEvent{
+				{Kind: "run"},
+				{Kind: "error", Error: "failed", ErrorKind: string(tt.kind)},
+			})
+			if got != tt.want {
+				t.Fatalf("runStatus() = %q, want %q", got, tt.want)
+			}
+		})
+	}
+}
+
 func TestListRunSummariesWithOptionsFiltersAndLimits(t *testing.T) {
 	st := store.NewMemoryStore()
 	scoped := store.Scope(st, "agent", "runner")
@@ -317,7 +488,7 @@ func TestListRunSummariesWithOptionsFiltersAndLimits(t *testing.T) {
 		{Time: time.Unix(0, 1), RunID: "run-old", Agent: "runner", Kind: "run"},
 		{Time: time.Unix(0, 2), RunID: "run-old", Agent: "runner", Kind: "done"},
 		{Time: time.Unix(0, 3), RunID: "run-new", Agent: "runner", TraceID: "abcdef1234567890", Kind: "run"},
-		{Time: time.Unix(0, 4), RunID: "run-new", Agent: "runner", Kind: "error", Error: "boom"},
+		{Time: time.Unix(0, 4), RunID: "run-new", Agent: "runner", Kind: "error", Error: "rate limit exceeded", ErrorKind: string(ai.ErrorKindRateLimited)},
 	}
 	for _, e := range events {
 		b, err := json.Marshal(e)
@@ -329,11 +500,11 @@ func TestListRunSummariesWithOptionsFiltersAndLimits(t *testing.T) {
 		}
 	}
 
-	got, err := ListRunSummariesWithOptions(st, "runner", RunListOptions{Status: "error", TraceID: "abcdef", Limit: 1})
+	got, err := ListRunSummariesWithOptions(st, "runner", RunListOptions{Status: "rate_limited", TraceID: "abcdef", Limit: 1})
 	if err != nil {
 		t.Fatal(err)
 	}
-	if len(got) != 1 || got[0].RunID != "run-new" || got[0].Status != "error" {
+	if len(got) != 1 || got[0].RunID != "run-new" || got[0].Status != "rate_limited" {
 		t.Fatalf("filtered summaries = %#v", got)
 	}
 }

agent/resilience_test.go

@@ -102,3 +102,30 @@ func TestCanceledAskContextSkipsToolExecution(t *testing.T) {
 		t.Fatalf("plan persisted after canceled tool context: %q", plan)
 	}
 }
+
+func TestToolCallTimeoutPropagatesDeadlineToCustomTool(t *testing.T) {
+	var sawDeadline bool
+	a := newTestAgent(
+		Name("tool-timeout"),
+		ToolCallTimeout(10*time.Millisecond),
+		WithTool("slow", "slow tool", nil, func(ctx context.Context, input map[string]any) (string, error) {
+			if _, ok := ctx.Deadline(); ok {
+				sawDeadline = true
+			}
+			<-ctx.Done()
+			return "", ctx.Err()
+		}),
+	)
+
+	start := time.Now()
+	content := toolContent(a.toolHandler(), "slow", nil)
+	if !sawDeadline {
+		t.Fatal("custom tool did not receive a deadline")
+	}
+	if !strings.Contains(content, context.DeadlineExceeded.Error()) {
+		t.Fatalf("tool result = %q, want deadline exceeded", content)
+	}
+	if elapsed := time.Since(start); elapsed > 200*time.Millisecond {
+		t.Fatalf("tool call took %s, want bounded timeout", elapsed)
+	}
+}

agent/stream.go

@@ -0,0 +1,240 @@
+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
+	}
+	if terminalAgentRunStatus(run.Status) {
+		return nil, errors.New("agent: checkpointed run is terminal with status " + run.Status)
+	}
+
+	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

@@ -0,0 +1,175 @@
+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,6 +43,7 @@ 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,7 +2,11 @@ package atlascloud
 
 import (
 	"context"
+	"encoding/json"
 	"errors"
+	"io"
+	"net/http"
+	"net/http/httptest"
 	"testing"
 
 	"go-micro.dev/v6/ai"
@@ -81,16 +85,58 @@ func TestProvider_Generate_NoAPIKey(t *testing.T) {
 	}
 }
 
-func TestProvider_Stream_NotImplemented(t *testing.T) {
-	p := NewProvider()
+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()
 
-	req := &ai.Request{
-		Prompt: "Hello",
+	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")
 	}
 
-	_, err := p.Stream(context.Background(), req)
-	if !errors.Is(err, ai.ErrStreamingUnsupported) {
-		t.Fatalf("Stream error = %v, want ErrStreamingUnsupported", err)
+	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)
 	}
 }
 

ai/capabilities_test.go

@@ -34,7 +34,7 @@ func TestRegisteredProviders(t *testing.T) {
 	}
 
 	got = ai.RegisteredProviders("stream")
-	want = []string{"openai"}
+	want = []string{"atlascloud", "groq", "mistral", "openai", "together"}
 	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}},
+		{Provider: "atlascloud", Capabilities: ai.Capabilities{Model: true, Image: true, Video: true, Stream: true}},
 		{Provider: "gemini", Capabilities: ai.Capabilities{Model: true}},
-		{Provider: "groq", Capabilities: ai.Capabilities{Model: true}},
-		{Provider: "mistral", Capabilities: ai.Capabilities{Model: true}},
+		{Provider: "groq", Capabilities: ai.Capabilities{Model: true, Stream: true}},
+		{Provider: "mistral", Capabilities: ai.Capabilities{Model: true, Stream: true}},
 		{Provider: "openai", Capabilities: ai.Capabilities{Model: true, Image: true, Stream: true}},
-		{Provider: "together", Capabilities: ai.Capabilities{Model: true}},
+		{Provider: "together", Capabilities: ai.Capabilities{Model: true, Stream: 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}) {
+	if caps := ai.ProviderCapabilities("atlascloud"); caps != (ai.Capabilities{Model: true, Image: true, Video: true, Stream: 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{"openai", "test-stream"}
+	want := []string{"atlascloud", "groq", "mistral", "openai", "test-stream", "together"}
 	if !reflect.DeepEqual(got, want) {
 		t.Fatalf("RegisteredProviders(stream) = %#v, want %#v", got, want)
 	}

ai/groq/groq.go

@@ -22,12 +22,14 @@ 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 {
@@ -119,7 +121,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 nil, fmt.Errorf("%w: groq provider", ai.ErrStreamingUnsupported)
+	return openaiapi.Stream(ctx, p.opts, req, "/v1/chat/completions")
 }
 
 func (p *Provider) callAPI(ctx context.Context, req map[string]any) (*ai.Response, map[string]any, error) {

ai/groq/groq_test.go

@@ -2,7 +2,11 @@ package groq
 
 import (
 	"context"
+	"encoding/json"
 	"errors"
+	"io"
+	"net/http"
+	"net/http/httptest"
 	"testing"
 
 	"go-micro.dev/v6/ai"
@@ -40,9 +44,44 @@ func TestProvider_Generate_NoAPIKey(t *testing.T) {
 	}
 }
 
-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)
+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)
 	}
 }
 

ai/internal/openaiapi/stream.go

@@ -0,0 +1,117 @@
+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,12 +22,14 @@ 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 {
@@ -119,7 +121,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 nil, fmt.Errorf("%w: mistral provider", ai.ErrStreamingUnsupported)
+	return openaiapi.Stream(ctx, p.opts, req, "/v1/chat/completions")
 }
 
 func (p *Provider) callAPI(ctx context.Context, req map[string]any) (*ai.Response, map[string]any, error) {

ai/mistral/mistral_test.go

@@ -2,7 +2,11 @@ package mistral
 
 import (
 	"context"
+	"encoding/json"
 	"errors"
+	"io"
+	"net/http"
+	"net/http/httptest"
 	"testing"
 
 	"go-micro.dev/v6/ai"
@@ -40,9 +44,44 @@ func TestProvider_Generate_NoAPIKey(t *testing.T) {
 	}
 }
 
-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)
+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)
 	}
 }
 

ai/stream_conformance_test.go

@@ -0,0 +1,203 @@
+package ai_test
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"io"
+	"net/http"
+	"net/http/httptest"
+	"reflect"
+	"strings"
+	"testing"
+	"time"
+
+	"go-micro.dev/v6/ai"
+	_ "go-micro.dev/v6/ai/anthropic"
+	_ "go-micro.dev/v6/ai/atlascloud"
+	_ "go-micro.dev/v6/ai/gemini"
+	_ "go-micro.dev/v6/ai/groq"
+	_ "go-micro.dev/v6/ai/mistral"
+	_ "go-micro.dev/v6/ai/openai"
+	_ "go-micro.dev/v6/ai/together"
+)
+
+func TestStreamProvidersConformToOpenAICompatibleSSE(t *testing.T) {
+	providers := conformingStreamProviders(t)
+
+	for _, provider := range providers {
+		provider := provider
+		t.Run(provider, func(t *testing.T) {
+			var sawRequest bool
+			ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+				sawRequest = true
+				if r.URL.Path != "/v1/chat/completions" {
+					t.Fatalf("path = %s, want /v1/chat/completions", r.URL.Path)
+				}
+				if got := r.Header.Get("Accept"); got != "text/event-stream" {
+					t.Fatalf("Accept = %q, want text/event-stream", got)
+				}
+				if got := r.Header.Get("Authorization"); got != "Bearer test-key" {
+					t.Fatalf("Authorization = %q, want bearer API key", got)
+				}
+
+				var body map[string]any
+				if err := json.NewDecoder(r.Body).Decode(&body); err != nil {
+					t.Fatalf("decode request: %v", err)
+				}
+				if body["model"] == "" {
+					t.Fatal("request omitted model")
+				}
+				if body["stream"] != true {
+					t.Fatalf("stream = %#v, want true", body["stream"])
+				}
+				streamOptions, ok := body["stream_options"].(map[string]any)
+				if !ok || streamOptions["include_usage"] != true {
+					t.Fatalf("stream_options = %#v, want include_usage=true", body["stream_options"])
+				}
+				messages, ok := body["messages"].([]any)
+				if !ok || len(messages) != 4 {
+					t.Fatalf("messages = %#v, want system + history + prompt", body["messages"])
+				}
+				wantRoles := []string{"system", "user", "assistant", "user"}
+				for i, wantRole := range wantRoles {
+					message, ok := messages[i].(map[string]any)
+					if !ok || message["role"] != wantRole {
+						t.Fatalf("message[%d] = %#v, want role %q", i, messages[i], wantRole)
+					}
+				}
+
+				w.Header().Set("Content-Type", "text/event-stream")
+				_, _ = w.Write([]byte(": keepalive\n\n"))
+				_, _ = w.Write([]byte("event: ignored\n\n"))
+				_, _ = 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\":3,\"completion_tokens\":2,\"total_tokens\":5}}\n\n"))
+				_, _ = w.Write([]byte("data: [DONE]\n\n"))
+			}))
+			defer ts.Close()
+
+			model := ai.New(provider, ai.WithAPIKey("test-key"), ai.WithBaseURL(ts.URL))
+			if model == nil {
+				t.Fatalf("ai.New(%q) returned nil", provider)
+			}
+			stream, err := model.Stream(context.Background(), &ai.Request{
+				SystemPrompt: "system",
+				Messages: []ai.Message{
+					{Role: "user", Content: "previous question"},
+					{Role: "assistant", Content: "previous answer"},
+				},
+				Prompt: "current question",
+			})
+			if err != nil {
+				t.Fatalf("Stream returned error: %v", err)
+			}
+			defer stream.Close()
+			if !sawRequest {
+				t.Fatal("server did not receive stream request")
+			}
+
+			assertStreamReply(t, stream, "hel")
+			assertStreamReply(t, stream, "lo")
+			usage, err := stream.Recv()
+			if err != nil {
+				t.Fatalf("usage chunk error: %v", err)
+			}
+			if usage.Reply != "" || usage.Usage != (ai.Usage{InputTokens: 3, OutputTokens: 2, TotalTokens: 5}) {
+				t.Fatalf("usage chunk = %#v", usage)
+			}
+			if _, err := stream.Recv(); !errors.Is(err, io.EOF) {
+				t.Fatalf("final error = %v, want EOF", err)
+			}
+		})
+	}
+}
+
+func TestStreamProvidersCloseCancelsInFlightRequest(t *testing.T) {
+	for _, provider := range conformingStreamProviders(t) {
+		provider := provider
+		t.Run(provider, func(t *testing.T) {
+			released := make(chan struct{})
+			ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+				w.Header().Set("Content-Type", "text/event-stream")
+				_, _ = w.Write([]byte("data: {\"choices\":[{\"delta\":{\"content\":\"hel\"}}]}\n\n"))
+				if f, ok := w.(http.Flusher); ok {
+					f.Flush()
+				}
+				<-r.Context().Done()
+				close(released)
+			}))
+			defer ts.Close()
+
+			stream, err := ai.New(provider, ai.WithAPIKey("test-key"), ai.WithBaseURL(ts.URL)).Stream(context.Background(), &ai.Request{Prompt: "Hello"})
+			if err != nil {
+				t.Fatalf("Stream returned error: %v", err)
+			}
+			assertStreamReply(t, stream, "hel")
+			if err := stream.Close(); err != nil {
+				t.Fatalf("Close returned error: %v", err)
+			}
+			if err := stream.Close(); err != nil {
+				t.Fatalf("second Close returned error: %v", err)
+			}
+
+			select {
+			case <-released:
+			case <-time.After(time.Second):
+				t.Fatal("server did not observe canceled stream request")
+			}
+		})
+	}
+}
+
+func TestUnsupportedProvidersReturnStreamingUnsupportedAndStayUnregistered(t *testing.T) {
+	for _, provider := range []string{"anthropic", "gemini"} {
+		provider := provider
+		t.Run(provider, func(t *testing.T) {
+			if caps := ai.ProviderCapabilities(provider); caps.Stream {
+				t.Fatalf("ProviderCapabilities(%q).Stream = true, want false", provider)
+			}
+			_, err := ai.New(provider, ai.WithAPIKey("test-key")).Stream(context.Background(), &ai.Request{Prompt: "Hello"})
+			if !errors.Is(err, ai.ErrStreamingUnsupported) {
+				t.Fatalf("Stream error = %v, want ErrStreamingUnsupported", err)
+			}
+			if err != nil && strings.Contains(err.Error(), "test-key") {
+				t.Fatal("streaming unsupported error leaked API key")
+			}
+		})
+	}
+}
+
+func conformingStreamProviders(t *testing.T) []string {
+	t.Helper()
+	providers := ai.RegisteredProviders("stream")
+	allowed := map[string]struct{}{
+		"atlascloud": {},
+		"groq":       {},
+		"mistral":    {},
+		"openai":     {},
+		"together":   {},
+	}
+	var out []string
+	for _, provider := range providers {
+		if _, ok := allowed[provider]; ok {
+			out = append(out, provider)
+		}
+	}
+	want := []string{"atlascloud", "groq", "mistral", "openai", "together"}
+	if !reflect.DeepEqual(out, want) {
+		t.Fatalf("conforming stream providers = %#v, want %#v (registered stream providers: %#v)", out, want, providers)
+	}
+	return out
+}
+
+func assertStreamReply(t *testing.T, stream ai.Stream, want string) {
+	t.Helper()
+	chunk, err := stream.Recv()
+	if err != nil {
+		t.Fatalf("Recv error = %v, want reply %q", err, want)
+	}
+	if chunk.Reply != want {
+		t.Fatalf("Reply = %q, want %q", chunk.Reply, want)
+	}
+}

ai/together/together.go

@@ -22,12 +22,14 @@ 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 {
@@ -119,7 +121,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 nil, fmt.Errorf("%w: together provider", ai.ErrStreamingUnsupported)
+	return openaiapi.Stream(ctx, p.opts, req, "/v1/chat/completions")
 }
 
 func (p *Provider) callAPI(ctx context.Context, req map[string]any) (*ai.Response, map[string]any, error) {

ai/together/together_test.go

@@ -2,7 +2,11 @@ package together
 
 import (
 	"context"
+	"encoding/json"
 	"errors"
+	"io"
+	"net/http"
+	"net/http/httptest"
 	"testing"
 
 	"go-micro.dev/v6/ai"
@@ -40,9 +44,44 @@ func TestProvider_Generate_NoAPIKey(t *testing.T) {
 	}
 }
 
-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)
+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)
 	}
 }
 

cmd/micro/cli/deploy/deploy.go

@@ -43,6 +43,9 @@ func Deploy(c *cli.Context) error {
 	}
 
 	target, remotePath := resolveDeployTarget(c, target, cfg)
+	if c.Bool("dry-run") {
+		return printDeployPlan(c, target, cfg, remotePath)
+	}
 
 	return deploySSH(c, target, cfg, remotePath)
 }
@@ -96,6 +99,81 @@ func showDeployTargets(cfg *config.Config) error {
 	return fmt.Errorf("%s", sb.String())
 }
 
+func printDeployPlan(c *cli.Context, target string, cfg *config.Config, remotePath string) error {
+	dir := c.Args().Get(1)
+	if dir == "" {
+		dir = "."
+	}
+
+	absDir, err := filepath.Abs(dir)
+	if err != nil {
+		return fmt.Errorf("failed to get absolute path: %w", err)
+	}
+
+	if cfg == nil {
+		cfg, _ = config.Load(absDir)
+	}
+	if remotePath == "" {
+		remotePath = defaultRemotePath
+	}
+
+	services, err := deployServices(absDir, cfg, c.String("service"))
+	if err != nil {
+		return err
+	}
+
+	fmt.Println()
+	fmt.Println("  \033[1mmicro deploy --dry-run\033[0m")
+	fmt.Println()
+	fmt.Printf("  Target      \033[36m%s\033[0m\n", target)
+	fmt.Printf("  Remote path %s\n", remotePath)
+	fmt.Printf("  Services    %s\n", strings.Join(services, ", "))
+	fmt.Println()
+	fmt.Println("  Plan:")
+	fmt.Println("    1. Build linux/amd64 service binaries")
+	fmt.Printf("    2. Copy binaries to %s/bin/\n", remotePath)
+	fmt.Println("    3. Enable and restart micro@<service> systemd units")
+	fmt.Println("    4. Check service health")
+	fmt.Println()
+	fmt.Println("  No SSH, rsync, systemd, or remote deployment was performed.")
+	return nil
+}
+
+func deployServices(absDir string, cfg *config.Config, filterService string) ([]string, error) {
+	if filterService != "" && cfg != nil {
+		found := false
+		for _, svc := range cfg.Services {
+			if svc.Name == filterService {
+				found = true
+				break
+			}
+		}
+		if !found && len(cfg.Services) > 0 {
+			return nil, fmt.Errorf("service '%s' not found in configuration", filterService)
+		}
+	}
+
+	if cfg != nil && len(cfg.Services) > 0 {
+		sorted, err := cfg.TopologicalSort()
+		if err != nil {
+			return nil, err
+		}
+		services := make([]string, 0, len(sorted))
+		for _, svc := range sorted {
+			if filterService == "" || svc.Name == filterService {
+				services = append(services, svc.Name)
+			}
+		}
+		return services, nil
+	}
+
+	services := []string{filepath.Base(absDir)}
+	if filterService != "" && filterService != services[0] {
+		return nil, fmt.Errorf("service '%s' not found (only '%s' available)", filterService, services[0])
+	}
+	return services, nil
+}
+
 func deploySSH(c *cli.Context, target string, cfg *config.Config, remotePath string) error {
 	dir := c.Args().Get(1)
 	if dir == "" {
@@ -121,19 +199,10 @@ func deploySSH(c *cli.Context, target string, cfg *config.Config, remotePath str
 	fmt.Println()
 	fmt.Printf("  Target     \033[36m%s\033[0m\n\n", target)
 
-	// Early validation: Check if the requested service exists before SSH checks
-	filterService := c.String("service")
-	if filterService != "" && cfg != nil {
-		found := false
-		for _, svc := range cfg.Services {
-			if svc.Name == filterService {
-				found = true
-				break
-			}
-		}
-		if !found && len(cfg.Services) > 0 {
-			return fmt.Errorf("service '%s' not found in configuration", filterService)
-		}
+	// Early validation: resolve services before SSH checks.
+	services, err := deployServices(absDir, cfg, c.String("service"))
+	if err != nil {
+		return err
 	}
 
 	// Step 1: Check SSH connectivity
@@ -153,28 +222,6 @@ func deploySSH(c *cli.Context, target string, cfg *config.Config, remotePath str
 	fmt.Println("\u2713")
 
 	// Step 3: Build binaries
-	var services []string
-	if cfg != nil && len(cfg.Services) > 0 {
-		sorted, err := cfg.TopologicalSort()
-		if err != nil {
-			return err
-		}
-		for _, svc := range sorted {
-			// If --service flag is provided, only include that service
-			if filterService == "" || svc.Name == filterService {
-				services = append(services, svc.Name)
-			}
-		}
-	} else {
-		// Single service project
-		services = []string{filepath.Base(absDir)}
-
-		// If --service flag was provided for a single-service project, validate it matches
-		if filterService != "" && filterService != services[0] {
-			return fmt.Errorf("service '%s' not found (only '%s' available)", filterService, services[0])
-		}
-	}
-
 	fmt.Printf("  Building binaries...       ")
 	if err := buildBinaries(absDir, cfg, c.Bool("build"), services); err != nil {
 		fmt.Println("\u2717")
@@ -493,6 +540,10 @@ The deploy process:
 				Name:  "service",
 				Usage: "Deploy only a specific service (for multi-service projects)",
 			},
+			&cli.BoolFlag{
+				Name:  "dry-run",
+				Usage: "Print the deployment plan without building, connecting, copying, or restarting services",
+			},
 		},
 	})
 }

cmd/micro/cli/deploy/deploy_test.go

@@ -17,6 +17,7 @@ func newDeployTestContext(t *testing.T, args ...string) *cli.Context {
 	set.String("ssh", "", "")
 	set.String("service", "", "")
 	set.Bool("build", false, "")
+	set.Bool("dry-run", false, "")
 	if err := set.Parse(args); err != nil {
 		t.Fatalf("parse flags: %v", err)
 	}
@@ -118,3 +119,49 @@ deploy prod
 		t.Fatalf("deploy target = %#v", prod)
 	}
 }
+
+func TestDeployDryRunPlansConfiguredTargetWithoutRemoteSideEffects(t *testing.T) {
+	dir := t.TempDir()
+	if err := os.WriteFile(dir+"/micro.mu", []byte(`service api
+    path ./api
+
+deploy prod
+    ssh deploy@prod.example.com
+    path /srv/micro
+`), 0644); err != nil {
+		t.Fatalf("write config: %v", err)
+	}
+
+	oldwd, err := os.Getwd()
+	if err != nil {
+		t.Fatalf("getwd: %v", err)
+	}
+	if err := os.Chdir(dir); err != nil {
+		t.Fatalf("chdir: %v", err)
+	}
+	t.Cleanup(func() {
+		if err := os.Chdir(oldwd); err != nil {
+			t.Errorf("restore cwd: %v", err)
+		}
+	})
+
+	ctx := newDeployTestContext(t, "--dry-run", "prod")
+	if err := Deploy(ctx); err != nil {
+		t.Fatalf("dry-run deploy: %v", err)
+	}
+}
+
+func TestDeployDryRunValidatesRequestedService(t *testing.T) {
+	ctx := newDeployTestContext(t, "--dry-run", "--service", "missing", "prod")
+	cfg := &config.Config{Services: map[string]*config.Service{
+		"api": {Name: "api", Path: "./api"},
+	}}
+
+	err := printDeployPlan(ctx, "deploy@prod.example.com", cfg, defaultRemotePath)
+	if err == nil {
+		t.Fatal("expected dry-run to validate service names")
+	}
+	if !strings.Contains(err.Error(), "service 'missing' not found in configuration") {
+		t.Fatalf("unexpected error: %v", err)
+	}
+}

cmd/micro/inspect/inspect.go

@@ -0,0 +1,169 @@
+// 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

@@ -0,0 +1,64 @@
+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,6 +12,7 @@ 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/run/run.go

@@ -498,10 +498,13 @@ func printBanner(services []*serviceProcess, gw *server.Gateway, watching bool,
 		fmt.Printf("  Dashboard   \033[36mhttp://localhost%s\033[0m\n", gw.Addr())
 		fmt.Printf("  API         \033[36mhttp://localhost%s/api/{service}/{method}\033[0m\n", gw.Addr())
 		fmt.Printf("  Agent       \033[36mhttp://localhost%s/agent\033[0m\n", gw.Addr())
+		// MCP tools are served on the gateway by default — every endpoint is an
+		// AI-callable tool, so surface it rather than hiding it behind a flag.
+		fmt.Printf("  MCP Tools   \033[36mhttp://localhost%s/mcp/tools\033[0m\n", gw.Addr())
 		fmt.Printf("  Health      \033[36mhttp://localhost%s/health\033[0m\n", gw.Addr())
 		if mcpAddr != "" {
-			fmt.Printf("  MCP         \033[36mhttp://localhost%s\033[0m\n", mcpAddr)
-			fmt.Printf("  MCP Tools   \033[36mhttp://localhost%s/mcp/tools\033[0m\n", mcpAddr)
+			// Optional standalone MCP protocol server (e.g. for MCP clients).
+			fmt.Printf("  MCP Server  \033[36mhttp://localhost%s\033[0m (full MCP protocol)\n", mcpAddr)
 			fmt.Printf("  WebSocket   \033[36mws://localhost%s/mcp/ws\033[0m\n", mcpAddr)
 		}
 	}

cmd/micro/zero_to_hero_cli_test.go

@@ -19,7 +19,7 @@ func TestZeroToHeroCLIBoundaries(t *testing.T) {
 		}
 	}
 
-	for _, want := range []string{"run", "chat", "flow"} {
+	for _, want := range []string{"run", "chat", "flow", "inspect", "deploy"} {
 		if !commands[want] {
 			t.Fatalf("missing %q command", want)
 		}
@@ -27,4 +27,24 @@ 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")
+	}
+
+	var hasDeployDryRun bool
+	for _, command := range microcmd.DefaultCmd.App().Commands {
+		if command.Name != "deploy" {
+			continue
+		}
+		for _, flag := range command.Flags {
+			for _, name := range flag.Names() {
+				if name == "dry-run" {
+					hasDeployDryRun = true
+				}
+			}
+		}
+	}
+	if !hasDeployDryRun {
+		t.Fatal("missing deploy boundary: deploy --dry-run")
+	}
 }

events/memory.go

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

events/options.go

@@ -1,11 +1,25 @@
 package events
 
-import "time"
+import (
+	"time"
 
-type Options struct{}
+	"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 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

examples/README.md

@@ -80,10 +80,11 @@ A workflow as ordered, checkpointed steps that survives a crash and resumes wher
 - **Checkpoint** — each step is persisted; on `Resume`, completed steps are not re-run (no duplicate side effects)
 
 ### [support](./support/)
-A real-world support desk — the "zero to hero" shape in one runnable file:
-- **services** (`customers`, `tickets`, `notify`) become the agent's tools automatically
-- **flow** turns a `ticket.created` event into work for the agent (the event is the prompt)
-- **guardrail** — the agent triages freely but can't email a customer without passing the approval gate
+A maintained 0-to-hero reference path in one runnable file:
+- **scaffold** typed `customers`, `tickets`, and `notify` services
+- **run/chat** with a support agent that uses those services as tools
+- **inspect** the event-driven `intake` flow and approval gate
+- **CI** keeps the deterministic mock-model journey runnable with `go test ./examples/support`
 
 ## Coming Soon
 

examples/agent-durable/README.md

@@ -0,0 +1,45 @@
+# Durable agent run resume
+
+This example shows the agent-side counterpart to `examples/flow-durable`: an
+agent run is checkpointed with the same `Checkpoint` interface used by flows,
+then resumed after an interruption without repeating a completed side effect.
+The sample uses an in-memory store to keep repeated local runs deterministic;
+use your service store for process-restart recovery.
+
+Run it with:
+
+```sh
+go run ./examples/agent-durable
+```
+
+The demo model calls `inventory.reserve`, then fails to mimic a process dying
+after the tool call was checkpointed. `micro.AgentPending` finds the unfinished
+run and `micro.AgentResume` continues it from the saved checkpoint. The final
+`tool executions: 1` line is the important bit: the reservation tool was not
+called a second time during resume.
+
+## When to use this instead of a durable flow
+
+Use a durable flow when the path is known ahead of time: ordered service calls,
+retries, timers, compensation, and a precise resume stage such as `reserve` or
+`charge`. Use a checkpointed agent run when the path is open-ended and the model
+may choose tools dynamically, but completed tool side effects still must not be
+replayed after a crash or provider failure.
+
+They compose: keep deterministic business process in `flow-durable`, then hand
+off the judgment-heavy step to a checkpointed agent when the workflow needs
+model-directed tool use. Both use the same `Checkpoint` backend, so inspection
+and recovery can share one run-history store.
+
+In a service, use the same pattern at startup:
+
+```go
+pending, _ := micro.AgentPending(ctx, agent)
+for _, run := range pending {
+    _, _ = micro.AgentResume(ctx, agent, run.ID)
+}
+```
+
+`context.Context` cancellation and deadlines are still honored by checkpoint
+loads/saves, model calls, and tool calls. Runs with terminal statuses such as
+`done`, `canceled`, and `expired` are not returned by `AgentPending`.

examples/agent-durable/main.go

@@ -0,0 +1,88 @@
+// Package main demonstrates durable agent runs: a checkpointed agent can
+// resume after a crash without re-executing completed tool calls.
+package main
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"sync/atomic"
+
+	micro "go-micro.dev/v6"
+	"go-micro.dev/v6/ai"
+	"go-micro.dev/v6/store"
+)
+
+func main() {
+	ctx := context.Background()
+	checkpoint := micro.StoreCheckpoint(store.NewMemoryStore(), "durable-agent-demo")
+	model := &demoModel{failFirst: true}
+	ai.Register("durable-demo", func(opts ...ai.Option) ai.Model {
+		_ = model.Init(opts...)
+		return model
+	})
+	var reservations atomic.Int32
+
+	ag := micro.NewAgent("durable-agent-demo",
+		micro.AgentWithCheckpoint(checkpoint),
+		micro.AgentProvider("durable-demo"),
+		micro.AgentTool("inventory.reserve", "reserve inventory exactly once", map[string]any{
+			"sku": map[string]any{"type": "string"},
+		}, func(ctx context.Context, input map[string]any) (string, error) {
+			count := reservations.Add(1)
+			return fmt.Sprintf("reserved %s (execution %d)", input["sku"], count), nil
+		}),
+	)
+
+	_, err := ag.Ask(ctx, "reserve sku-123 and confirm")
+	fmt.Println("initial run:", err)
+
+	pending, err := micro.AgentPending(ctx, ag)
+	if err != nil {
+		panic(err)
+	}
+	if len(pending) == 0 {
+		panic("expected a checkpointed run to resume")
+	}
+
+	resp, err := micro.AgentResume(ctx, ag, pending[0].ID)
+	if err != nil {
+		panic(err)
+	}
+	fmt.Println("resumed reply:", resp.Reply)
+	fmt.Println("tool executions:", reservations.Load())
+}
+
+type demoModel struct {
+	failFirst bool
+	opts      ai.Options
+}
+
+func (m *demoModel) Init(opts ...ai.Option) error {
+	m.opts = ai.NewOptions(opts...)
+	return nil
+}
+func (m *demoModel) Options() ai.Options { return m.opts }
+func (m *demoModel) String() string      { return "durable-demo" }
+
+func (m *demoModel) Generate(ctx context.Context, req *ai.Request, opts ...ai.GenerateOption) (*ai.Response, error) {
+	if m.opts.ToolHandler != nil {
+		res := m.opts.ToolHandler(ctx, ai.ToolCall{
+			ID:    "reserve-1",
+			Name:  "inventory.reserve",
+			Input: map[string]any{"sku": "sku-123"},
+		})
+		if res.Content == "" {
+			return nil, errors.New("reservation tool returned no content")
+		}
+	}
+	if m.failFirst {
+		m.failFirst = false
+		return nil, errors.New("simulated process interruption after checkpointed tool call")
+	}
+	return &ai.Response{Reply: "sku-123 is reserved; no duplicate reservation was made"}, nil
+}
+
+func (m *demoModel) Stream(context.Context, *ai.Request, ...ai.GenerateOption) (ai.Stream, error) {
+	return nil, ai.ErrStreamingUnsupported
+}

examples/agent-durable/main_test.go

@@ -0,0 +1,48 @@
+package main
+
+import (
+	"bytes"
+	"io"
+	"os"
+	"strings"
+	"testing"
+)
+
+func TestDurableAgentExampleResumesWithoutReplayingTool(t *testing.T) {
+	out := captureStdout(t, main)
+	if !strings.Contains(out, "simulated process interruption after checkpointed tool call") {
+		t.Fatalf("example output %q did not show the initial interrupted run", out)
+	}
+	if !strings.Contains(out, "resumed reply: sku-123 is reserved; no duplicate reservation was made") {
+		t.Fatalf("example output %q did not show the resumed response", out)
+	}
+	if !strings.Contains(out, "tool executions: 1") {
+		t.Fatalf("example output %q did not prove the tool was not replayed", out)
+	}
+}
+
+func captureStdout(t *testing.T, fn func()) string {
+	t.Helper()
+
+	old := os.Stdout
+	r, w, err := os.Pipe()
+	if err != nil {
+		t.Fatalf("pipe stdout: %v", err)
+	}
+	os.Stdout = w
+
+	var buf bytes.Buffer
+	done := make(chan struct{})
+	go func() {
+		_, _ = io.Copy(&buf, r)
+		close(done)
+	}()
+
+	fn()
+
+	_ = w.Close()
+	os.Stdout = old
+	<-done
+	_ = r.Close()
+	return buf.String()
+}

examples/agent-human-input/README.md

@@ -34,3 +34,11 @@ calls that tool, Go Micro persists the run with status `paused` and stage
 `input-required`. Plain `agent.Resume` continues to support completed, failed,
 and approval-paused runs; input-required runs are resumed with
 `agent.ResumeInput` so the human response is explicit.
+
+## Cancellation and deadlines
+
+`ResumeInput` uses the caller's `context.Context` for checkpoint reads, writes,
+and the resumed model/tool turn. If the context is canceled or its deadline
+expires before the resume is committed, the call returns the context error and
+the checkpointed run remains `paused` at `input-required`; list it with
+`agent.Pending` and retry with a fresh context once the operator is ready.

examples/support/README.md

@@ -1,9 +1,25 @@
-# Support desk
+# Zero-to-hero support desk
 
-A real-world agent built the Go Micro way: a few services, an agent that
-manages them, an event that triggers it, and a human-in-the-loop gate on the
-one action that touches a customer. It's the "zero to hero" shape in one
-runnable file.
+A maintained 0-to-hero reference for the Go Micro lifecycle: scaffold a few
+typed services, run them in one process, let an agent chat with those services
+as tools, then inspect the durable flow that triggered the work. It is one
+runnable file and one CI smoke test, so the reference path stays honest as the
+framework evolves.
+
+## The path
+
+1. **Scaffold services** — `customers`, `tickets`, and `notify` are ordinary
+   typed Go Micro services. Their request/response structs and method comments
+   become the tool contract the agent sees.
+2. **Run the harness** — the example starts an in-memory registry, broker,
+   client, store, services, agent, and flow in one process; no external
+   dependencies or API key are required for the default run.
+3. **Chat through an agent** — the `support` agent receives the ticket event as
+   a prompt and calls service tools to look up the customer, triage the ticket,
+   and draft a reply.
+4. **Inspect the workflow** — the `intake` flow records the event-driven run and
+   prints the agent result, showing the service → agent → workflow lifecycle as
+   one runtime.
 
 ## The scenario
 
@@ -44,6 +60,12 @@ agent, which:
 go run main.go            # mock model — deterministic, no API key
 ```
 
+The maintained check is the same deterministic path:
+
+```bash
+go test ./examples/support
+```
+
 Against a live model, the agent reasons about the ticket itself instead of
 following the script:
 

examples/support/main_test.go

@@ -1,9 +1,32 @@
 package main
 
-import "testing"
+import (
+	"os"
+	"strings"
+	"testing"
+)
 
 func TestRunSupportMockSmoke(t *testing.T) {
 	if err := runSupport("mock"); err != nil {
 		t.Fatalf("support example failed: %v", err)
 	}
 }
+
+func TestZeroToHeroReadmeDocumentsLifecycle(t *testing.T) {
+	b, err := os.ReadFile("README.md")
+	if err != nil {
+		t.Fatalf("read README.md: %v", err)
+	}
+	doc := string(b)
+	for _, want := range []string{
+		"Scaffold services",
+		"Run the harness",
+		"Chat through an agent",
+		"Inspect the workflow",
+		"go test ./examples/support",
+	} {
+		if !strings.Contains(doc, want) {
+			t.Fatalf("README.md missing zero-to-hero step %q", want)
+		}
+	}
+}

flow/analyze.go

@@ -0,0 +1,212 @@
+package flow
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"sort"
+	"strings"
+	"time"
+
+	"go-micro.dev/v6/ai"
+)
+
+// AnalyzeOptions configures Analyze.
+type AnalyzeOptions struct {
+	// MaxFeedbackSamples bounds the number of representative grader feedback
+	// strings retained per candidate. Values <= 0 use a small default.
+	MaxFeedbackSamples int
+}
+
+// AnalyzeOption configures Analyze.
+type AnalyzeOption func(*AnalyzeOptions)
+
+// AnalyzeMaxFeedbackSamples sets how many grader feedback examples are kept for
+// each candidate in the report.
+func AnalyzeMaxFeedbackSamples(n int) AnalyzeOption {
+	return func(o *AnalyzeOptions) { o.MaxFeedbackSamples = n }
+}
+
+// Report is the machine-readable output of Analyze. Candidates are ordered from
+// worst to best so an agent, CLI, or human can pick the first improvement to try.
+type Report struct {
+	Candidates []Candidate `json:"candidates"`
+}
+
+// Candidate identifies one underperforming flow step and the trace evidence that
+// made it worth improving.
+type Candidate struct {
+	Step           string        `json:"step"`
+	Metric         string        `json:"metric"`
+	Score          float64       `json:"score"`
+	Runs           int           `json:"runs"`
+	Failures       int           `json:"failures"`
+	PassRate       float64       `json:"pass_rate"`
+	ErrorRate      float64       `json:"error_rate"`
+	AverageRetries float64       `json:"average_retries"`
+	P50Latency     time.Duration `json:"p50_latency"`
+	P95Latency     time.Duration `json:"p95_latency"`
+	SampleFeedback []string      `json:"sample_feedback,omitempty"`
+	RunIDs         []string      `json:"run_ids,omitempty"`
+}
+
+// Analyze aggregates a bounded window of persisted flow runs and returns ranked
+// hill-climbing candidates. It uses the same Run records read by Checkpoint.List:
+// failed verification fields in step results drive pass-rate and feedback, step
+// status drives error rate, and retry attempts contribute retry pressure. An
+// empty window returns an empty report.
+func Analyze(runs []Run, opts ...AnalyzeOption) Report {
+	o := AnalyzeOptions{MaxFeedbackSamples: 3}
+	for _, opt := range opts {
+		opt(&o)
+	}
+	if o.MaxFeedbackSamples <= 0 {
+		o.MaxFeedbackSamples = 3
+	}
+
+	stats := map[string]*stepStats{}
+	for _, run := range runs {
+		for _, step := range run.Steps {
+			if step.Name == "" {
+				continue
+			}
+			s := stats[step.Name]
+			if s == nil {
+				s = &stepStats{}
+				stats[step.Name] = s
+			}
+			s.runs++
+			s.runIDs = appendUnique(s.runIDs, run.ID)
+			if step.Attempts > 1 {
+				s.retries += step.Attempts - 1
+			}
+			if step.Status == "failed" || step.Error != "" {
+				s.errors++
+			}
+			if len(run.Steps) > 0 && !run.Started.IsZero() && !run.Updated.IsZero() {
+				s.latencies = append(s.latencies, run.Updated.Sub(run.Started)/time.Duration(len(run.Steps)))
+			}
+			passed, feedback, ok := verificationFields(step.Result)
+			if ok {
+				s.graded++
+				if !passed {
+					s.gradeFailures++
+					if feedback != "" && len(s.feedback) < o.MaxFeedbackSamples {
+						s.feedback = append(s.feedback, feedback)
+					}
+				}
+			}
+		}
+	}
+
+	report := Report{}
+	for step, s := range stats {
+		if s.runs == 0 {
+			continue
+		}
+		failures := s.errors + s.gradeFailures
+		passRate := 1.0
+		if s.graded > 0 {
+			passRate = float64(s.graded-s.gradeFailures) / float64(s.graded)
+		} else if s.errors > 0 {
+			passRate = float64(s.runs-s.errors) / float64(s.runs)
+		}
+		errorRate := float64(s.errors) / float64(s.runs)
+		avgRetries := float64(s.retries) / float64(s.runs)
+		score := float64(s.gradeFailures)*3 + float64(s.errors)*2 + avgRetries
+		metric := "pass_rate"
+		if s.gradeFailures == 0 && s.errors > 0 {
+			metric = "error_rate"
+		} else if s.gradeFailures == 0 && s.errors == 0 && s.retries > 0 {
+			metric = "retry_count"
+		}
+		report.Candidates = append(report.Candidates, Candidate{
+			Step: step, Metric: metric, Score: score, Runs: s.runs, Failures: failures,
+			PassRate: passRate, ErrorRate: errorRate, AverageRetries: avgRetries,
+			P50Latency: percentile(s.latencies, 0.50), P95Latency: percentile(s.latencies, 0.95),
+			SampleFeedback: append([]string(nil), s.feedback...), RunIDs: append([]string(nil), s.runIDs...),
+		})
+	}
+	sort.SliceStable(report.Candidates, func(i, j int) bool {
+		a, b := report.Candidates[i], report.Candidates[j]
+		if a.Score == b.Score {
+			return a.Step < b.Step
+		}
+		return a.Score > b.Score
+	})
+	return report
+}
+
+type stepStats struct {
+	runs, graded, gradeFailures, errors, retries int
+	feedback, runIDs                             []string
+	latencies                                    []time.Duration
+}
+
+// PromptOptimizer proposes prompt improvements for a candidate without mutating
+// the source flow. Applying the returned prompt stays explicitly gated by the caller.
+type PromptOptimizer struct{ model ai.Model }
+
+// LLMOptimizer returns an optimizer that asks model to revise prompts for
+// Analyze candidates. The model is injected so tests and callers can use mocks.
+func LLMOptimizer(model ai.Model) *PromptOptimizer { return &PromptOptimizer{model: model} }
+
+// OptimizePrompt asks the model for a revised prompt for candidate using the
+// current prompt and trace feedback. It returns only the proposal; it never
+// modifies a Flow, Step, or Checkpoint.
+func (o *PromptOptimizer) OptimizePrompt(ctx context.Context, candidate Candidate, currentPrompt string) (string, error) {
+	if o == nil || o.model == nil {
+		return "", fmt.Errorf("flow: LLMOptimizer requires a model")
+	}
+	prompt := fmt.Sprintf("Revise this workflow step prompt to improve the failing step.\nStep: %s\nMetric: %s\nScore: %.2f\nFeedback:\n- %s\n\nCurrent prompt:\n%s\n\nReturn only the revised prompt.", candidate.Step, candidate.Metric, candidate.Score, strings.Join(candidate.SampleFeedback, "\n- "), currentPrompt)
+	resp, err := o.model.Generate(ctx, &ai.Request{Prompt: prompt})
+	if err != nil {
+		return "", err
+	}
+	proposal := strings.TrimSpace(resp.Answer)
+	if proposal == "" {
+		proposal = strings.TrimSpace(resp.Reply)
+	}
+	if proposal == "" {
+		return "", fmt.Errorf("flow: LLMOptimizer returned an empty prompt")
+	}
+	return proposal, nil
+}
+
+func verificationFields(result string) (bool, string, bool) {
+	if result == "" {
+		return false, "", false
+	}
+	var obj map[string]any
+	if err := json.Unmarshal([]byte(result), &obj); err != nil {
+		return false, "", false
+	}
+	v, ok := obj["verification_passed"].(bool)
+	if !ok {
+		return false, "", false
+	}
+	fb, _ := obj["verification_feedback"].(string)
+	return v, fb, true
+}
+
+func appendUnique(values []string, value string) []string {
+	if value == "" {
+		return values
+	}
+	for _, v := range values {
+		if v == value {
+			return values
+		}
+	}
+	return append(values, value)
+}
+
+func percentile(values []time.Duration, p float64) time.Duration {
+	if len(values) == 0 {
+		return 0
+	}
+	sorted := append([]time.Duration(nil), values...)
+	sort.Slice(sorted, func(i, j int) bool { return sorted[i] < sorted[j] })
+	idx := int(float64(len(sorted)-1) * p)
+	return sorted[idx]
+}

flow/analyze_test.go

@@ -0,0 +1,89 @@
+package flow
+
+import (
+	"context"
+	"strings"
+	"testing"
+	"time"
+
+	"go-micro.dev/v6/ai"
+)
+
+func TestAnalyzeRanksFailedGraderStepAbovePassingStep(t *testing.T) {
+	now := time.Now()
+	runs := []Run{
+		{ID: "run-1", Started: now, Updated: now.Add(time.Second), Steps: []StepRecord{
+			{Name: "draft", Status: "done", Attempts: 2, Result: `{"verification_passed":false,"verification_feedback":"cite sources"}`},
+			{Name: "publish", Status: "done", Attempts: 1, Result: `{"verification_passed":true,"verification_feedback":"ok"}`},
+		}},
+		{ID: "run-2", Started: now, Updated: now.Add(2 * time.Second), Steps: []StepRecord{
+			{Name: "draft", Status: "done", Attempts: 1, Result: `{"verification_passed":false,"verification_feedback":"too vague"}`},
+			{Name: "publish", Status: "done", Attempts: 1, Result: `{"verification_passed":true,"verification_feedback":"ok"}`},
+		}},
+	}
+
+	report := Analyze(runs)
+	if len(report.Candidates) != 2 {
+		t.Fatalf("Analyze returned %d candidates, want 2", len(report.Candidates))
+	}
+	if got := report.Candidates[0].Step; got != "draft" {
+		t.Fatalf("top candidate = %q, want draft", got)
+	}
+	if report.Candidates[0].PassRate != 0 {
+		t.Fatalf("draft pass rate = %v, want 0", report.Candidates[0].PassRate)
+	}
+}
+
+func TestAnalyzeCarriesFeedbackSamplesAndRunIDs(t *testing.T) {
+	report := Analyze([]Run{{ID: "run-9", Steps: []StepRecord{{
+		Name: "grade", Status: "done", Attempts: 3,
+		Result: `{"verification_passed":false,"verification_feedback":"include totals"}`,
+	}}}})
+	if len(report.Candidates) != 1 {
+		t.Fatalf("candidates = %d, want 1", len(report.Candidates))
+	}
+	c := report.Candidates[0]
+	if len(c.SampleFeedback) != 1 || c.SampleFeedback[0] != "include totals" {
+		t.Fatalf("feedback = %#v, want include totals", c.SampleFeedback)
+	}
+	if len(c.RunIDs) != 1 || c.RunIDs[0] != "run-9" {
+		t.Fatalf("run ids = %#v, want run-9", c.RunIDs)
+	}
+	if c.AverageRetries != 2 {
+		t.Fatalf("average retries = %v, want 2", c.AverageRetries)
+	}
+}
+
+func TestAnalyzeEmptyWindowReturnsEmptyReport(t *testing.T) {
+	if got := Analyze(nil); len(got.Candidates) != 0 {
+		t.Fatalf("empty Analyze candidates = %d, want 0", len(got.Candidates))
+	}
+}
+
+func TestLLMOptimizerReturnsProposalWithoutMutatingFlow(t *testing.T) {
+	f := New("optimize", Prompt("original prompt"))
+	before := f.opts.Prompt
+	optimizer := LLMOptimizer(&optimizerModel{reply: "revised prompt"})
+	proposal, err := optimizer.OptimizePrompt(context.Background(), Candidate{Step: "draft", Metric: "pass_rate", SampleFeedback: []string{"cite sources"}}, before)
+	if err != nil {
+		t.Fatalf("OptimizePrompt returned error: %v", err)
+	}
+	if !strings.Contains(proposal, "revised") {
+		t.Fatalf("proposal = %q, want revised prompt", proposal)
+	}
+	if f.opts.Prompt != before {
+		t.Fatalf("flow prompt mutated to %q, want %q", f.opts.Prompt, before)
+	}
+}
+
+type optimizerModel struct{ reply string }
+
+func (m *optimizerModel) Init(...ai.Option) error { return nil }
+func (m *optimizerModel) Options() ai.Options     { return ai.Options{} }
+func (m *optimizerModel) Generate(context.Context, *ai.Request, ...ai.GenerateOption) (*ai.Response, error) {
+	return &ai.Response{Reply: m.reply}, nil
+}
+func (m *optimizerModel) Stream(context.Context, *ai.Request, ...ai.GenerateOption) (ai.Stream, error) {
+	return nil, ai.ErrStreamingUnsupported
+}
+func (m *optimizerModel) String() string { return "optimizer" }

flow/example_verify_test.go

@@ -0,0 +1,45 @@
+package flow_test
+
+import (
+	"context"
+	"fmt"
+	"strings"
+
+	"go-micro.dev/v6/flow"
+)
+
+func ExampleVerify() {
+	generate := func(_ context.Context, in flow.State) (flow.State, error) {
+		if strings.Contains(in.String(), "feedback") {
+			in.Data = []byte(`{"answer":"include a source"}`)
+			return in, nil
+		}
+		in.Data = []byte(`{"answer":"draft"}`)
+		return in, nil
+	}
+	grader := func(_ context.Context, out flow.State) (bool, string, error) {
+		return strings.Contains(out.String(), "source"), "add a source", nil
+	}
+
+	out, _ := flow.Verify(generate, grader, flow.VerifyMaxAttempts(2))(context.Background(), flow.State{})
+	fmt.Println(strings.Contains(out.String(), `"verification_passed":true`))
+	// Output: true
+}
+
+func ExampleAnalyze() {
+	runs := []flow.Run{{
+		ID: "run-1",
+		Steps: []flow.StepRecord{{
+			Name:   "draft",
+			Status: "done",
+			Result: `{"verification_passed":false,"verification_feedback":"add a source"}`,
+		}},
+	}}
+
+	report := flow.Analyze(runs)
+	fmt.Println(report.Candidates[0].Step)
+	fmt.Println(report.Candidates[0].SampleFeedback[0])
+	// Output:
+	// draft
+	// add a source
+}

flow/verify.go

@@ -0,0 +1,191 @@
+package flow
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"strings"
+	"time"
+
+	"go-micro.dev/v6/ai"
+)
+
+// Grader checks a step output against a rubric. It returns pass=true when the
+// output is acceptable; otherwise feedback should explain what the next attempt
+// should fix.
+type Grader func(ctx context.Context, out State) (pass bool, feedback string, err error)
+
+// VerifyOptions configure Verify.
+type VerifyOptions struct {
+	// MaxAttempts bounds how many times the body can run. Default 2.
+	MaxAttempts int
+	// Backoff waits between failed grades. Zero means retry immediately.
+	Backoff time.Duration
+	// FeedbackField is the JSON field used to thread grader feedback into the
+	// next attempt's input. Default "feedback".
+	FeedbackField string
+}
+
+// VerifyOption configures Verify.
+type VerifyOption func(*VerifyOptions)
+
+// VerifyMaxAttempts sets the total attempt budget for Verify. Values <= 0 use
+// the default of 2.
+func VerifyMaxAttempts(n int) VerifyOption { return func(o *VerifyOptions) { o.MaxAttempts = n } }
+
+// VerifyBackoff sets the delay between failed verification attempts.
+func VerifyBackoff(d time.Duration) VerifyOption { return func(o *VerifyOptions) { o.Backoff = d } }
+
+// VerifyFeedbackField sets the JSON field used to pass grader feedback to the
+// next body attempt. Empty values use "feedback".
+func VerifyFeedbackField(field string) VerifyOption {
+	return func(o *VerifyOptions) { o.FeedbackField = field }
+}
+
+// Verify runs body, grades its output, and retries with grader feedback threaded
+// into the next input until the grader passes or MaxAttempts is exhausted. It is
+// a StepFunc, so it composes directly as Step.Run with Loop, LLM, Call, Agent, or
+// any code-defined step.
+//
+// On a failed grade, Verify adds the feedback to the next attempt's input as a
+// JSON field named "feedback" (or VerifyFeedbackField). When all attempts fail,
+// it returns the last output without error, annotated with verification fields so
+// the run can keep the bounded failure outcome in its state:
+// "verification_passed": false, "verification_feedback", and
+// "verification_attempts".
+func Verify(body StepFunc, grader Grader, opts ...VerifyOption) StepFunc {
+	o := VerifyOptions{MaxAttempts: 2, FeedbackField: "feedback"}
+	for _, op := range opts {
+		op(&o)
+	}
+	if o.MaxAttempts <= 0 {
+		o.MaxAttempts = 2
+	}
+	if o.FeedbackField == "" {
+		o.FeedbackField = "feedback"
+	}
+	return func(ctx context.Context, in State) (State, error) {
+		if body == nil {
+			return in, fmt.Errorf("flow: Verify requires a body step")
+		}
+		if grader == nil {
+			return in, fmt.Errorf("flow: Verify requires a grader")
+		}
+		cur := in
+		last := in
+		feedback := ""
+		for attempt := 1; attempt <= o.MaxAttempts; attempt++ {
+			if err := ctx.Err(); err != nil {
+				return last, err
+			}
+			if feedback != "" {
+				var err error
+				cur, err = stateWithField(cur, o.FeedbackField, feedback)
+				if err != nil {
+					return last, err
+				}
+			}
+			out, err := body(ctx, cur)
+			if err != nil {
+				return last, fmt.Errorf("verify attempt %d: %w", attempt, err)
+			}
+			last = out
+			pass, fb, err := grader(ctx, out)
+			if err != nil {
+				return last, fmt.Errorf("verify grade attempt %d: %w", attempt, err)
+			}
+			if pass {
+				return stateWithVerification(out, true, fb, attempt)
+			}
+			feedback = fb
+			cur = in
+			if attempt < o.MaxAttempts && o.Backoff > 0 {
+				select {
+				case <-time.After(o.Backoff):
+				case <-ctx.Done():
+					return last, ctx.Err()
+				}
+			}
+		}
+		return stateWithVerification(last, false, feedback, o.MaxAttempts)
+	}
+}
+
+// LLMGrader returns a grader that asks the flow model to judge the latest output
+// against rubric. The model should answer with pass/fail plus short feedback.
+// It reuses the flow's configured model, so it must run inside a flow.
+func LLMGrader(rubric string) Grader {
+	return func(ctx context.Context, out State) (bool, string, error) {
+		d := depsFrom(ctx)
+		if d == nil || d.model == nil {
+			return false, "", fmt.Errorf("flow: LLMGrader requires a flow model (set Provider/APIKey)")
+		}
+		prompt := fmt.Sprintf("Grade the latest result against this rubric:\n%s\n\nLatest result:\n%s\n\nAnswer with PASS or FAIL on the first line, followed by one short feedback sentence.", rubric, out.String())
+		resp, err := d.model.Generate(ctx, &ai.Request{Prompt: prompt})
+		if err != nil {
+			return false, "", err
+		}
+		reply := resp.Answer
+		if reply == "" {
+			reply = resp.Reply
+		}
+		return parseGrade(reply)
+	}
+}
+
+func parseGrade(reply string) (bool, string, error) {
+	text := strings.TrimSpace(reply)
+	if text == "" {
+		return false, "", fmt.Errorf("flow: LLMGrader returned an empty grade")
+	}
+	lines := strings.SplitN(text, "\n", 2)
+	first := strings.ToLower(strings.TrimSpace(lines[0]))
+	feedback := ""
+	if len(lines) > 1 {
+		feedback = strings.TrimSpace(lines[1])
+	}
+	pass := strings.HasPrefix(first, "pass") || isAffirmative(first)
+	if !pass && feedback == "" {
+		feedback = text
+	}
+	return pass, feedback, nil
+}
+
+func stateWithField(s State, field, value string) (State, error) {
+	var obj map[string]any
+	if len(s.Data) > 0 && json.Unmarshal(s.Data, &obj) == nil && obj != nil {
+		obj[field] = value
+		return stateWithObject(s, obj)
+	}
+	obj = map[string]any{field: value}
+	if len(s.Data) > 0 {
+		obj["data"] = s.String()
+	}
+	return stateWithObject(s, obj)
+}
+
+func stateWithVerification(s State, passed bool, feedback string, attempts int) (State, error) {
+	var obj map[string]any
+	if len(s.Data) > 0 && json.Unmarshal(s.Data, &obj) == nil && obj != nil {
+		obj["verification_passed"] = passed
+		obj["verification_feedback"] = feedback
+		obj["verification_attempts"] = attempts
+		return stateWithObject(s, obj)
+	}
+	obj = map[string]any{
+		"data":                  s.String(),
+		"verification_passed":   passed,
+		"verification_feedback": feedback,
+		"verification_attempts": attempts,
+	}
+	return stateWithObject(s, obj)
+}
+
+func stateWithObject(s State, obj map[string]any) (State, error) {
+	b, err := json.Marshal(obj)
+	if err != nil {
+		return s, err
+	}
+	s.Data = b
+	return s, nil
+}

flow/verify_test.go

@@ -0,0 +1,96 @@
+package flow
+
+import (
+	"context"
+	"strings"
+	"testing"
+)
+
+func TestVerifyPassesFirstTry(t *testing.T) {
+	attempts := 0
+	step := Verify(func(_ context.Context, in State) (State, error) {
+		attempts++
+		in.Data = []byte(`{"answer":"ok"}`)
+		return in, nil
+	}, func(context.Context, State) (bool, string, error) {
+		return true, "looks good", nil
+	}, VerifyMaxAttempts(3))
+
+	out, err := step(context.Background(), State{})
+	if err != nil {
+		t.Fatalf("Verify returned error: %v", err)
+	}
+	if attempts != 1 {
+		t.Fatalf("body attempts = %d, want 1", attempts)
+	}
+	var got map[string]any
+	if err := out.Scan(&got); err != nil {
+		t.Fatalf("scan output: %v", err)
+	}
+	if got["verification_passed"] != true {
+		t.Fatalf("verification_passed = %v, want true", got["verification_passed"])
+	}
+}
+
+func TestVerifyRetriesWithFeedback(t *testing.T) {
+	attempts := 0
+	var secondInput map[string]string
+	step := Verify(func(_ context.Context, in State) (State, error) {
+		attempts++
+		if attempts == 2 {
+			if err := in.Scan(&secondInput); err != nil {
+				t.Fatalf("scan second input: %v", err)
+			}
+		}
+		in.Data = []byte(`{"answer":"draft"}`)
+		return in, nil
+	}, func(_ context.Context, _ State) (bool, string, error) {
+		return attempts >= 2, "include citations", nil
+	}, VerifyMaxAttempts(3))
+
+	out, err := step(context.Background(), State{Data: []byte(`{"topic":"agents"}`)})
+	if err != nil {
+		t.Fatalf("Verify returned error: %v", err)
+	}
+	if attempts != 2 {
+		t.Fatalf("body attempts = %d, want 2", attempts)
+	}
+	if secondInput["feedback"] != "include citations" {
+		t.Fatalf("feedback = %q, want include citations", secondInput["feedback"])
+	}
+	if !strings.Contains(out.String(), `"verification_passed":true`) {
+		t.Fatalf("output missing successful verification annotation: %s", out.String())
+	}
+}
+
+func TestVerifyExhaustsAttemptsReturnsLastOutput(t *testing.T) {
+	attempts := 0
+	step := Verify(func(_ context.Context, in State) (State, error) {
+		attempts++
+		in.Data = []byte(`{"answer":"still wrong"}`)
+		return in, nil
+	}, func(context.Context, State) (bool, string, error) {
+		return false, "try again", nil
+	}, VerifyMaxAttempts(2))
+
+	out, err := step(context.Background(), State{})
+	if err != nil {
+		t.Fatalf("Verify returned error: %v", err)
+	}
+	if attempts != 2 {
+		t.Fatalf("body attempts = %d, want 2", attempts)
+	}
+	var got map[string]any
+	if err := out.Scan(&got); err != nil {
+		t.Fatalf("scan output: %v", err)
+	}
+	if got["verification_passed"] != false {
+		t.Fatalf("verification_passed = %v, want false", got["verification_passed"])
+	}
+	if got["verification_feedback"] != "try again" {
+		t.Fatalf("verification_feedback = %v, want try again", got["verification_feedback"])
+	}
+	if got["verification_attempts"] != float64(2) {
+		t.Fatalf("verification_attempts = %v, want 2", got["verification_attempts"])
+	}
+}

gateway/a2a/a2a.go

@@ -28,6 +28,7 @@ package a2a
 import (
 	"context"
 	"encoding/json"
+	"errors"
 	"fmt"
 	"io"
 	"log"
@@ -140,8 +141,11 @@ func (g *Gateway) Handler() http.Handler {
 	// Per-agent card (served at the agent's url and at its well-known path).
 	mux.HandleFunc("GET /agents/{name}", g.handleCard)
 	mux.HandleFunc("GET /agents/{name}/.well-known/agent.json", g.handleCard)
+	mux.HandleFunc("GET /agents/{name}/skills/{skill}", g.handleSkillCard)
+	mux.HandleFunc("GET /agents/{name}/skills/{skill}/.well-known/agent.json", g.handleSkillCard)
 	// Per-agent JSON-RPC endpoint.
 	mux.HandleFunc("POST /agents/{name}", g.handleRPC)
+	mux.HandleFunc("POST /agents/{name}/skills/{skill}", g.handleSkillRPC)
 	// Top-level well-known: serve the single agent's card if there's
 	// exactly one, otherwise point to the directory.
 	mux.HandleFunc("GET /.well-known/agent.json", g.handleWellKnown)
@@ -338,23 +342,17 @@ func Card(name, url, description string, services []string) AgentCard {
 			description = "Go Micro agent"
 		}
 	}
+	skills := skillsFromServices(services)
 	return AgentCard{
-		Name:            name,
-		Description:     description,
-		URL:             url,
-		Version:         "1.0.0",
-		ProtocolVersion: protocolVersion,
-		Capabilities:    Capabilities{Streaming: true, PushNotifications: true},
-		// The agent converses over a single Chat endpoint; advertise that
-		// as one skill, tagged with the services it manages.
+		Name:               name,
+		Description:        description,
+		URL:                url,
+		Version:            "1.0.0",
+		ProtocolVersion:    protocolVersion,
+		Capabilities:       Capabilities{Streaming: true, PushNotifications: true},
 		DefaultInputModes:  []string{"text/plain"},
 		DefaultOutputModes: []string{"text/plain"},
-		Skills: []Skill{{
-			ID:          "chat",
-			Name:        "Chat",
-			Description: "Converse with the agent to operate its services.",
-			Tags:        services,
-		}},
+		Skills:             skills,
 	}
 }
 
@@ -371,6 +369,19 @@ func (g *Gateway) lookupCard(name string) (AgentCard, bool) {
 	return g.card(name, meta), true
 }
 
+func (g *Gateway) lookupSkillCard(name, skillID string) (AgentCard, Skill, bool) {
+	card, ok := g.lookupCard(name)
+	if !ok {
+		return AgentCard{}, Skill{}, false
+	}
+	for _, skill := range card.Skills {
+		if skill.ID == skillID {
+			return card, skill, true
+		}
+	}
+	return AgentCard{}, Skill{}, false
+}
+
 // ---------------------------------------------------------------------------
 // HTTP handlers
 // ---------------------------------------------------------------------------
@@ -393,6 +404,17 @@ func (g *Gateway) handleCard(w http.ResponseWriter, r *http.Request) {
 	writeJSON(w, http.StatusOK, card)
 }
 
+func (g *Gateway) handleSkillCard(w http.ResponseWriter, r *http.Request) {
+	card, skill, ok := g.lookupSkillCard(r.PathValue("name"), r.PathValue("skill"))
+	if !ok {
+		http.NotFound(w, r)
+		return
+	}
+	card.URL = g.opts.BaseURL + "/agents/" + r.PathValue("name") + "/skills/" + skill.ID
+	card.Skills = []Skill{skill}
+	writeJSON(w, http.StatusOK, card)
+}
+
 func (g *Gateway) handleWellKnown(w http.ResponseWriter, r *http.Request) {
 	cards, err := g.cards()
 	if err != nil {
@@ -421,6 +443,18 @@ func (g *Gateway) handleRPC(w http.ResponseWriter, r *http.Request) {
 	})
 }
 
+func (g *Gateway) handleSkillRPC(w http.ResponseWriter, r *http.Request) {
+	name := r.PathValue("name")
+	_, skill, ok := g.lookupSkillCard(name, r.PathValue("skill"))
+	if !ok {
+		writeRPC(w, nil, nil, &rpcError{Code: errInvalidParams, Message: "unknown agent skill: " + name + "/" + r.PathValue("skill")})
+		return
+	}
+	g.disp.serve(w, r, func(ctx context.Context, text string) (string, error) {
+		return g.callAgent(ctx, name, skillPrompt(skill, text))
+	})
+}
+
 // dispatcher handles A2A JSON-RPC requests against an Invoke function and
 // retains recent tasks for tasks/get. It is shared by the gateway (one
 // per registry) and embedded agents (one per agent).
@@ -456,7 +490,7 @@ func (d *dispatcher) serveWithStream(w http.ResponseWriter, r *http.Request, inv
 		d.send(requestContext(r.Context()), w, req, invoke)
 	case "message/stream":
 		if streamInvoke != nil {
-			d.streamChunks(requestContext(r.Context()), w, req, streamInvoke)
+			d.streamChunks(requestContext(r.Context()), w, req, streamInvoke, invoke)
 			return
 		}
 		d.stream(requestContext(r.Context()), w, req, invoke)
@@ -505,7 +539,7 @@ func (d *dispatcher) stream(ctx context.Context, w http.ResponseWriter, req rpcR
 	}
 }
 
-func (d *dispatcher) streamChunks(ctx context.Context, w http.ResponseWriter, req rpcRequest, invoke StreamInvoke) {
+func (d *dispatcher) streamChunks(ctx context.Context, w http.ResponseWriter, req rpcRequest, invoke StreamInvoke, fallback Invoke) {
 	var p sendParams
 	if err := json.Unmarshal(req.Params, &p); err != nil {
 		writeRPC(w, req.ID, nil, &rpcError{Code: errInvalidParams, Message: "invalid params"})
@@ -518,6 +552,10 @@ func (d *dispatcher) streamChunks(ctx context.Context, w http.ResponseWriter, re
 	}
 	stream, err := invoke(ctx, text)
 	if err != nil {
+		if errors.Is(err, ai.ErrStreamingUnsupported) && fallback != nil {
+			d.stream(ctx, w, req, fallback)
+			return
+		}
 		writeRPC(w, req.ID, nil, &rpcError{Code: errInternal, Message: err.Error()})
 		return
 	}
@@ -853,6 +891,68 @@ func (d *dispatcher) deliverPush(taskID string, task *Task) {
 	}
 }
 
+func skillsFromServices(services []string) []Skill {
+	if len(services) == 0 {
+		return []Skill{{ID: "chat", Name: "Chat", Description: "Converse with the agent to operate its services."}}
+	}
+	seen := map[string]bool{}
+	var skills []Skill
+	for _, service := range services {
+		service = strings.TrimSpace(service)
+		if service == "" {
+			continue
+		}
+		id := skillID(service)
+		if id == "" || seen[id] {
+			continue
+		}
+		seen[id] = true
+		skills = append(skills, Skill{
+			ID:          id,
+			Name:        skillName(service),
+			Description: fmt.Sprintf("Operate the %s service through this agent.", service),
+			Tags:        []string{service},
+		})
+	}
+	if len(skills) == 0 {
+		return []Skill{{ID: "chat", Name: "Chat", Description: "Converse with the agent to operate its services."}}
+	}
+	return skills
+}
+
+func skillID(service string) string {
+	service = strings.ToLower(strings.TrimSpace(service))
+	var b strings.Builder
+	dash := false
+	for _, r := range service {
+		if (r >= 'a' && r <= 'z') || (r >= '0' && r <= '9') {
+			b.WriteRune(r)
+			dash = false
+			continue
+		}
+		if !dash && b.Len() > 0 {
+			b.WriteByte('-')
+			dash = true
+		}
+	}
+	return strings.Trim(b.String(), "-")
+}
+
+func skillName(service string) string {
+	parts := strings.FieldsFunc(service, func(r rune) bool { return r == '-' || r == '_' || r == '.' || r == '/' || r == ' ' })
+	for i, part := range parts {
+		if part == "" {
+			continue
+		}
+		parts[i] = strings.ToUpper(part[:1]) + part[1:]
+	}
+	return strings.Join(parts, " ")
+}
+
+func skillPrompt(skill Skill, text string) string {
+	return fmt.Sprintf("Use the %q skill (%s) for this request.\n\n%s", skill.Name, skill.ID, text)
+}
+
 func textOf(parts []Part) string {
 	var b strings.Builder
 	for _, p := range parts {

gateway/a2a/a2a_test.go

@@ -51,7 +51,7 @@ func newGatewayWithAgent(t *testing.T) (*httptest.Server, func()) {
 		server.Name("echo"),
 		server.Address("127.0.0.1:0"),
 		server.Registry(reg),
-		server.Metadata(map[string]string{"type": "agent", "services": ""}),
+		server.Metadata(map[string]string{"type": "agent", "services": "task,project"}),
 	)
 	if err := pb.RegisterAgentHandler(srv, echoAgent{}); err != nil {
 		t.Fatalf("register agent handler: %v", err)
@@ -88,8 +88,40 @@ func TestAgentCardFromRegistry(t *testing.T) {
 	if card.URL != "http://gw/agents/echo" {
 		t.Errorf("card url = %q", card.URL)
 	}
-	if card.ProtocolVersion == "" || len(card.Skills) == 0 {
-		t.Errorf("card missing protocolVersion or skills: %+v", card)
+	if card.ProtocolVersion == "" {
+		t.Errorf("card missing protocolVersion: %+v", card)
+	}
+	if got := skillIDs(card.Skills); strings.Join(got, ",") != "task,project" {
+		t.Errorf("skill IDs = %v, want [task project]", got)
+	}
+}
+
+func TestSkillEndpointServesFocusedCardAndRoutesRPC(t *testing.T) {
+	ts, cleanup := newGatewayWithAgent(t)
+	defer cleanup()
+
+	resp, err := http.Get(ts.URL + "/agents/echo/skills/task/.well-known/agent.json")
+	if err != nil {
+		t.Fatalf("get skill card: %v", err)
+	}
+	defer resp.Body.Close()
+	if resp.StatusCode != http.StatusOK {
+		t.Fatalf("skill card status = %d", resp.StatusCode)
+	}
+	var card AgentCard
+	if err := json.NewDecoder(resp.Body).Decode(&card); err != nil {
+		t.Fatalf("decode skill card: %v", err)
+	}
+	if card.URL != "http://gw/agents/echo/skills/task" || len(card.Skills) != 1 || card.Skills[0].ID != "task" {
+		t.Fatalf("skill card = %+v, want task-only card at skill URL", card)
+	}
+
+	task := rpcTask(t, ts.URL+"/agents/echo/skills/task", `{
+		"jsonrpc":"2.0","id":1,"method":"message/send",
+		"params":{"message":{"role":"user","kind":"message","messageId":"m1",
+			"parts":[{"kind":"text","text":"ping"}]}}}`)
+	if task.Status.State != stateCompleted || textOf(task.Artifacts[0].Parts) != "pong" {
+		t.Fatalf("skill task = %+v, want completed pong", task)
 	}
 }
 
@@ -433,6 +465,61 @@ func TestMessageStreamChunksPropagatesCancellationAndClosesStream(t *testing.T)
 	}
 }
 
+func TestMessageStreamChunksFallsBackWhenUnsupported(t *testing.T) {
+	d := newDispatcher()
+	body := `{"jsonrpc":"2.0","id":1,"method":"message/stream","params":{"message":{"role":"user","parts":[{"kind":"text","text":"ping"}],"kind":"message"}}}`
+	req := httptest.NewRequest(http.MethodPost, "/", bytes.NewBufferString(body))
+	rr := httptest.NewRecorder()
+	var streamed bool
+	var fallbackText string
+
+	d.serveWithStream(rr, req, func(ctx context.Context, text string) (string, error) {
+		fallbackText = text
+		return "pong", nil
+	}, func(ctx context.Context, text string) (ai.Stream, error) {
+		streamed = true
+		return nil, fmt.Errorf("%w: test provider", ai.ErrStreamingUnsupported)
+	})
+
+	if !streamed {
+		t.Fatal("stream invoke was not attempted")
+	}
+	if fallbackText != "ping" {
+		t.Fatalf("fallback text = %q, want ping", fallbackText)
+	}
+	if ct := rr.Result().Header.Get("Content-Type"); !strings.HasPrefix(ct, "text/event-stream") {
+		t.Fatalf("content-type = %q, want text/event-stream", ct)
+	}
+	var events []struct {
+		Result Task      `json:"result"`
+		Error  *rpcError `json:"error"`
+	}
+	for _, line := range strings.Split(strings.TrimSpace(rr.Body.String()), "\n") {
+		line = strings.TrimSpace(line)
+		if line == "" {
+			continue
+		}
+		line = strings.TrimPrefix(line, "data: ")
+		var event struct {
+			Result Task      `json:"result"`
+			Error  *rpcError `json:"error"`
+		}
+		if err := json.Unmarshal([]byte(line), &event); err != nil {
+			t.Fatalf("decode event %q: %v", line, err)
+		}
+		events = append(events, event)
+	}
+	if len(events) != 1 {
+		t.Fatalf("events = %d, want 1; body %s", len(events), rr.Body.String())
+	}
+	if events[0].Error != nil {
+		t.Fatalf("fallback event error: %+v", events[0].Error)
+	}
+	if events[0].Result.Status.State != stateCompleted || textOf(events[0].Result.Artifacts[0].Parts) != "pong" {
+		t.Fatalf("fallback task = %+v, want completed pong", events[0].Result)
+	}
+}
+
 func TestTasksResubscribeStreamsCurrentAndSubsequentEvents(t *testing.T) {
 	d := newDispatcher()
 	initial := &Task{ID: "task-1", ContextID: "ctx-1", Kind: "task", Status: TaskStatus{State: stateWorking, Timestamp: time.Now().UTC().Format(time.RFC3339)}}
@@ -644,3 +731,11 @@ func rpcTask(t *testing.T, url, body string) Task {
 	}
 	return resp.Result
 }
+
+func skillIDs(skills []Skill) []string {
+	ids := make([]string, 0, len(skills))
+	for _, skill := range skills {
+		ids = append(ids, skill.ID)
+	}
+	return ids
+}

gateway/mcp/httpjsonrpc.go

@@ -0,0 +1,129 @@
+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

@@ -0,0 +1,131 @@
+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

@@ -0,0 +1,64 @@
+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,6 +122,19 @@ 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,10 +19,13 @@ 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.
 
-## Developer experience (ranked)
+## Work queue (ranked)
 
-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.
+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. **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/harness/a2a-stream-fallback/main.go

@@ -0,0 +1,184 @@
+// A2A stream fallback harness.
+//
+// It exercises the gateway boundary that fronts an agent over A2A. The agent is
+// configured with tools and memory, but its model streaming path deliberately
+// reports ai.ErrStreamingUnsupported; the A2A gateway must fall back to the
+// normal Ask path and still complete the same tool-calling run.
+package main
+
+import (
+	"bufio"
+	"bytes"
+	"context"
+	"encoding/json"
+	"errors"
+	"flag"
+	"fmt"
+	"io"
+	"net/http"
+	"net/http/httptest"
+	"os"
+	"strings"
+	"time"
+
+	"go-micro.dev/v6/agent"
+	"go-micro.dev/v6/ai"
+	"go-micro.dev/v6/gateway/a2a"
+	"go-micro.dev/v6/registry"
+	"go-micro.dev/v6/store"
+)
+
+type mockModel struct{ opts ai.Options }
+
+func newMock(opts ...ai.Option) ai.Model {
+	m := &mockModel{}
+	_ = m.Init(opts...)
+	return m
+}
+
+func (m *mockModel) Init(opts ...ai.Option) error {
+	for _, o := range opts {
+		o(&m.opts)
+	}
+	return nil
+}
+func (m *mockModel) Options() ai.Options { return m.opts }
+func (m *mockModel) String() string      { return "mock" }
+func (m *mockModel) Stream(context.Context, *ai.Request, ...ai.GenerateOption) (ai.Stream, error) {
+	return nil, ai.ErrStreamingUnsupported
+}
+func (m *mockModel) Generate(ctx context.Context, req *ai.Request, _ ...ai.GenerateOption) (*ai.Response, error) {
+	if req.Prompt == "" {
+		return nil, errors.New("missing prompt")
+	}
+	if len(req.Messages) == 0 || req.Messages[len(req.Messages)-1].Role != "user" {
+		return nil, fmt.Errorf("missing user history: %+v", req.Messages)
+	}
+	if len(req.Tools) == 0 || m.opts.ToolHandler == nil {
+		return nil, errors.New("missing tools or tool handler")
+	}
+	res := m.opts.ToolHandler(ctx, ai.ToolCall{ID: "a2a-fallback-call", Name: "fallback_echo", Input: map[string]any{"value": "a2a-fallback"}})
+	if res.Content == "" {
+		return nil, errors.New("empty tool result")
+	}
+	return &ai.Response{Reply: "fallback completed", Answer: res.Content, ToolCalls: []ai.ToolCall{{ID: "a2a-fallback-call", Name: "fallback_echo", Input: map[string]any{"value": "a2a-fallback"}, Result: res.Content}}}, nil
+}
+
+func providerKey(provider string) string {
+	if v := os.Getenv("MICRO_AI_API_KEY"); v != "" {
+		return v
+	}
+	env := map[string]string{
+		"anthropic": "ANTHROPIC_API_KEY", "openai": "OPENAI_API_KEY",
+		"gemini": "GEMINI_API_KEY", "groq": "GROQ_API_KEY", "mistral": "MISTRAL_API_KEY",
+		"together": "TOGETHER_API_KEY", "atlascloud": "ATLASCLOUD_API_KEY",
+	}[provider]
+	return os.Getenv(env)
+}
+
+func main() {
+	provider := flag.String("provider", "mock", "LLM provider: mock (default), anthropic, openai, ...")
+	flag.Parse()
+
+	apiKey := ""
+	if *provider == "mock" {
+		ai.Register("mock", newMock)
+	} else {
+		apiKey = providerKey(*provider)
+		if apiKey == "" {
+			fmt.Printf("no API key for provider %q — set MICRO_AI_API_KEY or the provider's key env\n", *provider)
+			return
+		}
+	}
+
+	fmt.Printf("\n\033[1mA2A streaming fallback conformance (provider: %s)\033[0m\n", *provider)
+	reg := registry.NewMemoryRegistry()
+	st := store.NewMemoryStore()
+	var sawTool, sawRunInfo bool
+	ag := agent.New(
+		agent.Name("a2a-fallback"),
+		agent.Provider(*provider),
+		agent.APIKey(apiKey),
+		agent.Prompt("Use fallback_echo exactly once with value a2a-fallback, then answer with the tool result."),
+		agent.WithRegistry(reg),
+		agent.WithStore(st),
+		agent.WithMemory(agent.NewInMemory(8)),
+		agent.ModelCallTimeout(45*time.Second),
+		agent.WithTool("fallback_echo", "Echo the A2A fallback marker.", map[string]any{
+			"value": map[string]any{"type": "string", "description": "value to echo"},
+		}, func(ctx context.Context, input map[string]any) (string, error) {
+			sawTool = true
+			info, ok := ai.RunInfoFrom(ctx)
+			if !ok || info.RunID == "" || info.Agent != "a2a-fallback" {
+				return "", fmt.Errorf("unexpected run info: %+v", info)
+			}
+			sawRunInfo = true
+			if input["value"] != "a2a-fallback" {
+				return "", fmt.Errorf("unexpected value %v", input["value"])
+			}
+			return `{"marker":"a2a-fallback-ok"}`, nil
+		}),
+	)
+
+	card := a2a.Card("a2a-fallback", "http://example.invalid/a2a-fallback", "", nil)
+	handler := a2a.NewAgentStreamHandler(card, func(ctx context.Context, text string) (string, error) {
+		resp, err := ag.Ask(ctx, text)
+		if err != nil {
+			return "", err
+		}
+		return resp.Reply, nil
+	}, ag.Stream)
+
+	body := []byte(`{"jsonrpc":"2.0","id":1,"method":"message/stream","params":{"message":{"role":"user","parts":[{"kind":"text","text":"Run the A2A fallback conformance check."}],"kind":"message"}}}`)
+	req := httptest.NewRequest(http.MethodPost, "/", bytes.NewReader(body))
+	req.Header.Set("Content-Type", "application/json")
+	rr := httptest.NewRecorder()
+	handler.ServeHTTP(rr, req)
+	res := rr.Result()
+	defer res.Body.Close()
+	if res.StatusCode != http.StatusOK {
+		b, _ := io.ReadAll(res.Body)
+		fmt.Fprintf(os.Stderr, "unexpected status %d: %s\n", res.StatusCode, b)
+		os.Exit(1)
+	}
+	if ct := res.Header.Get("Content-Type"); !strings.HasPrefix(ct, "text/event-stream") {
+		fmt.Fprintf(os.Stderr, "content-type = %q, want text/event-stream\n", ct)
+		os.Exit(1)
+	}
+	payload, err := readSSEData(res.Body)
+	if err != nil {
+		fmt.Fprintln(os.Stderr, err)
+		os.Exit(1)
+	}
+	if !strings.Contains(payload, "a2a-fallback-ok") {
+		fmt.Fprintf(os.Stderr, "stream payload missing marker: %s\n", payload)
+		os.Exit(1)
+	}
+	if !sawTool || !sawRunInfo {
+		fmt.Fprintf(os.Stderr, "tool=%v runInfo=%v\n", sawTool, sawRunInfo)
+		os.Exit(1)
+	}
+	fmt.Println("\n\033[32m✓ A2A message/stream fell back to Ask and preserved tool/run metadata\033[0m")
+}
+
+func readSSEData(r io.Reader) (string, error) {
+	scanner := bufio.NewScanner(r)
+	var payload strings.Builder
+	for scanner.Scan() {
+		line := scanner.Text()
+		if data, ok := strings.CutPrefix(line, "data: "); ok {
+			payload.WriteString(data)
+			payload.WriteByte('\n')
+		}
+	}
+	if err := scanner.Err(); err != nil {
+		return "", err
+	}
+	if payload.Len() == 0 {
+		return "", errors.New("no SSE data received")
+	}
+	if !json.Valid([]byte(strings.TrimSpace(payload.String()))) {
+		return "", fmt.Errorf("SSE data is not JSON: %s", payload.String())
+	}
+	return payload.String(), nil
+}

internal/harness/provider-conformance/README.md

@@ -7,13 +7,17 @@ suite is safe for local development, forks, and scheduled CI.
 
 ## What it exercises
 
-`go run ./internal/harness/provider-conformance` fans out over the harnesses in
-`internal/harness`:
+`go run ./internal/harness/provider-conformance` fans out over the provider-facing
+agent test and the harnesses in `internal/harness`:
 
+- `agent` — provider tool-call conformance through `agent.Ask`, including run metadata propagation.
 - `universe` — service discovery plus agent tool calls over the real runtime.
 - `agent-flow` — a workflow event that drives an agent to call services.
 - `plan-delegate` — plan persistence plus agent-to-agent delegation and service
   calls.
+- `a2a-stream-fallback` — A2A `message/stream` through the gateway, including
+  fallback from unsupported provider streaming to the tool-calling `Ask` path while
+  preserving run metadata.
 
 The command also emits the registered provider capability matrix so the run shows
 which providers advertise model, image, video, and streaming support.
@@ -59,13 +63,14 @@ go run ./internal/harness/provider-conformance \
 ## Scheduled CI behavior
 
 The `Harness (E2E)` workflow runs on pushes and pull requests with deterministic
-mock LLMs. On the daily schedule and manual dispatch it also runs the live
+mock LLMs, including `provider-conformance -providers mock`. On the daily schedule and manual dispatch it also runs the live
 provider conformance job. That job:
 
-1. reads the provider keys from repository secrets,
-2. skips providers whose secrets are absent,
-3. fails when any configured provider fails a harness, and
-4. uploads JSON and Markdown coverage artifacts for the run.
+1. runs the same `agent`, `universe`, `agent-flow`, `plan-delegate`, and `a2a-stream-fallback` harness list,
+2. reads the provider keys from repository secrets,
+3. skips providers whose secrets are absent,
+4. fails when any configured provider fails a harness, and
+5. uploads JSON and Markdown coverage artifacts for the run.
 
 The job also appends the Markdown summary and capability matrix to the GitHub
 Actions step summary, making configured, skipped, and failed provider coverage

internal/harness/provider-conformance/main.go

@@ -46,7 +46,7 @@ var providerEnv = map[string]string{
 
 func main() {
 	providersFlag := flag.String("providers", "anthropic,openai,gemini,groq,mistral,together,atlascloud", "comma-separated providers to check; use mock for deterministic local checks")
-	harnessesFlag := flag.String("harnesses", "universe,agent-flow,plan-delegate", "comma-separated harness names under internal/harness")
+	harnessesFlag := flag.String("harnesses", "agent,universe,agent-flow,plan-delegate,a2a-stream-fallback", "comma-separated harness names under internal/harness; agent runs the provider tool-call conformance test")
 	timeoutFlag := flag.Duration("timeout", 10*time.Minute, "timeout per provider/harness run")
 	requireConfiguredFlag := flag.Bool("require-configured", false, "fail when a selected live provider is missing an API key")
 	capabilitiesFlag := flag.Bool("capabilities", true, "print the registered provider capability matrix before running conformance")
@@ -244,6 +244,9 @@ func validateSelection(providers, harnesses []string) error {
 	}
 
 	for _, harness := range harnesses {
+		if harness == "agent" {
+			continue
+		}
 		if strings.Contains(harness, string(os.PathSeparator)) || harness == "." || harness == ".." {
 			return fmt.Errorf("invalid harness name %q", harness)
 		}
@@ -323,6 +326,10 @@ func localRPCEnv(env []string) []string {
 }
 
 func runHarness(provider, harness string, timeout time.Duration) error {
+	if harness == "agent" {
+		return runAgentConformance(provider, timeout)
+	}
+
 	ctx, cancel := context.WithTimeout(context.Background(), timeout)
 	defer cancel()
 
@@ -358,3 +365,28 @@ func runHarness(provider, harness string, timeout time.Duration) error {
 	}
 	return nil
 }
+
+func runAgentConformance(provider string, timeout time.Duration) error {
+	ctx, cancel := context.WithTimeout(context.Background(), timeout)
+	defer cancel()
+
+	testProvider := provider
+	if provider == "mock" {
+		testProvider = "fake"
+	}
+	cmd := exec.CommandContext(ctx, "go", "test", "./agent", "-run", "TestAgentProviderConformanceMatrix", "-count=1", "-v")
+	cmd.Dir = repoRoot()
+	cmd.Stdout = os.Stdout
+	cmd.Stderr = os.Stderr
+	cmd.Env = localRPCEnv(append(os.Environ(),
+		"GO_MICRO_AGENT_CONFORMANCE_LIVE=1",
+		"GO_MICRO_AGENT_CONFORMANCE_PROVIDERS="+testProvider,
+	))
+	if err := cmd.Run(); err != nil {
+		if ctx.Err() == context.DeadlineExceeded {
+			return fmt.Errorf("timed out after %s", timeout)
+		}
+		return err
+	}
+	return nil
+}

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

@@ -4,13 +4,35 @@ This directory owns the no-secret reference scenario for the Go Micro
 services → agents → workflows lifecycle. It is intentionally small and
 scripted so CI can run it on every push without external services or model keys.
 
-`run.sh` verifies three boundaries together:
+`run.sh` verifies four boundaries together:
 
 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 flow runs` remains available for durable workflow run
-   history inspection.
+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.
+4. **Deploy** — `micro deploy --dry-run <target>` remains available as the
+   deployment-boundary checkpoint. The dry run resolves configured deploy targets
+   and services and prints the remote build/copy/systemd/health plan without
+   building binaries, opening SSH connections, running `rsync`, or touching
+   remote infrastructure.
 
 After the CLI boundary smoke checks, the script runs the deterministic harnesses
-that boot real services, agents, workflows, store-backed run history, and A2A
-with only the LLM mocked.
+that boot real services, agents, workflows, store-backed run history, plan/delegate,
+and A2A with only the LLM mocked.
+
+## Local and CI entry points
+
+The default GitHub harness workflow runs this script on every push and pull
+request after the 0→1 scaffold contract. Developers can run the same no-secret
+contract locally with:
+
+```sh
+make harness
+```
+
+That target intentionally exercises both 0→1 scaffold variants, the 0→hero
+scenario, the event-driven agent-flow harness, and mock provider conformance, so
+the public scaffold → run/chat → inspect → deploy lifecycle stays executable
+outside CI as well. Live provider checks remain separate and gated by configured
+API keys (`make provider-conformance` or the scheduled/manual CI job).

internal/harness/zero-to-hero-ci/run.sh

@@ -7,6 +7,7 @@ cd "$ROOT"
 # Keep the developer inner-loop boundaries executable and discoverable in CI
 # without secrets or long-running daemons.
 go test ./cmd/micro -run TestZeroToHeroCLIBoundaries -count=1
+go test ./cmd/micro/cli/deploy -run TestDeployDryRun -count=1
 
 # Deterministic no-secret reference scenarios. These use the real Go Micro
 # runtime and mock only the LLM provider.

internal/website/blog/5.md

@@ -11,7 +11,7 @@ description: "Unified service creation, cleaner handler registration, and modula
 
 *March 4, 2026 — By the Go Micro Team*
 
-Go Micro has always prioritized getting out of your way. But over time, the API accumulated multiple ways to do the same thing — `micro.NewService()`, `micro.NewService()`, `service.New()`, three different handler registration patterns. If you're building something for AI agents or running a modular monolith, you shouldn't have to choose between equivalent APIs.
+Go Micro has always prioritized getting out of your way. But over time, the API accumulated multiple ways to do the same thing — `micro.New(name)`, `micro.NewService(micro.Name(...))`, `service.New()`, three different handler registration patterns. If you're building something for AI agents or running a modular monolith, you shouldn't have to choose between equivalent APIs.
 
 We've cleaned it up. Here's what changed and why.
 
@@ -33,7 +33,7 @@ service := micro.NewService("greeter")
 service := micro.NewService("greeter", micro.Address(":8080"))
 ```
 
-Name is always the first argument. Options follow. `NewService` still works (it's deprecated, not removed), but every example, doc, and guide now uses `micro.NewService()`.
+Name is always the first argument. Options follow. `micro.New` still works as a deprecated alias, but every example, doc, and guide now uses `micro.NewService("name")`.
 
 ## Clean Handler Registration
 
@@ -107,7 +107,7 @@ Your Go comments become tool descriptions. Your struct tags become parameter sch
 
 If you're building new services, use `micro.NewService("name", opts...)` and `service.Handle()`. That's it.
 
-If you have existing code using `micro.NewService()` or `service.Server().Handle()`, everything still works — we didn't break anything. But the docs, examples, and guides all point to the new patterns now.
+If you have existing code using `micro.New("name")` or `service.Server().Handle()`, it still works. For v6, update name-less `micro.NewService(opts...)` calls to `micro.NewService("name", opts...)`. The docs, examples, and guides all point to the new patterns now.
 
 The goal is simple: when someone asks "how do I create a service?", there should be exactly one answer.
 

internal/website/docs/getting-started.md

@@ -16,7 +16,7 @@ Go Micro has three core abstractions:
 
 ## Prerequisites
 
-- **Go 1.21+** for development. The `curl` install below gives you the `micro` binary without Go, but `micro run` compiles your services, so you'll want Go installed to build them.
+- **Go 1.24+** for development. The `curl` install below gives you the `micro` binary without Go, but `micro run` compiles your services, so you'll want Go installed to build them.
 - An **LLM provider key** (Anthropic, OpenAI, Gemini, …) *only* for the AI features — `micro run --prompt`, `micro chat`, and agents. Plain services need no key. Set it before running, e.g. `export ANTHROPIC_API_KEY=sk-ant-...`.
 
 ## Install

internal/website/docs/guides/a2a-protocol.md

@@ -69,10 +69,18 @@ A card looks like:
   "capabilities": { "streaming": true, "pushNotifications": true },
   "defaultInputModes": ["text/plain"],
   "defaultOutputModes": ["text/plain"],
-  "skills": [{ "id": "chat", "name": "Chat", "tags": ["task", "project"] }]
+  "skills": [
+    { "id": "task", "name": "Task", "tags": ["task"] },
+    { "id": "project", "name": "Project", "tags": ["project"] }
+  ]
 }
 ```
 
+Each managed service is advertised as its own typed skill. Clients can call the
+whole agent at `/agents/task-mgr`, or address one skill directly at
+`/agents/task-mgr/skills/task`; the skill endpoint serves a focused card and
+routes the request to the same agent with that skill selected.
+
 ## Calling an agent
 
 A2A uses JSON-RPC 2.0 over HTTP. Send a message with `message/send`; the gateway runs the agent and returns a completed `Task`:

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

@@ -36,13 +36,54 @@ 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 (flows already do) | In progress |
+| Durable runs | Checkpoint and resume an agent run with the same checkpoint backend flows use | Shipped |
 | 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
+```
+
+Choose the boundary deliberately: use a durable flow when the steps are known
+(`reserve`, `charge`, `confirm`) and each step has deterministic retry/resume
+semantics. Use a checkpointed agent run when the model is deciding which tools to
+call or how many turns it needs, but the side effects of completed tool calls
+still need crash-safe resume. Flows and agents share the same `Checkpoint`
+interface, so a flow can safely dispatch to a checkpointed agent for the
+open-ended part.
+
+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,6 +83,44 @@ 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 | No |
+| `atlascloud` | Yes | Yes | Yes | Yes |
 | `gemini` | Yes | No | No | No |
-| `groq` | Yes | No | No | No |
-| `mistral` | Yes | No | No | No |
+| `groq` | Yes | No | No | Yes |
+| `mistral` | Yes | No | No | Yes |
 | `openai` | Yes | Yes | No | Yes |
-| `together` | Yes | No | No | No |
+| `together` | Yes | No | No | Yes |
 
 ## Step 1: Implement the `ai.Model` Interface
 

internal/website/docs/guides/plan-delegate.md

@@ -13,7 +13,7 @@ They are exposed to the model as ordinary tools. There is no separate graph runt
 
 ## Prerequisites
 
-- Go 1.21+
+- Go 1.24+
 - An API key for any supported provider (Anthropic, OpenAI, Gemini, Groq, Mistral, Together, Atlas Cloud)
 
 ```bash

internal/website/docs/roadmap.md

@@ -36,14 +36,13 @@ 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** — pause and resume mid-run (`input-required`), beyond the binary `ApproveTool` gate.
+- **Human-in-the-loop** — broaden pause/resume UX around `input-required` runs and approvals.
 - **A2A** — richer live-stream reconnection (`tasks/resubscribe`) and `input-required` handoffs.
 
 ## Developer experience (ongoing)

micro.go

@@ -23,12 +23,21 @@ 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
 
@@ -164,6 +173,32 @@ 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",