docs(priorities): refresh architect queue

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

.github/workflows/harness.yml

@@ -33,8 +33,8 @@ jobs:
         run: go run ./internal/harness/universe
       - name: Agent-flow harness
         run: go run ./internal/harness/agent-flow
-      - name: 0→hero plan-delegate workflow harness
-        run: go run ./internal/harness/plan-delegate
+      - name: 0→hero run/chat/inspect reference scenario
+        run: ./internal/harness/zero-to-hero-ci/run.sh
 
   harness-live:
     name: Provider harnesses (live LLM conformance)
@@ -75,6 +75,20 @@ jobs:
             -summary-json provider-conformance-summary.json \
             -summary-markdown provider-conformance-summary.md \
             -capabilities-markdown provider-capabilities.md
+      - name: Publish provider conformance summary
+        if: always()
+        run: |
+          if [ -f provider-conformance-summary.md ]; then
+            cat provider-conformance-summary.md >> "$GITHUB_STEP_SUMMARY"
+          fi
+          if [ -f provider-capabilities.md ]; then
+            {
+              echo
+              echo "## Registered provider capabilities"
+              echo
+              cat provider-capabilities.md
+            } >> "$GITHUB_STEP_SUMMARY"
+          fi
       - name: Upload provider conformance summary
         if: always()
         uses: actions/upload-artifact@v4

CHANGELOG.md

@@ -19,7 +19,7 @@ else is additive. See the [v5 → v6 migration guide](internal/website/docs/guid
 - **JWT auth ported in-module.** The external `github.com/micro/plugins/v5/auth/jwt` (pinned to v5) is replaced by `go-micro.dev/v6/auth/jwt/token`, now on the maintained `golang-jwt/jwt/v5`; the deprecated `dgrijalva/jwt-go` dependency is dropped.
 
 ### Added
-- **A2A protocol — both directions** — `gateway/a2a` exposes registered agents over the open Agent2Agent (A2A) protocol so agents on other frameworks can discover and call them: Agent Cards are generated from registry metadata (the same way the MCP gateway derives tools), and incoming tasks are translated to the agent's existing `Agent.Chat` RPC, with no per-agent code (`micro a2a serve`). The outbound `a2a.Client` calls external A2A agents by URL, wired into `flow.A2A(url)` (a workflow step) and `delegate` to an `http(s)` URL (from inside an agent). An agent can also serve A2A **directly** without a gateway via `AgentA2A(addr)` (`a2a.NewAgentHandler`), handling tasks in-process. The JSON-RPC binding includes `message/send`, `message/stream` (SSE), `tasks/get`, multi-turn continuation by `taskId`/`contextId`, best-effort push notification callbacks, and card discovery. `input-required` and `tasks/resubscribe` remain unsupported. (`gateway/a2a/`, `cmd/micro/a2a/`)
+- **A2A protocol — both directions** — `gateway/a2a` exposes registered agents over the open Agent2Agent (A2A) protocol so agents on other frameworks can discover and call them: Agent Cards are generated from registry metadata (the same way the MCP gateway derives tools), and incoming tasks are translated to the agent's existing `Agent.Chat` RPC, with no per-agent code (`micro a2a serve`). The outbound `a2a.Client` calls external A2A agents by URL, wired into `flow.A2A(url)` (a workflow step) and `delegate` to an `http(s)` URL (from inside an agent). An agent can also serve A2A **directly** without a gateway via `AgentA2A(addr)` (`a2a.NewAgentHandler`), handling tasks in-process. The JSON-RPC binding includes `message/send`, `message/stream` (SSE), `tasks/get`, multi-turn continuation by `taskId`/`contextId`, best-effort push notification callbacks, `tasks/resubscribe`, `input-required` handoffs, and card discovery. (`gateway/a2a/`, `cmd/micro/a2a/`)
 - **Agents (`micro.NewAgent`)** — an agent is a service with an LLM inside: it discovers its assigned services as tools, runs the model's tool loop, registers a `Chat` RPC endpoint, and is reachable like any service. `Ask` for programmatic use; `micro chat` discovers and routes to agents; `micro agent list`/`describe`. (`agent/`)
 - **Plan & delegate** — two built-in agent tools added to every agent: `plan` (an ordered, store-persisted plan surfaced back in the prompt) and `delegate` (hand a self-contained subtask to a registered agent over RPC, otherwise to an ephemeral sub-agent). No harness or graph — they're plain tools. (`agent/builtin.go`, `examples/agent-plan-delegate/`)
 - **Agent guardrails** — `MaxSteps` (stop on count), `LoopLimit` (stop repeated no-progress calls; on by default), and `ApproveTool` (human-in-the-loop / policy gate before each action), enforced at the one point every tool call passes through. (`agent/`, guide + blog)

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,14 @@ 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 TestZeroToOneContract -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
 
 # Run the same harnesses against every configured live provider. Providers
 # without API keys are skipped; configured providers must pass.

agent/agent.go

@@ -92,6 +92,11 @@ type agentImpl struct {
 	// Ask. The model provider only sees a refused tool result; the agent
 	// converts it into a durable paused run instead of completing the run.
 	pause *approvalPause
+
+	// currentRun points at the checkpoint record for the Ask currently
+	// holding mu. Tool execution updates it so resumed runs can reuse
+	// completed tool results without replaying side effects.
+	currentRun *flow.Run
 }
 
 // New creates a new Agent.
@@ -134,6 +139,10 @@ 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 != "" {
@@ -141,12 +150,19 @@ func (a *agentImpl) setup() {
 	}
 
 	a.tools = ai.NewTools(a.opts.Registry, ai.ToolClient(a.opts.Client))
-	modelOpts = append(modelOpts, ai.WithToolHandler(a.toolHandler()))
+	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.
@@ -202,17 +218,6 @@ func (a *agentImpl) Stream(ctx context.Context, message string) (ai.Stream, erro
 	})
 }
 
-// Resume returns the response for a checkpointed agent run. Completed runs are
-// returned from the checkpoint without calling the model or replaying tool
-// calls; failed or in-progress runs continue from the saved input message.
-func Resume(ctx context.Context, ag Agent, runID string) (*Response, error) {
-	a, ok := ag.(*agentImpl)
-	if !ok {
-		return nil, fmt.Errorf("agent resume: unsupported agent implementation %T", ag)
-	}
-	return a.resume(ctx, runID)
-}
-
 // Pending returns checkpointed agent runs that have not completed. It mirrors
 // flow.Pending for startup recovery loops that drain durable agent work.
 func Pending(ctx context.Context, ag Agent) ([]flow.Run, error) {
@@ -231,16 +236,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
@@ -253,6 +260,8 @@ func (a *agentImpl) askLocked(ctx context.Context, runID, message, parentRunID s
 		Agent:    a.opts.Name,
 	})
 	run := a.newCheckpointRun(runID, message, parentRunID, existing)
+	a.currentRun = &run
+	defer func() { a.currentRun = nil }()
 	if err := a.saveRun(ctx, run); err != nil {
 		return nil, err
 	}
@@ -283,6 +292,9 @@ func (a *agentImpl) askLocked(ctx context.Context, runID, message, parentRunID s
 		run.Status = "failed"
 		run.Steps[0].Status = "failed"
 		run.Steps[0].Error = err.Error()
+		if a.currentRun != nil {
+			run.Steps = a.currentRun.Steps
+		}
 		_ = a.saveRun(ctx, run)
 		return nil, err
 	}
@@ -290,6 +302,10 @@ func (a *agentImpl) askLocked(ctx context.Context, runID, message, parentRunID s
 		run.Status = "paused"
 		run.State.Stage = agentApprovalStep
 		run.State.Data = []byte(message)
+		if a.pause.Tool == toolHumanInput {
+			run.State.Stage = agentInputStep
+			_ = run.State.Set(inputPause{OriginalMessage: message, Prompt: a.pause.Message})
+		}
 		run.Steps[0].Status = "paused"
 		run.Steps[0].Error = a.pause.Message
 		run.Steps[0].Result = a.pause.Tool
@@ -326,6 +342,12 @@ func (a *agentImpl) askLocked(ctx context.Context, runID, message, parentRunID s
 	if b, marshalErr := json.Marshal(res); marshalErr == nil {
 		run.State.Data = b
 	}
+	if a.currentRun != nil {
+		run.Steps = a.currentRun.Steps
+	}
+	if len(run.Steps) == 0 {
+		run.Steps = []flow.StepRecord{{Name: agentAskStep}}
+	}
 	run.Steps[0].Status = "done"
 	run.Steps[0].Attempts++
 	run.Steps[0].Result = reply

agent/builtin.go

@@ -20,8 +20,9 @@ import (
 // the discovered service tools. There is no separate harness or graph:
 // the LLM calls them like any other tool.
 const (
-	toolPlan     = "plan"
-	toolDelegate = "delegate"
+	toolPlan       = "plan"
+	toolDelegate   = "delegate"
+	toolHumanInput = "request_input"
 )
 
 // builtinTools returns the tool definitions exposed to the model in
@@ -41,6 +42,18 @@ func builtinTools() []ai.Tool {
 				},
 			},
 		},
+		{
+			Name:         toolHumanInput,
+			OriginalName: toolHumanInput,
+			Description: "Pause this agent run when you need missing information, a decision, or other human input before you can continue. " +
+				"The run is checkpointed as input-required and can be resumed with the human response without losing completed tool history.",
+			Properties: map[string]any{
+				"prompt": map[string]any{
+					"type":        "string",
+					"description": "The specific question, decision, or instruction needed from the human operator.",
+				},
+			},
+		},
 		{
 			Name:         toolDelegate,
 			OriginalName: toolDelegate,
@@ -78,6 +91,9 @@ func Builtins(opts ...Option) (tools []ai.Tool, handle func(name string, input m
 		case toolPlan:
 			r := a.handlePlan(ai.ToolCall{Name: name, Input: input})
 			return r.Value, r.Content, true
+		case toolHumanInput:
+			r := a.handleHumanInput(ai.ToolCall{Name: name, Input: input})
+			return r.Value, r.Content, true
 		case toolDelegate:
 			r := a.handleDelegate(context.Background(), ai.ToolCall{Name: name, Input: input})
 			return r.Value, r.Content, true
@@ -102,8 +118,9 @@ func (a *agentImpl) toolHandler() ai.ToolHandler {
 
 	// Innermost first: base, then guardrails (approve → loop → step →
 	// plan), then developer wrappers outermost. Wrapping reverses order,
-	// so the result runs plan → step → loop → approve → base.
+	// so the result runs plan → step → loop → approve → checkpoint → base.
 	h := a.baseHandler()
+	h = a.checkpointToolWrap(h)
 	h = a.approveWrap(h)
 	h = a.loopWrap(h)
 	h = a.stepWrap(h)
@@ -145,6 +162,9 @@ func (a *agentImpl) baseHandler() ai.ToolHandler {
 				return ai.ToolResult{ID: call.ID, Value: out, Content: out}
 			}
 		}
+		if call.Name == toolHumanInput {
+			return a.handleHumanInput(call)
+		}
 		if call.Name == toolDelegate {
 			return a.handleDelegate(ctx, call)
 		}
@@ -205,6 +225,11 @@ type approvalPause struct {
 	Message string
 }
 
+type inputPause struct {
+	OriginalMessage string `json:"original_message"`
+	Prompt          string `json:"prompt"`
+}
+
 func (a *agentImpl) approveWrap(next ai.ToolHandler) ai.ToolHandler {
 	return func(ctx context.Context, call ai.ToolCall) ai.ToolResult {
 		if a.opts.Approve != nil {
@@ -232,6 +257,17 @@ func (a *agentImpl) handlePlan(call ai.ToolCall) ai.ToolResult {
 	return ai.ToolResult{ID: call.ID, Value: call.Input, Content: string(data)}
 }
 
+// handleHumanInput records that the model needs operator input before it can continue.
+func (a *agentImpl) handleHumanInput(call ai.ToolCall) ai.ToolResult {
+	prompt, _ := call.Input["prompt"].(string)
+	prompt = strings.TrimSpace(prompt)
+	if prompt == "" {
+		prompt = "human input required"
+	}
+	a.pause = &approvalPause{Tool: toolHumanInput, Message: prompt}
+	return refused(call.ID, ai.RefusedApproval, "input-required: "+prompt)
+}
+
 // handleDelegate hands a subtask to another agent. Delegate-first:
 // if 'to' names a registered agent, it is called via RPC. Otherwise an
 // ephemeral sub-agent is created with a fresh, isolated context, asked

agent/builtin_test.go

@@ -11,15 +11,15 @@ import (
 
 func TestBuiltinTools(t *testing.T) {
 	tools := builtinTools()
-	if len(tools) != 2 {
-		t.Fatalf("builtinTools() = %d tools, want 2", len(tools))
+	if len(tools) != 3 {
+		t.Fatalf("builtinTools() = %d tools, want 3", len(tools))
 	}
 	names := map[string]bool{}
 	for _, tl := range tools {
 		names[tl.Name] = true
 	}
-	if !names[toolPlan] || !names[toolDelegate] {
-		t.Errorf("builtin tools = %v, want plan and delegate", names)
+	if !names[toolPlan] || !names[toolDelegate] || !names[toolHumanInput] {
+		t.Errorf("builtin tools = %v, want plan, request_input, and delegate", names)
 	}
 }
 
@@ -109,8 +109,8 @@ func TestBuiltinsAccessor(t *testing.T) {
 		WithRegistry(registry.NewMemoryRegistry()),
 	)
 
-	if len(tools) != 2 {
-		t.Fatalf("Builtins() returned %d tools, want 2", len(tools))
+	if len(tools) != 3 {
+		t.Fatalf("Builtins() returned %d tools, want 3", len(tools))
 	}
 
 	// A name that isn't a built-in falls through (ok == false).

agent/checkpoint.go

@@ -6,12 +6,14 @@ import (
 	"fmt"
 	"time"
 
+	"go-micro.dev/v6/ai"
 	"go-micro.dev/v6/flow"
 )
 
 const (
 	agentAskStep      = "ask"
 	agentApprovalStep = "approval"
+	agentInputStep    = "input-required"
 )
 
 func (a *agentImpl) newCheckpointRun(runID, message, parentRunID string, existing *flow.Run) flow.Run {
@@ -35,6 +37,7 @@ func (a *agentImpl) newCheckpointRun(runID, message, parentRunID string, existin
 		}
 		run.Steps[0].Status = "in_progress"
 		run.Steps[0].Error = ""
+		run.Steps[0].Result = ""
 	}
 	return run
 }
@@ -49,6 +52,17 @@ func (a *agentImpl) saveRun(ctx context.Context, run flow.Run) error {
 	return nil
 }
 
+// Resume returns the response for a checkpointed agent run. Completed runs are
+// returned from the checkpoint without calling the model or replaying tool
+// calls; failed or in-progress runs continue from the saved input message.
+func Resume(ctx context.Context, ag Agent, runID string) (*Response, error) {
+	a, ok := ag.(*agentImpl)
+	if !ok {
+		return nil, fmt.Errorf("agent resume: unsupported agent implementation %T", ag)
+	}
+	return a.resume(ctx, runID)
+}
+
 func (a *agentImpl) resume(ctx context.Context, runID string) (*Response, error) {
 	if a.opts.Checkpoint == nil {
 		return nil, fmt.Errorf("agent %s has no checkpoint configured", a.opts.Name)
@@ -61,6 +75,9 @@ func (a *agentImpl) resume(ctx context.Context, runID string) (*Response, error)
 		return nil, fmt.Errorf("agent run %s not found", runID)
 	}
 	if run.Status == "paused" {
+		if run.State.Stage == agentInputStep {
+			return nil, fmt.Errorf("agent run %s is input-required; resume with ResumeInput", runID)
+		}
 		run.Status = "running"
 		run.State.Stage = agentAskStep
 	}
@@ -78,7 +95,52 @@ 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
+// request_input tool. The supplied input is appended to the original request so
+// the same run can continue with durable checkpoint and completed tool history.
+func ResumeInput(ctx context.Context, ag Agent, runID, input string) (*Response, error) {
+	a, ok := ag.(*agentImpl)
+	if !ok {
+		return nil, fmt.Errorf("agent resume input: unsupported agent implementation %T", ag)
+	}
+	return a.resumeInput(ctx, runID, input)
+}
+
+func (a *agentImpl) resumeInput(ctx context.Context, runID, input string) (*Response, error) {
+	if a.opts.Checkpoint == nil {
+		return nil, fmt.Errorf("agent %s has no checkpoint configured", a.opts.Name)
+	}
+	run, ok, err := a.opts.Checkpoint.Load(ctx, runID)
+	if err != nil {
+		return nil, err
+	}
+	if !ok {
+		return nil, fmt.Errorf("agent run %s not found", runID)
+	}
+	if run.Status != "paused" || run.State.Stage != agentInputStep {
+		return nil, fmt.Errorf("agent run %s is not waiting for human input", runID)
+	}
+	var p inputPause
+	if err := run.State.Scan(&p); err != nil {
+		return nil, fmt.Errorf("agent run %s input state decode: %w", runID, err)
+	}
+	message := p.OriginalMessage
+	if message == "" {
+		message = string(run.State.Data)
+	}
+	message += "\n\nHuman input: " + input
+	run.Status = "running"
+	run.State.Stage = agentAskStep
+	run.State.Data = []byte(message)
+	a.mu.Lock()
+	defer a.mu.Unlock()
+	if a.model == nil {
+		a.setup()
+	}
+	return a.askLocked(ctx, run.ID, message, run.ParentID, &run, true)
 }
 
 func (a *agentImpl) pending(ctx context.Context) ([]flow.Run, error) {
@@ -97,3 +159,60 @@ func (a *agentImpl) pending(ctx context.Context) ([]flow.Run, error) {
 	}
 	return out, nil
 }
+
+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 {
+			return next(ctx, call)
+		}
+		name := toolCheckpointName(call)
+		if rec, ok := findStep(a.currentRun.Steps, name); ok && rec.Status == "done" {
+			return ai.ToolResult{ID: call.ID, Value: rec.Result, Content: rec.Result}
+		}
+
+		idx := upsertStep(&a.currentRun.Steps, flow.StepRecord{Name: name, Status: "in_progress"})
+		_ = a.saveRun(ctx, *a.currentRun)
+		res := next(ctx, call)
+		a.currentRun.Steps[idx].Attempts++
+		if res.Refused != "" {
+			a.currentRun.Steps[idx].Status = "failed"
+			a.currentRun.Steps[idx].Error = res.Content
+			_ = a.saveRun(ctx, *a.currentRun)
+			return res
+		}
+		a.currentRun.Steps[idx].Status = "done"
+		a.currentRun.Steps[idx].Result = res.Content
+		a.currentRun.Steps[idx].Error = ""
+		_ = a.saveRun(ctx, *a.currentRun)
+		return res
+	}
+}
+
+func toolCheckpointName(call ai.ToolCall) string {
+	b, _ := json.Marshal(call.Input)
+	return "tool:" + call.Name + ":" + string(b)
+}
+
+func findStep(steps []flow.StepRecord, name string) (flow.StepRecord, bool) {
+	for _, step := range steps {
+		if step.Name == name {
+			return step, true
+		}
+	}
+	return flow.StepRecord{}, false
+}
+
+func upsertStep(steps *[]flow.StepRecord, rec flow.StepRecord) int {
+	for i := range *steps {
+		if (*steps)[i].Name == rec.Name {
+			(*steps)[i].Status = rec.Status
+			(*steps)[i].Error = rec.Error
+			return i
+		}
+	}
+	if len(*steps) == 0 || (*steps)[0].Name != agentAskStep {
+		*steps = append([]flow.StepRecord{{Name: agentAskStep, Status: "in_progress"}}, (*steps)...)
+	}
+	*steps = append(*steps, rec)
+	return len(*steps) - 1
+}

agent/checkpoint_test.go

@@ -2,6 +2,8 @@ package agent
 
 import (
 	"context"
+	"errors"
+	"strings"
 	"testing"
 
 	"go-micro.dev/v6/ai"
@@ -48,6 +50,124 @@ func TestResumeCompletedCheckpointDoesNotReplayModel(t *testing.T) {
 	}
 }
 
+func TestResumeFailedCheckpointDoesNotReplayCompletedTool(t *testing.T) {
+	ctx := context.Background()
+	cp := flow.StoreCheckpoint(store.NewStore(), "tool-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: "external.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("model connection dropped after tool")
+		}
+		return &ai.Response{Reply: "finished from checkpoint"}, nil
+	}
+	defer func() { fakeGen = nil }()
+
+	a := newTestAgent(Name("tool-resume-agent"), WithCheckpoint(cp),
+		WithTool("external.charge", "charge once", nil, func(context.Context, map[string]any) (string, error) {
+			toolRuns++
+			return "charged", nil
+		}))
+	_, err := a.Ask(ctx, "charge order 42")
+	if err == nil {
+		t.Fatal("Ask succeeded, want simulated failure")
+	}
+	if toolRuns != 1 {
+		t.Fatalf("tool executions after failed Ask = %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))
+	}
+	resp, err := Resume(ctx, a, runs[0].ID)
+	if err != nil {
+		t.Fatalf("Resume: %v", err)
+	}
+	if resp.Reply != "finished from checkpoint" {
+		t.Fatalf("Resume reply = %q", resp.Reply)
+	}
+	if toolRuns != 1 {
+		t.Fatalf("tool executions after Resume = %d, want completed tool was not replayed", toolRuns)
+	}
+}
+
+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")
@@ -65,6 +185,64 @@ func TestPendingReturnsUnfinishedAgentRuns(t *testing.T) {
 	}
 }
 
+func TestHumanInputPauseResumesSameRunWithInput(t *testing.T) {
+	ctx := context.Background()
+	cp := flow.StoreCheckpoint(store.NewStore(), "input-agent")
+	calls := 0
+	fakeGen = func(ctx context.Context, opts ai.Options, req *ai.Request) (*ai.Response, error) {
+		calls++
+		if calls == 1 {
+			if opts.ToolHandler != nil {
+				opts.ToolHandler(ctx, ai.ToolCall{ID: "input-1", Name: toolHumanInput, Input: map[string]any{"prompt": "Which region should I deploy to?"}})
+			}
+			return &ai.Response{Reply: "waiting"}, nil
+		}
+		if !strings.Contains(req.Prompt, "Human input: us-east-1") {
+			t.Fatalf("resumed prompt = %q, want human input", req.Prompt)
+		}
+		return &ai.Response{Reply: "deploying to us-east-1"}, nil
+	}
+	defer func() { fakeGen = nil }()
+
+	a := newTestAgent(Name("input-agent"), WithCheckpoint(cp))
+	_, err := a.Ask(ctx, "deploy the service")
+	if 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 || runs[0].Status != "paused" || runs[0].State.Stage != agentInputStep {
+		t.Fatalf("paused runs = %#v, want one input-required run", runs)
+	}
+	var pause inputPause
+	if err := runs[0].State.Scan(&pause); err != nil {
+		t.Fatalf("Scan pause: %v", err)
+	}
+	if pause.OriginalMessage != "deploy the service" || pause.Prompt != "Which region should I deploy to?" {
+		t.Fatalf("pause = %#v", pause)
+	}
+
+	if _, err := Resume(ctx, a, runs[0].ID); err == nil || !strings.Contains(err.Error(), "ResumeInput") {
+		t.Fatalf("Resume input-required err = %v, want guidance", err)
+	}
+	resp, err := ResumeInput(ctx, a, runs[0].ID, "us-east-1")
+	if err != nil {
+		t.Fatalf("ResumeInput: %v", err)
+	}
+	if resp.RunID != runs[0].ID || resp.Reply != "deploying to us-east-1" {
+		t.Fatalf("response = %#v", resp)
+	}
+	loaded, ok, err := cp.Load(ctx, runs[0].ID)
+	if err != nil || !ok {
+		t.Fatalf("Load resumed run ok=%v err=%v", ok, err)
+	}
+	if loaded.Status != "done" {
+		t.Fatalf("resumed run status = %q, want done", loaded.Status)
+	}
+}
+
 func TestApprovalDenialPausesCheckpointedRunAndResumeContinues(t *testing.T) {
 	ctx := context.Background()
 	cp := flow.StoreCheckpoint(store.NewStore(), "approval-agent")

agent/memory.go

@@ -3,6 +3,7 @@ package agent
 import (
 	"encoding/json"
 	"fmt"
+	"sort"
 	"strings"
 	"sync"
 
@@ -123,17 +124,31 @@ func (m *storeMemory) Recall(query string, limit int) []ai.Message {
 		limit = 5
 	}
 	terms := recallTerms(query)
-	var out []ai.Message
-	for i := len(m.archive) - 1; i >= 0 && len(out) < limit; i-- {
+	type match struct {
+		msg   ai.Message
+		score int
+		index int
+	}
+	matches := make([]match, 0, len(m.archive))
+	for i := len(m.archive) - 1; i >= 0; i-- {
 		msg := m.archive[i]
-		text := strings.ToLower(fmt.Sprint(msg.Content))
-		for _, term := range terms {
-			if strings.Contains(text, term) {
-				out = append(out, msg)
-				break
-			}
+		if score := recallScore(msg, terms); score > 0 {
+			matches = append(matches, match{msg: msg, score: score, index: i})
 		}
 	}
+	sort.SliceStable(matches, func(i, j int) bool {
+		if matches[i].score != matches[j].score {
+			return matches[i].score > matches[j].score
+		}
+		return matches[i].index > matches[j].index
+	})
+	if len(matches) > limit {
+		matches = matches[:limit]
+	}
+	out := make([]ai.Message, 0, len(matches))
+	for _, match := range matches {
+		out = append(out, match.msg)
+	}
 	return out
 }
 
@@ -228,6 +243,17 @@ func compactText(s string, max int) string {
 	return s
 }
 
+func recallScore(msg ai.Message, terms []string) int {
+	text := strings.ToLower(fmt.Sprint(msg.Content))
+	score := 0
+	for _, term := range terms {
+		if strings.Contains(text, term) {
+			score++
+		}
+	}
+	return score
+}
+
 func recallTerms(query string) []string {
 	seen := map[string]bool{}
 	var terms []string

agent/memory_test.go

@@ -62,6 +62,46 @@ func TestWithMemoryUsed(t *testing.T) {
 	}
 }
 
+func TestCompactingMemoryRecallRanksSpecificMatches(t *testing.T) {
+	m := NewCompactingMemory(store.NewMemoryStore(), "agent/rank/history", 3, 1).(MemoryRecall)
+	writer := m.(Memory)
+	writer.Add("user", "alpha budget is 42")
+	writer.Add("assistant", "noted")
+	writer.Add("user", "beta budget is 7")
+	writer.Add("assistant", "noted")
+	writer.Add("user", "alpha owner is sam")
+
+	recalled := m.Recall("alpha budget", 2)
+	if len(recalled) == 0 {
+		t.Fatal("expected recalled messages")
+	}
+	if got := recalled[0].Content.(string); !strings.Contains(got, "alpha budget is 42") {
+		t.Fatalf("top recall = %q, want alpha budget match", got)
+	}
+}
+
+func TestCompactingMemoryArchivePersistsAndReloads(t *testing.T) {
+	st := store.NewMemoryStore()
+	m := NewCompactingMemory(st, "agent/reload/history", 3, 1)
+	m.Add("user", "alpha budget is 42")
+	m.Add("assistant", "noted")
+	m.Add("user", "beta budget is 7")
+	m.Add("assistant", "noted")
+
+	reloaded := NewCompactingMemory(st, "agent/reload/history", 3, 1)
+	recall, ok := reloaded.(MemoryRecall)
+	if !ok {
+		t.Fatal("compacting memory should support recall")
+	}
+	recalled := recall.Recall("alpha budget", 1)
+	if len(recalled) != 1 {
+		t.Fatalf("recalled %d messages, want 1", len(recalled))
+	}
+	if got := recalled[0].Content.(string); !strings.Contains(got, "alpha budget is 42") {
+		t.Fatalf("reloaded recall = %q, want alpha budget", got)
+	}
+}
+
 // A custom tool is offered to the model and dispatched to its handler.
 func TestWithToolExposedAndDispatched(t *testing.T) {
 	var got map[string]any

agent/otel.go

@@ -31,6 +31,8 @@ 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"
@@ -38,20 +40,22 @@ const (
 )
 
 type RunEvent struct {
-	Time      time.Time `json:"time"`
-	RunID     string    `json:"run_id"`
-	ParentID  string    `json:"parent_id,omitempty"`
-	TraceID   string    `json:"trace_id,omitempty"`
-	SpanID    string    `json:"span_id,omitempty"`
-	Agent     string    `json:"agent"`
-	Kind      string    `json:"kind"`
-	Name      string    `json:"name,omitempty"`
-	Provider  string    `json:"provider,omitempty"`
-	Model     string    `json:"model,omitempty"`
-	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"`
 }
 
 type Usage = ai.Usage
@@ -144,7 +148,7 @@ func (m *tracedModel) Generate(ctx context.Context, req *ai.Request, opts ...ai.
 		if resp != nil {
 			usage = resp.Usage
 		}
-		e := RunEvent{Time: time.Now(), RunID: info.RunID, ParentID: info.ParentID, Agent: info.Agent, Kind: "model", Provider: provider, Model: model, 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()
 		}
@@ -162,6 +166,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
@@ -175,7 +185,7 @@ func (m *tracedModel) Generate(ctx context.Context, req *ai.Request, opts ...ai.
 		span.SetStatus(codes.Ok, "")
 	}
 	span.End()
-	e := RunEvent{Time: time.Now(), RunID: info.RunID, ParentID: info.ParentID, Agent: info.Agent, Kind: "model", Provider: provider, Model: model, 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()
 	}
@@ -253,9 +263,46 @@ func (a *agentImpl) recordSpanEvent(span trace.Span, e RunEvent) {
 		e.TraceID = sc.TraceID().String()
 		e.SpanID = sc.SpanID().String()
 	}
+	span.AddEvent("agent."+e.Kind, trace.WithTimestamp(e.Time), trace.WithAttributes(runEventAttributes(e)...))
 	a.recordRunEvent(e)
 }
 
+func runEventAttributes(e RunEvent) []attribute.KeyValue {
+	attrs := []attribute.KeyValue{
+		attribute.String(AttrRunID, e.RunID),
+		attribute.String(AttrAgentName, e.Agent),
+	}
+	if e.ParentID != "" {
+		attrs = append(attrs, attribute.String(AttrParentRunID, e.ParentID))
+	}
+	if e.Name != "" {
+		attrs = append(attrs, attribute.String("agent.event.name", e.Name))
+	}
+	if e.Provider != "" {
+		attrs = append(attrs, attribute.String(AttrProvider, e.Provider))
+	}
+	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))
+	}
+	attrs = appendUsage(attrs, e.Tokens)
+	if e.Refused != "" {
+		attrs = append(attrs, attribute.Bool(AttrGuardrailBlock, true), attribute.String(AttrRefusal, e.Refused))
+	}
+	if e.Error != "" {
+		attrs = append(attrs, attribute.String("agent.error", e.Error))
+	}
+	return attrs
+}
+
 func (a *agentImpl) recordRunEvent(e RunEvent) {
 	if e.RunID == "" {
 		return

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 {
@@ -71,6 +75,16 @@ func TestAgentOpenTelemetrySpans(t *testing.T) {
 	if runID == "" {
 		t.Fatal("run span missing run id attribute")
 	}
+	var runEvents []trace.Event
+	for _, s := range spans {
+		if s.Name() == spanNameRun {
+			runEvents = s.Events()
+			break
+		}
+	}
+	if !spanEventHasRunInfo(runEvents, "agent.run", runID, "runner") || !spanEventHasRunInfo(runEvents, "agent.done", runID, "runner") {
+		t.Fatalf("run span missing run-info events: %#v", runEvents)
+	}
 	for _, s := range spans {
 		if s.Name() != spanNameModelCall && s.Name() != spanNameToolCall {
 			continue
@@ -79,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 {
@@ -115,6 +132,90 @@ func TestAgentOpenTelemetrySpans(t *testing.T) {
 	}
 }
 
+type failingOtelModel struct{ opts ai.Options }
+
+func (m *failingOtelModel) Init(opts ...ai.Option) error {
+	for _, o := range opts {
+		o(&m.opts)
+	}
+	return nil
+}
+func (m *failingOtelModel) Options() ai.Options { return m.opts }
+func (m *failingOtelModel) String() string      { return "otelfail" }
+func (m *failingOtelModel) Stream(context.Context, *ai.Request, ...ai.GenerateOption) (ai.Stream, error) {
+	return nil, nil
+}
+func (m *failingOtelModel) Generate(context.Context, *ai.Request, ...ai.GenerateOption) (*ai.Response, error) {
+	return nil, errors.New("provider exploded")
+}
+
+func init() {
+	ai.Register("otelfail", func(opts ...ai.Option) ai.Model { return &failingOtelModel{opts: ai.NewOptions(opts...)} })
+}
+
+func TestAgentOpenTelemetrySpansModelFailure(t *testing.T) {
+	exp := tracetest.NewInMemoryExporter()
+	tp := trace.NewTracerProvider(trace.WithSyncer(exp))
+	st := store.NewMemoryStore()
+	a := New(Name("failing-runner"), Provider("otelfail"), WithStore(st), TraceProvider(tp))
+	if _, err := a.Ask(context.Background(), "hello"); err == nil {
+		t.Fatal("Ask succeeded, want provider error")
+	}
+
+	spans := exp.GetSpans().Snapshots()
+	var sawRunError, sawModelError bool
+	for _, s := range spans {
+		attrs := spanAttributes(s.Attributes())
+		switch s.Name() {
+		case spanNameRun:
+			if attrs[AttrAgentName] == "failing-runner" && s.Status().Code == codesError {
+				sawRunError = true
+			}
+		case spanNameModelCall:
+			if attrs[AttrAgentName] == "failing-runner" && attrs[AttrAttempt] == "1" && s.Status().Code == codesError {
+				sawModelError = true
+			}
+		}
+	}
+	if !sawRunError || !sawModelError {
+		t.Fatalf("missing error spans: run=%v model=%v spans=%d", sawRunError, sawModelError, len(spans))
+	}
+
+	summaries, err := ListRunSummaries(st, "failing-runner")
+	if err != nil {
+		t.Fatal(err)
+	}
+	if len(summaries) != 1 || summaries[0].Status != "error" || summaries[0].LastError == "" {
+		t.Fatalf("unexpected failure summary: %#v", summaries)
+	}
+	events, err := LoadRunEvents(st, "failing-runner", summaries[0].RunID)
+	if err != nil {
+		t.Fatal(err)
+	}
+	var sawModelEvent bool
+	for _, event := range events {
+		if event.Kind == "model" && event.Attempt == 1 && event.MaxAttempts == 1 && event.Error != "" {
+			sawModelEvent = true
+		}
+	}
+	if !sawModelEvent {
+		t.Fatalf("missing failed model event with attempt metadata: %#v", events)
+	}
+}
+
+func spanEventHasRunInfo(events []trace.Event, name, runID, agentName string) bool {
+	for _, event := range events {
+		if event.Name != name {
+			continue
+		}
+		attrs := spanAttributes(event.Attributes)
+		if attrs[AttrRunID] == runID && attrs[AttrAgentName] == agentName {
+			return true
+		}
+	}
+	return false
+}
+
 func spanAttributes(attrs []attribute.KeyValue) map[string]string {
 	out := make(map[string]string, len(attrs))
 	for _, attr := range attrs {

agent/stream.go

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

agent/stream_test.go

@@ -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/anthropic/anthropic.go

@@ -77,11 +77,9 @@ func (p *Provider) Generate(ctx context.Context, req *ai.Request, opts ...ai.Gen
 	// Build initial request
 	apiReq := map[string]any{
 		"model":      p.opts.Model,
-		"max_tokens": 8192,
+		"max_tokens": anthropicMaxTokens(p.opts),
 		"system":     req.SystemPrompt,
-		"messages": []map[string]any{
-			{"role": "user", "content": req.Prompt},
-		},
+		"messages":   threadAnthropicMessages(req),
 	}
 
 	if len(anthropicTools) > 0 {
@@ -101,10 +99,9 @@ func (p *Provider) Generate(ctx context.Context, req *ai.Request, opts ...ai.Gen
 
 	// Tool execution loop: execute tools, send results back, repeat
 	// until the model responds with text only (no more tool calls)
-	messages := []map[string]any{
-		{"role": "user", "content": req.Prompt},
-		{"role": "assistant", "content": cleanContent(rawContent)},
-	}
+	messages := append(threadAnthropicMessages(req),
+		map[string]any{"role": "assistant", "content": cleanContent(rawContent)},
+	)
 
 	pendingCalls := resp.ToolCalls
 
@@ -127,7 +124,7 @@ func (p *Provider) Generate(ctx context.Context, req *ai.Request, opts ...ai.Gen
 
 		followUpReq := map[string]any{
 			"model":      p.opts.Model,
-			"max_tokens": 8192,
+			"max_tokens": anthropicMaxTokens(p.opts),
 			"system":     req.SystemPrompt,
 			"messages":   messages,
 		}
@@ -270,3 +267,24 @@ func cleanContent(raw any) any {
 	}
 	return cleaned
 }
+
+// threadAnthropicMessages builds the Anthropic messages array from the
+// conversation history (req.Messages) followed by the current prompt. The
+// system prompt is sent separately via the top-level "system" field.
+func threadAnthropicMessages(req *ai.Request) []map[string]any {
+	msgs := make([]map[string]any, 0, len(req.Messages)+1)
+	for _, m := range req.Messages {
+		msgs = append(msgs, map[string]any{"role": m.Role, "content": m.Content})
+	}
+	if req.Prompt != "" {
+		msgs = append(msgs, map[string]any{"role": "user", "content": req.Prompt})
+	}
+	return msgs
+}
+
+func anthropicMaxTokens(o ai.Options) int {
+	if o.MaxTokens > 0 {
+		return o.MaxTokens
+	}
+	return 8192
+}

ai/atlascloud/atlascloud.go

@@ -20,6 +20,7 @@
 package atlascloud
 
 import (
+	"bufio"
 	"bytes"
 	"context"
 	"encoding/json"
@@ -42,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.
@@ -91,13 +93,21 @@ func (p *Provider) Generate(ctx context.Context, req *ai.Request, opts ...ai.Gen
 
 	messages := []map[string]any{
 		{"role": "system", "content": req.SystemPrompt},
-		{"role": "user", "content": req.Prompt},
+	}
+	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":    p.opts.Model,
 		"messages": messages,
 	}
+	if p.opts.MaxTokens > 0 {
+		apiReq["max_tokens"] = p.opts.MaxTokens
+	}
 
 	if len(tools) > 0 {
 		apiReq["tools"] = tools
@@ -142,8 +152,111 @@ func (p *Provider) Generate(ctx context.Context, req *ai.Request, opts ...ai.Gen
 	return resp, nil
 }
 
+// Stream generates a streaming response from Atlas Cloud's OpenAI-compatible
+// chat completions endpoint, emitting content deltas as they arrive.
 func (p *Provider) Stream(ctx context.Context, req *ai.Request, opts ...ai.GenerateOption) (ai.Stream, error) {
-	return nil, fmt.Errorf("%w: atlascloud provider", ai.ErrStreamingUnsupported)
+	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":          p.opts.Model,
+		"messages":       messages,
+		"stream":         true,
+		"stream_options": map[string]any{"include_usage": true},
+	}
+	if p.opts.MaxTokens > 0 {
+		apiReq["max_tokens"] = p.opts.MaxTokens
+	}
+	reqBody, err := json.Marshal(apiReq)
+	if err != nil {
+		return nil, fmt.Errorf("failed to marshal stream request: %w", err)
+	}
+	apiURL := strings.TrimRight(p.opts.BaseURL, "/") + "/v1/chat/completions"
+	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 "+p.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 &atlasStream{body: httpResp.Body, scanner: bufio.NewScanner(httpResp.Body)}, nil
+}
+
+type atlasStream struct {
+	body    io.ReadCloser
+	scanner *bufio.Scanner
+	closed  bool
+}
+
+func (s *atlasStream) 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
+		}
+		// Final chunk (after include_usage) carries token usage and no content.
+		if chunk.Usage != nil {
+			return &ai.Response{Usage: ai.Usage{
+				InputTokens:  chunk.Usage.PromptTokens,
+				OutputTokens: chunk.Usage.CompletionTokens,
+				TotalTokens:  chunk.Usage.TotalTokens,
+			}}, nil
+		}
+		continue
+	}
+	if err := s.scanner.Err(); err != nil {
+		return nil, err
+	}
+	return nil, io.EOF
+}
+
+func (s *atlasStream) Close() error {
+	if s.closed {
+		return nil
+	}
+	s.closed = true
+	return s.body.Close()
 }
 
 func (p *Provider) callAPI(ctx context.Context, req map[string]any) (*ai.Response, map[string]any, error) {

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/openai/openai.go

@@ -85,7 +85,12 @@ func (p *Provider) Generate(ctx context.Context, req *ai.Request, opts ...ai.Gen
 	// Build messages
 	messages := []map[string]any{
 		{"role": "system", "content": req.SystemPrompt},
-		{"role": "user", "content": req.Prompt},
+	}
+	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})
 	}
 
 	// Build initial request
@@ -93,6 +98,9 @@ func (p *Provider) Generate(ctx context.Context, req *ai.Request, opts ...ai.Gen
 		"model":    p.opts.Model,
 		"messages": messages,
 	}
+	if p.opts.MaxTokens > 0 {
+		apiReq["max_tokens"] = p.opts.MaxTokens
+	}
 
 	if len(openaiTools) > 0 {
 		apiReq["tools"] = openaiTools
@@ -146,12 +154,21 @@ 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) {
 	messages := []map[string]any{
 		{"role": "system", "content": req.SystemPrompt},
-		{"role": "user", "content": req.Prompt},
+	}
+	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":    p.opts.Model,
-		"messages": messages,
-		"stream":   true,
+		"model":          p.opts.Model,
+		"messages":       messages,
+		"stream":         true,
+		"stream_options": map[string]any{"include_usage": true},
+	}
+	if p.opts.MaxTokens > 0 {
+		apiReq["max_tokens"] = p.opts.MaxTokens
 	}
 	reqBody, err := json.Marshal(apiReq)
 	if err != nil {
@@ -203,14 +220,27 @@ func (s *openAIStream) Recv() (*ai.Response, error) {
 					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 == "" {
-			continue
+		if len(chunk.Choices) > 0 && chunk.Choices[0].Delta.Content != "" {
+			return &ai.Response{Reply: chunk.Choices[0].Delta.Content}, nil
 		}
-		return &ai.Response{Reply: chunk.Choices[0].Delta.Content}, nil
+		// Final chunk (after include_usage) carries token usage and no content.
+		if chunk.Usage != nil {
+			return &ai.Response{Usage: ai.Usage{
+				InputTokens:  chunk.Usage.PromptTokens,
+				OutputTokens: chunk.Usage.CompletionTokens,
+				TotalTokens:  chunk.Usage.TotalTokens,
+			}}, nil
+		}
+		continue
 	}
 	if err := s.scanner.Err(); err != nil {
 		return nil, err

ai/openai/openai_test.go

@@ -8,6 +8,7 @@ import (
 	"net/http"
 	"net/http/httptest"
 	"testing"
+	"time"
 
 	"go-micro.dev/v6/ai"
 )
@@ -126,6 +127,58 @@ func TestProvider_Stream(t *testing.T) {
 	}
 }
 
+func TestProvider_StreamPropagatesMalformedChunk(t *testing.T) {
+	ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		w.Header().Set("Content-Type", "text/event-stream")
+		_, _ = w.Write([]byte("data: {bad json}\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 _, err := stream.Recv(); err == nil {
+		t.Fatal("Recv returned nil error for malformed chunk")
+	}
+}
+
+func TestProvider_StreamCloseReleasesResponse(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()
+
+	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)
+	}
+	first, err := stream.Recv()
+	if err != nil || first.Reply != "hel" {
+		t.Fatalf("first chunk = %#v, %v; want hel", first, err)
+	}
+	if err := stream.Close(); err != nil {
+		t.Fatalf("Close returned error: %v", err)
+	}
+
+	select {
+	case <-released:
+	case <-time.After(time.Second):
+		t.Fatal("server did not observe closed stream request")
+	}
+}
+
 func TestProvider_ImageRegistration(t *testing.T) {
 	ig := ai.NewImage("openai", ai.WithAPIKey("test"))
 	if ig == nil {

ai/options.go

@@ -16,6 +16,8 @@ type Options struct {
 	BaseURL string
 	// ToolHandler handles tool calls (optional, for automatic tool execution)
 	ToolHandler ToolHandler
+	// MaxTokens caps the length of the response (0 = provider default)
+	MaxTokens int
 }
 
 // GenerateOptions for generate call
@@ -91,3 +93,11 @@ func WithTools(t *Tools) Option {
 		}
 	}
 }
+
+// WithMaxTokens caps the number of tokens in the response. 0 leaves the
+// provider default in place.
+func WithMaxTokens(n int) Option {
+	return func(o *Options) {
+		o.MaxTokens = n
+	}
+}

ai/retry.go

@@ -13,6 +13,12 @@ type StatusCoder interface {
 	StatusCode() int
 }
 
+// RetryAfterCoder is implemented by provider errors that expose a server
+// supplied retry delay, such as HTTP Retry-After on a 429/503 response.
+type RetryAfterCoder interface {
+	RetryAfter() time.Duration
+}
+
 // ErrorKind classifies provider-boundary failures into stable buckets callers
 // can inspect without parsing provider-specific error strings.
 type ErrorKind string
@@ -113,16 +119,7 @@ func GenerateWithRetry(ctx context.Context, m Model, req *Request, policy Genera
 		// Always back off between retries — exponential and capped — so an
 		// opt-in retry can never become a tight loop hammering the provider,
 		// even if Backoff was left at zero.
-		backoff := policy.Backoff
-		if backoff <= 0 {
-			backoff = 200 * time.Millisecond
-		}
-		if shift := attempt - 1; shift > 0 {
-			backoff <<= shift
-		}
-		if backoff > 30*time.Second {
-			backoff = 30 * time.Second
-		}
+		backoff := retryBackoff(err, attempt, policy.Backoff)
 		t := time.NewTimer(backoff)
 		select {
 		case <-ctx.Done():
@@ -136,6 +133,30 @@ func GenerateWithRetry(ctx context.Context, m Model, req *Request, policy Genera
 	return nil, &RetryError{Attempts: policy.MaxAttempts, Kind: ClassifyError(last), Err: last}
 }
 
+func retryBackoff(err error, attempt int, base time.Duration) time.Duration {
+	backoff := base
+	if backoff <= 0 {
+		backoff = 200 * time.Millisecond
+	}
+	if shift := attempt - 1; shift > 0 {
+		backoff <<= shift
+	}
+	if backoff > 30*time.Second {
+		backoff = 30 * time.Second
+	}
+
+	var retryAfter RetryAfterCoder
+	if errors.As(err, &retryAfter) {
+		if delay := retryAfter.RetryAfter(); delay > backoff {
+			backoff = delay
+		}
+	}
+	if backoff > 30*time.Second {
+		return 30 * time.Second
+	}
+	return backoff
+}
+
 // ClassifyError maps provider and context failures to stable operational kinds.
 func ClassifyError(err error) ErrorKind {
 	if err == nil {

ai/retry_test.go

@@ -139,6 +139,14 @@ type statusErr int
 func (e statusErr) Error() string   { return "provider status" }
 func (e statusErr) StatusCode() int { return int(e) }
 
+type retryAfterErr struct {
+	delay time.Duration
+}
+
+func (e retryAfterErr) Error() string             { return "rate limit exceeded" }
+func (e retryAfterErr) StatusCode() int           { return 429 }
+func (e retryAfterErr) RetryAfter() time.Duration { return e.delay }
+
 func TestClassifyErrorDistinguishesOperationalOutcomes(t *testing.T) {
 	tests := []struct {
 		name string
@@ -181,3 +189,35 @@ func TestGenerateWithRetryExposesRetryErrorKind(t *testing.T) {
 		t.Fatalf("retry error does not unwrap provider status: %v", err)
 	}
 }
+
+func TestGenerateWithRetryHonorsRetryAfterWhenLongerThanBackoff(t *testing.T) {
+	attempts := 0
+	model := retryModel{generate: func(context.Context, *Request, ...GenerateOption) (*Response, error) {
+		attempts++
+		if attempts == 1 {
+			return nil, retryAfterErr{delay: 25 * time.Millisecond}
+		}
+		return &Response{Reply: "ok"}, nil
+	}}
+
+	start := time.Now()
+	resp, err := GenerateWithRetry(context.Background(), model, &Request{Prompt: "hi"}, GeneratePolicy{
+		MaxAttempts: 2,
+		Backoff:     time.Millisecond,
+	})
+	if err != nil {
+		t.Fatalf("GenerateWithRetry returned error: %v", err)
+	}
+	if resp.Reply != "ok" {
+		t.Fatalf("reply = %q, want ok", resp.Reply)
+	}
+	if elapsed := time.Since(start); elapsed < 20*time.Millisecond {
+		t.Fatalf("retry delay = %s, want RetryAfter delay to dominate base backoff", elapsed)
+	}
+}
+
+func TestGenerateWithRetryCapsRetryAfter(t *testing.T) {
+	if got := retryBackoff(retryAfterErr{delay: time.Minute}, 1, time.Millisecond); got != 30*time.Second {
+		t.Fatalf("retryBackoff() = %s, want 30s cap", got)
+	}
+}

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

@@ -42,14 +42,28 @@ func Deploy(c *cli.Context) error {
 		return showDeployHelp()
 	}
 
+	target, remotePath := resolveDeployTarget(c, target, cfg)
+
+	return deploySSH(c, target, cfg, remotePath)
+}
+
+func resolveDeployTarget(c *cli.Context, target string, cfg *config.Config) (string, string) {
+	remotePath := c.String("path")
+	if remotePath == "" {
+		remotePath = defaultRemotePath
+	}
+
 	// Check if target is a named target from config
 	if cfg != nil {
 		if dt, ok := cfg.Deploy[target]; ok {
 			target = dt.SSH
+			if dt.Path != "" && !c.IsSet("path") {
+				remotePath = dt.Path
+			}
 		}
 	}
 
-	return deploySSH(c, target, cfg)
+	return target, remotePath
 }
 
 func showDeployHelp() error {
@@ -82,7 +96,7 @@ func showDeployTargets(cfg *config.Config) error {
 	return fmt.Errorf("%s", sb.String())
 }
 
-func deploySSH(c *cli.Context, target string, cfg *config.Config) error {
+func deploySSH(c *cli.Context, target string, cfg *config.Config, remotePath string) error {
 	dir := c.Args().Get(1)
 	if dir == "" {
 		dir = "."
@@ -98,7 +112,6 @@ func deploySSH(c *cli.Context, target string, cfg *config.Config) error {
 		cfg, _ = config.Load(absDir)
 	}
 
-	remotePath := c.String("path")
 	if remotePath == "" {
 		remotePath = defaultRemotePath
 	}

cmd/micro/cli/deploy/deploy_test.go

@@ -0,0 +1,120 @@
+package deploy
+
+import (
+	"flag"
+	"os"
+	"strings"
+	"testing"
+
+	"github.com/urfave/cli/v2"
+	"go-micro.dev/v6/cmd/micro/run/config"
+)
+
+func newDeployTestContext(t *testing.T, args ...string) *cli.Context {
+	t.Helper()
+	set := flag.NewFlagSet("deploy", flag.ContinueOnError)
+	set.String("path", defaultRemotePath, "")
+	set.String("ssh", "", "")
+	set.String("service", "", "")
+	set.Bool("build", false, "")
+	if err := set.Parse(args); err != nil {
+		t.Fatalf("parse flags: %v", err)
+	}
+	return cli.NewContext(cli.NewApp(), set, nil)
+}
+
+func TestDeployNoTargetExplainsInitAndDeployHandoff(t *testing.T) {
+	err := showDeployHelp()
+	if err == nil {
+		t.Fatal("expected missing target guidance")
+	}
+	msg := err.Error()
+	for _, want := range []string{
+		"no deployment target specified",
+		"sudo micro init --server",
+		"micro deploy user@your-server",
+		"deploy prod",
+	} {
+		if !strings.Contains(msg, want) {
+			t.Fatalf("missing %q in guidance:\n%s", want, msg)
+		}
+	}
+}
+
+func TestDeployListsConfiguredTargetsWhenNoTargetProvided(t *testing.T) {
+	err := showDeployTargets(&config.Config{Deploy: map[string]*config.DeployTarget{
+		"prod":    {Name: "prod", SSH: "deploy@prod.example.com"},
+		"staging": {Name: "staging", SSH: "deploy@staging.example.com"},
+	}})
+	if err == nil {
+		t.Fatal("expected configured target guidance")
+	}
+	msg := err.Error()
+	for _, want := range []string{
+		"Available deploy targets:",
+		"prod -> deploy@prod.example.com",
+		"staging -> deploy@staging.example.com",
+		"micro deploy <target>",
+	} {
+		if !strings.Contains(msg, want) {
+			t.Fatalf("missing %q in configured target guidance:\n%s", want, msg)
+		}
+	}
+}
+
+func TestResolveDeployTargetUsesConfigTargetAndPath(t *testing.T) {
+	ctx := newDeployTestContext(t, "prod")
+	cfg := &config.Config{Deploy: map[string]*config.DeployTarget{
+		"prod": {Name: "prod", SSH: "deploy@prod.example.com", Path: "/srv/micro"},
+	}}
+
+	target, remotePath := resolveDeployTarget(ctx, ctx.Args().First(), cfg)
+	if target != "deploy@prod.example.com" {
+		t.Fatalf("target = %q, want configured SSH", target)
+	}
+	if remotePath != "/srv/micro" {
+		t.Fatalf("remotePath = %q, want configured path", remotePath)
+	}
+}
+
+func TestResolveDeployTargetAllowsCLIPathOverride(t *testing.T) {
+	ctx := newDeployTestContext(t, "--path", "/tmp/micro", "prod")
+	cfg := &config.Config{Deploy: map[string]*config.DeployTarget{
+		"prod": {Name: "prod", SSH: "deploy@prod.example.com", Path: "/srv/micro"},
+	}}
+
+	target, remotePath := resolveDeployTarget(ctx, ctx.Args().First(), cfg)
+	if target != "deploy@prod.example.com" {
+		t.Fatalf("target = %q, want configured SSH", target)
+	}
+	if remotePath != "/tmp/micro" {
+		t.Fatalf("remotePath = %q, want CLI override", remotePath)
+	}
+}
+
+func TestDeployConfigParserSupportsDeployTargets(t *testing.T) {
+	dir := t.TempDir()
+	path := dir + "/micro.mu"
+	content := `service api
+    path ./api
+
+deploy prod
+    ssh deploy@prod.example.com
+    path /srv/micro
+`
+	if err := os.WriteFile(path, []byte(content), 0644); err != nil {
+		t.Fatalf("write config: %v", err)
+	}
+
+	cfg, err := config.ParseMu(path)
+	if err != nil {
+		t.Fatalf("parse config: %v", err)
+	}
+	prod := cfg.Deploy["prod"]
+	if prod == nil {
+		t.Fatal("missing prod deploy target")
+	}
+	if prod.SSH != "deploy@prod.example.com" || prod.Path != "/srv/micro" {
+		t.Fatalf("deploy target = %#v", prod)
+	}
+}

cmd/micro/cli/new/contract_test.go

@@ -1,20 +1,23 @@
 package new
 
 import (
+	"errors"
 	"flag"
 	"os"
 	"os/exec"
 	"path/filepath"
 	"strings"
 	"testing"
+	"time"
 
 	"github.com/urfave/cli/v2"
 )
 
 // TestZeroToOneContract locks the documented getting-started path:
 // `micro new helloworld` must produce an ordinary Go service that the Go
-// toolchain can build. The generated module is pointed back at this checkout
-// so the contract stays local and deterministic in CI.
+// toolchain can build, run long enough to start, and call through its generated
+// handler. The generated module is pointed back at this checkout so the
+// contract stays local and deterministic in CI.
 //
 // It shells out to `micro new` (which runs `go mod tidy`) and `go build`, so
 // it needs the Go toolchain and module access; it is skipped under `-short`.
@@ -29,6 +32,8 @@ func TestZeroToOneContract(t *testing.T) {
 
 	generated.replaceModule(t)
 	generated.build(t)
+	generated.run(t)
+	generated.call(t, "Alice", "Hello Alice")
 }
 
 // TestZeroToOneNoMCPContract keeps the MCP opt-out path honest. Some services
@@ -48,6 +53,8 @@ func TestZeroToOneNoMCPContract(t *testing.T) {
 
 	generated.replaceModule(t)
 	generated.build(t)
+	generated.run(t)
+	generated.call(t, "Bob", "Hello Bob")
 }
 
 type generatedService struct {
@@ -121,3 +128,74 @@ func (g generatedService) build(t *testing.T) {
 		t.Fatalf("generated service go build ./... failed: %v\n%s", err, out)
 	}
 }
+
+func (g generatedService) run(t *testing.T) {
+	t.Helper()
+
+	bin := filepath.Join(g.dir, "service-contract")
+	build := exec.Command("go", "build", "-o", bin, ".")
+	build.Dir = g.dir
+	if out, err := build.CombinedOutput(); err != nil {
+		t.Fatalf("generated service go build -o service-contract . failed: %v\n%s", err, out)
+	}
+
+	cmd := exec.Command(bin)
+	cmd.Dir = g.dir
+	var out strings.Builder
+	cmd.Stdout = &out
+	cmd.Stderr = &out
+
+	if err := cmd.Start(); err != nil {
+		t.Fatalf("generated service failed to start: %v\n%s", err, out.String())
+	}
+	done := make(chan error, 1)
+	go func() { done <- cmd.Wait() }()
+
+	select {
+	case err := <-done:
+		t.Fatalf("generated service exited early: %v\n%s", err, out.String())
+	case <-time.After(2 * time.Second):
+	}
+
+	if err := cmd.Process.Kill(); err != nil && !errors.Is(err, os.ErrProcessDone) {
+		t.Fatalf("failed to stop generated service: %v\n%s", err, out.String())
+	}
+	select {
+	case <-done:
+	case <-time.After(2 * time.Second):
+		t.Fatalf("generated service did not stop after kill\n%s", out.String())
+	}
+}
+
+func (g generatedService) call(t *testing.T, name, want string) {
+	t.Helper()
+
+	testPath := filepath.Join(g.dir, "handler", "contract_test.go")
+	testSrc := `package handler
+
+import (
+	"context"
+	"testing"
+)
+
+func TestGeneratedCallContract(t *testing.T) {
+	rsp := new(Response)
+	if err := New().Call(context.Background(), &Request{Name: "` + name + `"}, rsp); err != nil {
+		t.Fatal(err)
+	}
+	if rsp.Msg != "` + want + `" {
+		t.Fatalf("Call response = %q, want %q", rsp.Msg, "` + want + `")
+	}
+}
+`
+	if err := os.WriteFile(testPath, []byte(testSrc), 0644); err != nil {
+		t.Fatal(err)
+	}
+
+	cmd := exec.Command("go", "test", "./handler", "-run", "TestGeneratedCallContract", "-count=1")
+	cmd.Dir = g.dir
+	out, err := cmd.CombinedOutput()
+	if err != nil {
+		t.Fatalf("generated service call contract failed: %v\n%s", err, out)
+	}
+}

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/zero_to_hero_cli_test.go

@@ -0,0 +1,33 @@
+package main
+
+import (
+	"testing"
+
+	microcmd "go-micro.dev/v6/cmd"
+)
+
+func TestZeroToHeroCLIBoundaries(t *testing.T) {
+	commands := map[string]bool{}
+	subcommands := map[string]map[string]bool{}
+	for _, command := range microcmd.DefaultCmd.App().Commands {
+		commands[command.Name] = true
+		for _, subcommand := range command.Subcommands {
+			if subcommands[command.Name] == nil {
+				subcommands[command.Name] = map[string]bool{}
+			}
+			subcommands[command.Name][subcommand.Name] = true
+		}
+	}
+
+	for _, want := range []string{"run", "chat", "flow", "inspect"} {
+		if !commands[want] {
+			t.Fatalf("missing %q command", want)
+		}
+	}
+	if !subcommands["flow"]["runs"] {
+		t.Fatal("missing inspect boundary: flow runs")
+	}
+	if !subcommands["inspect"]["agent"] || !subcommands["inspect"]["flow"] {
+		t.Fatal("missing inspect boundary: inspect agent/flow")
+	}
+}

events/memory.go

@@ -19,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/agent-human-input/README.md

@@ -0,0 +1,36 @@
+# Agent Human Input Pause/Resume
+
+Agents can pause a durable run when the model needs a human decision before it
+can continue. This keeps the services → agents → workflows lifecycle in one
+runtime: services expose tools, the agent decides it needs operator input, and
+the same checkpointed run resumes once that input arrives.
+
+## Pattern
+
+```go
+cp := flow.StoreCheckpoint(nil, "deploy-agent")
+ag := agent.New(
+    agent.Name("deploy-agent"),
+    agent.WithCheckpoint(cp),
+)
+
+resp, err := ag.Ask(ctx, "Deploy the service")
+if err != nil {
+    // If the model called the built-in request_input tool, the run is saved as
+    // paused/input-required instead of losing state or completing early.
+    pending, _ := agent.Pending(ctx, ag)
+    runID := pending[0].ID
+
+    // Later, after an operator supplies the missing answer, the same run ID
+    // continues with the original prompt, human input, memory, and completed
+    // tool history intact.
+    resp, err = agent.ResumeInput(ctx, ag, runID, "Deploy to us-east-1")
+}
+_ = resp
+```
+
+The model sees a built-in `request_input` tool with a `prompt` argument. When it
+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.

examples/support/main.go

@@ -212,37 +212,39 @@ func waitFor(reg registry.Registry, names ...string) {
 	}
 }
 
-func main() {
-	provider := flag.String("provider", "mock", "LLM provider: mock (default), anthropic, openai, ...")
-	flag.Parse()
-
+func runSupport(provider string) error {
 	apiKey := ""
-	if *provider == "mock" {
+	if provider == "mock" {
 		ai.Register("mock", newMock)
-	} else if apiKey = providerKey(*provider); apiKey == "" {
-		fmt.Printf("no API key for provider %q — set MICRO_AI_API_KEY or the provider's key env\n", *provider)
-		os.Exit(1)
+	} else if apiKey = providerKey(provider); apiKey == "" {
+		return fmt.Errorf("no API key for provider %q — set MICRO_AI_API_KEY or the provider's key env", provider)
 	}
 
-	fmt.Printf("\n\033[1mSupport desk (provider: %s)\033[0m\n\n", *provider)
+	fmt.Printf("\n\033[1mSupport desk (provider: %s)\033[0m\n\n", provider)
 
 	// Shared in-memory infrastructure so the demo runs in one process.
 	reg := registry.NewMemoryRegistry()
 	br := broker.NewMemoryBroker()
 	if err := br.Connect(); err != nil {
-		fmt.Println("broker connect:", err)
-		os.Exit(1)
+		return fmt.Errorf("broker connect: %w", err)
 	}
 	cl := client.NewClient(client.Registry(reg), client.Selector(selector.NewSelector(selector.Registry(reg))))
 
 	// Services.
 	tickets := new(TicketService)
 	notify := new(NotifyService)
+	var services []service.Service
 	for name, h := range map[string]any{"customers": new(CustomerService), "tickets": tickets, "notify": notify} {
-		svc := service.New(service.Name(name), service.Registry(reg), service.Client(cl))
+		svc := service.New(service.Name(name), service.Address("127.0.0.1:0"), service.Registry(reg), service.Client(cl), service.HandleSignal(false))
 		_ = svc.Handle(h)
+		services = append(services, svc)
 		go svc.Run()
 	}
+	defer func() {
+		for _, svc := range services {
+			_ = svc.Server().Stop()
+		}
+	}()
 
 	// The support agent manages the three services. The approval gate is
 	// the human-in-the-loop: it can read and triage freely, but emailing a
@@ -250,10 +252,11 @@ func main() {
 	// it for a person or a policy; here we approve and log.
 	support := agent.New(
 		agent.Name("support"),
+		agent.Address("127.0.0.1:0"),
 		agent.Services("customers", "tickets", "notify"),
 		agent.Prompt("You are a support agent. For each ticket, look up the customer, set an "+
 			"appropriate priority, and reply to them. Escalate billing issues."),
-		agent.Provider(*provider), agent.APIKey(apiKey),
+		agent.Provider(provider), agent.APIKey(apiKey),
 		agent.ApproveTool(func(tool string, input map[string]any) (bool, string) {
 			if strings.Contains(tool, "Send") {
 				fmt.Printf("  \033[33m▣ approval gate\033[0m %s(%v) — approved\n", tool, input["to"])
@@ -275,8 +278,7 @@ func main() {
 		flow.Prompt("A new support ticket arrived: {{.Data}}. Handle it."),
 	)
 	if err := intake.Register(reg, br, cl); err != nil {
-		fmt.Println("flow register:", err)
-		os.Exit(1)
+		return fmt.Errorf("flow register: %w", err)
 	}
 	defer intake.Stop()
 
@@ -287,8 +289,7 @@ func main() {
 	fmt.Println("\033[1m> event:\033[0m events.ticket.created", string(body))
 	fmt.Println()
 	if err := br.Publish("events.ticket.created", &broker.Message{Body: body}); err != nil {
-		fmt.Println("publish:", err)
-		os.Exit(1)
+		return fmt.Errorf("publish: %w", err)
 	}
 
 	// Wait for the agent to act.
@@ -305,7 +306,18 @@ func main() {
 	}
 	if notify.sent >= 1 {
 		fmt.Println("\n\033[32m✓ ticket triaged and the customer was replied to — triggered by an event\033[0m")
-	} else {
-		fmt.Println("\n\033[31m✗ the agent did not complete the triage\033[0m")
+		return nil
+	}
+	fmt.Println("\n\033[31m✗ the agent did not complete the triage\033[0m")
+	return fmt.Errorf("support agent did not complete triage")
+}
+
+func main() {
+	provider := flag.String("provider", "mock", "LLM provider: mock (default), anthropic, openai, ...")
+	flag.Parse()
+
+	if err := runSupport(*provider); err != nil {
+		fmt.Println(err)
+		os.Exit(1)
 	}
 }

examples/support/main_test.go

@@ -0,0 +1,9 @@
+package main
+
+import "testing"
+
+func TestRunSupportMockSmoke(t *testing.T) {
+	if err := runSupport("mock"); err != nil {
+		t.Fatalf("support example failed: %v", err)
+	}
+}

gateway/a2a/a2a.go

@@ -21,8 +21,8 @@
 // Scope of this version: the JSON-RPC binding — `message/send`
 // (returns a completed Task), `message/stream` (SSE with the completed
 // Task event), `tasks/get`, multi-turn task continuation, push
-// notification delivery, and Agent Card discovery. `input-required` and
-// `tasks/resubscribe` are advertised as unsupported and are follow-ups.
+// notification delivery, input-required handoffs, `tasks/resubscribe`,
+// and Agent Card discovery.
 package a2a
 
 import (
@@ -236,9 +236,10 @@ type PushNotificationConfig struct {
 
 // Task states (JSON-RPC binding wire values).
 const (
-	stateCompleted = "completed"
-	stateFailed    = "failed"
-	stateWorking   = "working"
+	stateCompleted     = "completed"
+	stateFailed        = "failed"
+	stateWorking       = "working"
+	stateInputRequired = "input-required"
 )
 
 // JSON-RPC envelopes.
@@ -427,11 +428,12 @@ type dispatcher struct {
 	mu          sync.Mutex
 	tasks       map[string]*Task
 	pushConfigs map[string]PushNotificationConfig
+	watchers    map[string]map[chan *Task]struct{}
 	order       []string // task ids in insertion order, for bounded eviction
 }
 
 func newDispatcher() *dispatcher {
-	return &dispatcher{tasks: map[string]*Task{}, pushConfigs: map[string]PushNotificationConfig{}}
+	return &dispatcher{tasks: map[string]*Task{}, pushConfigs: map[string]PushNotificationConfig{}, watchers: map[string]map[chan *Task]struct{}{}}
 }
 
 func (d *dispatcher) serve(w http.ResponseWriter, r *http.Request, invoke Invoke) {
@@ -468,7 +470,7 @@ func (d *dispatcher) serveWithStream(w http.ResponseWriter, r *http.Request, inv
 		// v1 tasks complete synchronously, so they're already terminal.
 		writeRPC(w, req.ID, nil, &rpcError{Code: errNotCancelable, Message: "task is not cancelable"})
 	case "tasks/resubscribe":
-		writeRPC(w, req.ID, nil, &rpcError{Code: errMethodNotFound, Message: "resubscribe is not supported"})
+		d.resubscribe(requestContext(r.Context()), w, req)
 	default:
 		writeRPC(w, req.ID, nil, &rpcError{Code: errMethodNotFound, Message: "method not found: " + req.Method})
 	}
@@ -557,6 +559,7 @@ func (d *dispatcher) streamChunks(ctx context.Context, w http.ResponseWriter, re
 		}
 		reply.WriteString(chunk.Reply)
 		task := taskFromReplyWithIDs(p.Message, reply.String(), stateWorking, taskID, contextID)
+		d.store(task)
 		_ = enc.Encode(rpcResponse{JSONRPC: "2.0", ID: req.ID, Result: task})
 		flush()
 	}
@@ -577,6 +580,10 @@ func (d *dispatcher) run(ctx context.Context, params json.RawMessage, invoke Inv
 	if err != nil {
 		reply = "error: " + err.Error()
 		state = stateFailed
+		if isInputRequiredError(err) {
+			reply = err.Error()
+			state = stateInputRequired
+		}
 	}
 	task := d.taskFromReply(p.Message, reply, state)
 	d.store(task)
@@ -587,6 +594,49 @@ type getParams struct {
 	ID string `json:"id"`
 }
 
+func (d *dispatcher) resubscribe(ctx context.Context, w http.ResponseWriter, req rpcRequest) {
+	var p getParams
+	if err := json.Unmarshal(req.Params, &p); err != nil || p.ID == "" {
+		writeRPC(w, req.ID, nil, &rpcError{Code: errInvalidParams, Message: "invalid params"})
+		return
+	}
+	ch, task, unsubscribe := d.subscribe(p.ID)
+	if task == nil {
+		writeRPC(w, req.ID, nil, &rpcError{Code: errTaskNotFound, Message: "task not found"})
+		return
+	}
+	defer unsubscribe()
+
+	w.Header().Set("Content-Type", "text/event-stream")
+	w.Header().Set("Cache-Control", "no-cache")
+	w.Header().Set("Connection", "keep-alive")
+	w.WriteHeader(http.StatusOK)
+	enc := json.NewEncoder(sseWriter{w: w})
+	flush := func() {
+		if f, ok := w.(http.Flusher); ok {
+			f.Flush()
+		}
+	}
+	writeEvent := func(t *Task) bool {
+		_ = enc.Encode(rpcResponse{JSONRPC: "2.0", ID: req.ID, Result: t})
+		flush()
+		return isTerminal(t.Status.State)
+	}
+	if writeEvent(task) {
+		return
+	}
+	for {
+		select {
+		case <-ctx.Done():
+			return
+		case next := <-ch:
+			if writeEvent(next) {
+				return
+			}
+		}
+	}
+}
+
 func (d *dispatcher) get(w http.ResponseWriter, req rpcRequest) {
 	var p getParams
 	if err := json.Unmarshal(req.Params, &p); err != nil || p.ID == "" {
@@ -672,18 +722,59 @@ func (g *Gateway) callAgent(ctx context.Context, name, message string) (string,
 
 func (d *dispatcher) store(t *Task) {
 	d.mu.Lock()
+	_, exists := d.tasks[t.ID]
 	d.tasks[t.ID] = t
-	d.order = append(d.order, t.ID)
+	if !exists {
+		d.order = append(d.order, t.ID)
+	}
 	for len(d.order) > maxTasks {
 		oldest := d.order[0]
 		d.order = d.order[1:]
 		delete(d.tasks, oldest)
 		delete(d.pushConfigs, oldest)
 	}
+	for ch := range d.watchers[t.ID] {
+		select {
+		case ch <- t:
+		default:
+		}
+	}
 	d.mu.Unlock()
 	go d.deliverPush(t.ID, t)
 }
 
+func (d *dispatcher) subscribe(taskID string) (chan *Task, *Task, func()) {
+	d.mu.Lock()
+	defer d.mu.Unlock()
+	task := d.tasks[taskID]
+	if task == nil {
+		return nil, nil, func() {}
+	}
+	ch := make(chan *Task, 8)
+	if d.watchers[taskID] == nil {
+		d.watchers[taskID] = map[chan *Task]struct{}{}
+	}
+	d.watchers[taskID][ch] = struct{}{}
+	return ch, task, func() {
+		d.mu.Lock()
+		delete(d.watchers[taskID], ch)
+		if len(d.watchers[taskID]) == 0 {
+			delete(d.watchers, taskID)
+		}
+		close(ch)
+		d.mu.Unlock()
+	}
+}
+
+func isTerminal(state string) bool {
+	return state == stateCompleted || state == stateFailed || state == stateInputRequired
+}
+
+func isInputRequiredError(err error) bool {
+	msg := strings.ToLower(err.Error())
+	return strings.Contains(msg, "input-required") || strings.Contains(msg, "input required") || strings.Contains(msg, "paused for approval")
+}
+
 func (d *dispatcher) taskFromReply(input Message, reply, state string) *Task {
 	contextID := input.ContextID
 	taskID := input.TaskID

gateway/a2a/a2a_test.go

@@ -4,6 +4,7 @@ import (
 	"bytes"
 	"context"
 	"encoding/json"
+	"errors"
 	"fmt"
 	"io"
 	"net/http"
@@ -359,6 +360,193 @@ func TestMessageStreamChunksStoreFinalTask(t *testing.T) {
 	}
 }
 
+type contextStream struct {
+	ctx    context.Context
+	closed chan struct{}
+}
+
+func (s *contextStream) Recv() (*ai.Response, error) {
+	<-s.ctx.Done()
+	return nil, s.ctx.Err()
+}
+
+func (s *contextStream) Close() error {
+	close(s.closed)
+	return nil
+}
+
+func TestMessageStreamChunksPropagatesCancellationAndClosesStream(t *testing.T) {
+	d := newDispatcher()
+	ctx, cancel := context.WithCancel(context.Background())
+	closed := make(chan struct{})
+	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)).WithContext(ctx)
+	rr := httptest.NewRecorder()
+	cancel()
+
+	d.serveWithStream(rr, req, nil, func(ctx context.Context, text string) (ai.Stream, error) {
+		if text != "ping" {
+			t.Fatalf("stream text = %q, want ping", text)
+		}
+		return &contextStream{ctx: ctx, closed: closed}, nil
+	})
+
+	select {
+	case <-closed:
+	case <-time.After(time.Second):
+		t.Fatal("stream was not closed")
+	}
+
+	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())
+	}
+	event := events[0]
+	if event.Error == nil || event.Error.Code != errInternal || event.Error.Message != context.Canceled.Error() {
+		t.Fatalf("error = %+v, want context cancellation", event.Error)
+	}
+	if event.Result.Status.State != stateFailed || textOf(event.Result.Artifacts[0].Parts) != "error: context canceled" {
+		t.Fatalf("failed task = %+v, want context cancellation artifact", event.Result)
+	}
+
+	got := rpcTaskFromDispatcher(t, d, event.Result.ID)
+	if got.Status.State != stateFailed || textOf(got.Artifacts[0].Parts) != "error: context canceled" {
+		t.Fatalf("stored task = %+v, want failed cancellation", got)
+	}
+}
+
+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)}}
+	d.store(initial)
+
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	req := httptest.NewRequest(http.MethodPost, "/", bytes.NewBufferString(`{"jsonrpc":"2.0","id":1,"method":"tasks/resubscribe","params":{"id":"task-1"}}`)).WithContext(ctx)
+	rw := newFlushRecorder()
+	done := make(chan struct{})
+	go func() {
+		d.serve(rw, req, nil)
+		close(done)
+	}()
+
+	first := rw.next(t)
+	if first.Result.ID != initial.ID || first.Result.Status.State != stateWorking {
+		t.Fatalf("first resubscribe event = %+v, want current working task", first.Result)
+	}
+
+	final := &Task{ID: "task-1", ContextID: "ctx-1", Kind: "task", Status: TaskStatus{State: stateCompleted, Timestamp: time.Now().UTC().Format(time.RFC3339)}, Artifacts: []Artifact{textArtifact("done")}}
+	d.store(final)
+	second := rw.next(t)
+	if second.Result.ID != final.ID || second.Result.Status.State != stateCompleted || textOf(second.Result.Artifacts[0].Parts) != "done" {
+		t.Fatalf("second resubscribe event = %+v, want completed update", second.Result)
+	}
+	select {
+	case <-done:
+	case <-time.After(time.Second):
+		t.Fatal("resubscribe did not return after terminal update")
+	}
+}
+
+func TestInputRequiredErrorCreatesContinuableTask(t *testing.T) {
+	d := newDispatcher()
+	first := rpcTaskFromBody(t, d, `{
+		"jsonrpc":"2.0","id":1,"method":"message/send",
+		"params":{"message":{"role":"user","kind":"message","messageId":"m1",
+			"parts":[{"kind":"text","text":"start approval"}]}}}`, func(_ context.Context, text string) (string, error) {
+		return "", errors.New("agent run run-1 paused for approval: waiting for operator")
+	})
+	if first.Status.State != stateInputRequired {
+		t.Fatalf("state = %q, want input-required", first.Status.State)
+	}
+	if textOf(first.Artifacts[0].Parts) != "agent run run-1 paused for approval: waiting for operator" {
+		t.Fatalf("artifact = %+v, want handoff message", first.Artifacts)
+	}
+
+	body := fmt.Sprintf(`{
+		"jsonrpc":"2.0","id":2,"method":"message/send",
+		"params":{"message":{"role":"user","kind":"message","messageId":"m2","taskId":"%s","contextId":"%s",
+			"parts":[{"kind":"text","text":"approved"}]}}}`, first.ID, first.ContextID)
+	continued := rpcTaskFromBody(t, d, body, func(_ context.Context, text string) (string, error) {
+		return "continued after " + text, nil
+	})
+	if continued.ID != first.ID || continued.ContextID != first.ContextID {
+		t.Fatalf("continued identity = %s/%s, want %s/%s", continued.ID, continued.ContextID, first.ID, first.ContextID)
+	}
+	if continued.Status.State != stateCompleted || len(continued.History) != 4 {
+		t.Fatalf("continued task = %+v, want completed task with prior input-required history", continued)
+	}
+	if textOf(continued.History[1].Parts) != "agent run run-1 paused for approval: waiting for operator" || textOf(continued.History[3].Parts) != "continued after approved" {
+		t.Fatalf("continued history = %+v", continued.History)
+	}
+}
+
+type flushRecorder struct {
+	*httptest.ResponseRecorder
+	ch chan string
+}
+
+func newFlushRecorder() *flushRecorder {
+	return &flushRecorder{ResponseRecorder: httptest.NewRecorder(), ch: make(chan string, 16)}
+}
+
+func (r *flushRecorder) Flush() {
+	body := r.Body.String()
+	r.Body.Reset()
+	for _, line := range strings.Split(strings.TrimSpace(body), "\n") {
+		line = strings.TrimSpace(line)
+		if line != "" {
+			r.ch <- line
+		}
+	}
+}
+
+func (r *flushRecorder) next(t *testing.T) struct {
+	Result Task      `json:"result"`
+	Error  *rpcError `json:"error"`
+} {
+	t.Helper()
+	select {
+	case line := <-r.ch:
+		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)
+		}
+		if event.Error != nil {
+			t.Fatalf("event error: %+v", event.Error)
+		}
+		return event
+	case <-time.After(time.Second):
+		t.Fatal("timed out waiting for SSE event")
+	}
+	return struct {
+		Result Task      `json:"result"`
+		Error  *rpcError `json:"error"`
+	}{}
+}
+
 func rpcTaskFromDispatcher(t *testing.T, d *dispatcher, id string) Task {
 	t.Helper()
 	body := fmt.Sprintf(`{"jsonrpc":"2.0","id":2,"method":"tasks/get","params":{"id":"%s"}}`, id)
@@ -408,9 +596,9 @@ func TestUnknownMethod(t *testing.T) {
 	var resp struct {
 		Error *rpcError `json:"error"`
 	}
-	rpc(t, ts.URL+"/agents/echo", `{"jsonrpc":"2.0","id":1,"method":"tasks/resubscribe","params":{}}`, &resp)
+	rpc(t, ts.URL+"/agents/echo", `{"jsonrpc":"2.0","id":1,"method":"unknown","params":{}}`, &resp)
 	if resp.Error == nil || resp.Error.Code != errMethodNotFound {
-		t.Errorf("expected method-not-found for resubscribe, got %+v", resp.Error)
+		t.Errorf("expected method-not-found, got %+v", resp.Error)
 	}
 }
 

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,14 +19,11 @@ 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.
 
-## Now (ranked)
+## Work queue (ranked)
 
-1. **CI-verify the 0-to-1 getting-started path** ([#3234](https://github.com/micro/go-micro/issues/3234)) — add a deterministic no-key check for scaffold → run/build → call so the README/blog promise that building a service stays effortless cannot regress while the agent stack deepens.
-2. **CI-verify the 0-to-hero agent workflow** ([#3241](https://github.com/micro/go-micro/issues/3241)) — after the basic service contract is guarded, make the full services → agents → workflows story executable in CI with a maintained no-secret reference scenario for run → chat → inspect boundaries.
-
-## Later (ranked)
-
-3. **Add A2A resubscribe and input-required handoff support** ([#3235](https://github.com/micro/go-micro/issues/3235)) — after push notifications and multi-turn continuation shipped, finish the remaining long-running A2A interoperability gap: reconnecting to live task streams and carrying human-input-required handoffs through the gateway.
+1. **Trace agent runs with `RunInfo` OpenTelemetry spans** ([#3362](https://github.com/micro/go-micro/issues/3362)) — the highest-value Next-phase operability gap now that the getting-started contract shipped in CI: scheduled and looping agents need run/session IDs, model attempts, tool calls, delegation, errors, cancellation, and terminal status visible as traces without leaking sensitive payloads by default. This keeps the agent harness operable in the same runtime developers already deploy.
+2. **Add human-in-the-loop pause and resume for agent workflows** ([#3329](https://github.com/micro/go-micro/issues/3329)) — after memory compaction, tool-aware streaming, durable `Ask`/`StreamAsk` checkpoints, and the getting-started harness shipped, close the next workflow-operability gap: durable pending-input states that let humans approve or provide context and then resume the same service/agent/workflow runtime.
+3. **gateway/a2a: multiple typed skills per agent card** ([#3342](https://github.com/micro/go-micro/issues/3342)) — the A2A gateway now has the full task lifecycle (send/stream/get/cancel/resubscribe, push config, multi-turn, input-required); the remaining interop gap is skill granularity. `Card` only ever advertises one synthetic "chat" skill with services flattened into tags. Let an agent advertise N typed skills and route per skill, so domain-routing agents expose their real capabilities over A2A through the gateway instead of a hand-rolled handler.
 
 _Seeded by Claude Code from the roadmap + open issues; thereafter maintained by the
 architecture-review pass._

internal/harness/provider-conformance/README.md

@@ -0,0 +1,84 @@
+# Provider conformance
+
+This harness keeps the services → agents → workflows lifecycle honest across the
+supported AI providers. It runs the same end-to-end scenarios against each
+configured provider and treats missing provider keys as an explicit skip, so the
+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`:
+
+- `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.
+
+The command also emits the registered provider capability matrix so the run shows
+which providers advertise model, image, video, and streaming support.
+
+## Local usage
+
+Run the deterministic path with no secrets:
+
+```sh
+go run ./internal/harness/provider-conformance -providers mock
+```
+
+Run every live provider that has a key in the environment:
+
+```sh
+go run ./internal/harness/provider-conformance \
+  -summary-json provider-conformance-summary.json \
+  -summary-markdown provider-conformance-summary.md \
+  -capabilities-markdown provider-capabilities.md
+```
+
+Provider keys are read from `MICRO_AI_API_KEY` or the provider-specific variable:
+
+| Provider | Secret / environment variable |
+| --- | --- |
+| 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` |
+
+Use `-require-configured` when you want a selected provider without a key to fail
+instead of skip:
+
+```sh
+go run ./internal/harness/provider-conformance \
+  -providers anthropic,openai \
+  -require-configured
+```
+
+## 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
+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.
+
+The job also appends the Markdown summary and capability matrix to the GitHub
+Actions step summary, making configured, skipped, and failed provider coverage
+visible without downloading artifacts.
+
+## Adding a provider
+
+To bring a new provider into scheduled conformance:
+
+1. register its `ai` provider implementation and capability metadata,
+2. add the provider name and key variable to `providerEnv` in `main.go`,
+3. import the provider package in `main.go`,
+4. pass the matching repository secret through `.github/workflows/harness.yml`,
+   and
+5. run `go run ./internal/harness/provider-conformance -providers <name> \
+   -require-configured` with a live key before opening the change.

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

@@ -0,0 +1,32 @@
+# 0→hero CI harness
+
+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:
+
+1. **Run** — `micro run` remains available as the local development entry point.
+2. **Chat** — `micro chat` remains available as the interactive agent entry point.
+3. **Inspect** — `micro inspect agent <name>` and `micro inspect flow <name>`
+   remain available as the local run-history inspection step, with `micro flow
+   runs` preserving durable workflow history inspection.
+
+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.
+
+## 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 the documented getting-started path before
+the 0→hero scenario, so the public scaffold → run/chat → inspect 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

@@ -0,0 +1,13 @@
+#!/usr/bin/env bash
+set -euo pipefail
+
+ROOT="$(cd "$(dirname "${BASH_SOURCE[0]}")/../../.." && pwd)"
+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
+
+# Deterministic no-secret reference scenarios. These use the real Go Micro
+# runtime and mock only the LLM provider.
+go test ./internal/harness/universe ./internal/harness/plan-delegate -run 'Test.*Harness|TestPlanDelegateEndToEnd|TestPlanDelegateFlowHandoff' -count=1

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

@@ -172,17 +172,11 @@ This is the JSON-RPC binding for task execution:
 - **`tasks/get`** returns a recent task by id.
 - **Multi-turn continuation** keeps task state when a new message includes the previous `taskId`.
 - **`tasks/pushNotificationConfig/set` / `get`** stores and reads a task callback for best-effort update delivery.
+- **`tasks/resubscribe`** reconnects to an existing task stream, immediately emits the current task snapshot, then streams subsequent updates until the task reaches a terminal state.
+- **`input-required`** task state carries human-input handoffs (for example checkpointed approval pauses) in task status, artifacts, and history; continue the task by sending a follow-up message with the same `taskId` and `contextId`.
 - **Agent Card** discovery, generated from the registry.
 
-Both directions work: the gateway exposes your agents, and `a2a.Client` (via `flow.A2A` or `delegate` to a URL) calls external ones.
-
-Not yet supported (advertised as such on the card, so clients negotiate correctly):
-
-- **`tasks/resubscribe`** for reconnecting to a live stream.
-- Multi-turn `input-required` tasks.
-
-These are the natural follow-ups; the task binding is what makes a Go Micro
-agent both reachable from, and able to reach, the A2A ecosystem today.
+Both directions work: the gateway exposes your agents, and `a2a.Client` (via `flow.A2A` or `delegate` to a URL) calls external ones. The task binding is what makes a Go Micro agent both reachable from, and able to reach, the A2A ecosystem today.
 
 ## See also
 

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

@@ -36,13 +36,46 @@ 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
+```
+
+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/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",