Add CLI inspect command for run history

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

.github/workflows/harness.yml

@@ -3,8 +3,8 @@ name: Harness (E2E)
 # Runs the end-to-end harnesses for agents, services, flows, and provider
 # conformance. The default job uses deterministic mock LLMs and needs no
 # secrets. A second job runs the same harnesses against the live provider set and
-# fails if any selected provider secret is missing, so scheduled conformance
-# cannot silently degrade.
+# skips providers whose secrets are absent, so scheduled conformance
+# remains safe in no-key forks while still failing configured providers that drift.
 
 on:
   push:
@@ -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)
@@ -50,6 +50,17 @@ jobs:
         with:
           go-version: stable
           cache: true
+      - name: Agent provider conformance matrix
+        env:
+          GO_MICRO_AGENT_CONFORMANCE_LIVE: "1"
+          ANTHROPIC_API_KEY: ${{ secrets.ANTHROPIC_API_KEY }}
+          OPENAI_API_KEY: ${{ secrets.OPENAI_API_KEY }}
+          GEMINI_API_KEY: ${{ secrets.GEMINI_API_KEY }}
+          GROQ_API_KEY: ${{ secrets.GROQ_API_KEY }}
+          MISTRAL_API_KEY: ${{ secrets.MISTRAL_API_KEY }}
+          TOGETHER_API_KEY: ${{ secrets.TOGETHER_API_KEY }}
+          ATLASCLOUD_API_KEY: ${{ secrets.ATLASCLOUD_API_KEY }}
+        run: go test ./agent -run TestAgentProviderConformanceMatrix -count=1 -v
       - name: Provider conformance against live models
         env:
           ANTHROPIC_API_KEY: ${{ secrets.ANTHROPIC_API_KEY }}
@@ -61,10 +72,23 @@ jobs:
           ATLASCLOUD_API_KEY: ${{ secrets.ATLASCLOUD_API_KEY }}
         run: |
           go run ./internal/harness/provider-conformance \
-            -require-configured \
             -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. v1 is the synchronous JSON-RPC binding (`message/send`, `tasks/get`, card discovery); streaming and push notifications are advertised as 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)

README.md

@@ -237,7 +237,7 @@ Just as a service composes pluggable abstractions (registry, broker, store), an
 ```go
 agent := micro.NewAgent("assistant",
     micro.AgentProvider("anthropic"),                 // model — swap the provider
-    micro.AgentMemory(micro.NewInMemory(50)),         // memory — default is store-backed & durable
+    micro.AgentCompactMemory(40, 12),                 // memory — durable, summarized, recallable
     micro.AgentTool("weather", "Get the weather for a city",
         map[string]any{"city": map[string]any{"type": "string"}},
         func(ctx context.Context, in map[string]any) (string, error) {
@@ -247,7 +247,7 @@ agent := micro.NewAgent("assistant",
 )
 ```
 
-**Memory** is durable and store-backed by default (Postgres, NATS KV, or file), so an agent picks up where it left off after a restart — or supply your own with `AgentMemory`. **Tools** are your services automatically, plus any function you register with `AgentTool`.
+**Memory** is durable and store-backed by default (Postgres, NATS KV, or file), so an agent picks up where it left off after a restart — or supply your own with `AgentMemory`. Long-running agents can opt into `AgentCompactMemory(maxMessages, keepRecent)`: older turns are collapsed into a deterministic summary, recent turns stay verbatim, and relevant archived turns are recalled on future asks without replaying the whole conversation. **Tools** are your services automatically, plus any function you register with `AgentTool`.
 
 ### Paid tools (x402)
 

ROADMAP.md

@@ -15,7 +15,8 @@ The full, current roadmap lives at **[go-micro.dev/docs/roadmap](https://go-micr
 ## Where we are (v6)
 
 Services, agents (`plan`/`delegate`, guardrails, memory, tool middleware), durable
-flows, the MCP and A2A gateways (both directions), x402 paid tools, secure by
+flows, the MCP and A2A gateways (both directions, including A2A streaming,
+push notifications, and multi-turn continuation), x402 paid tools, secure by
 default.
 
 ## Principles
@@ -41,14 +42,14 @@ default.
 ## Next — agentic depth
 
 - **Durable agent loop** — resume a long run via `Checkpoint` (flows already do).
-- **Streaming** — `ai.Stream` + A2A `message/stream`, end to end.
+- **Streaming** — broaden provider-backed `ai.Stream` coverage and keep chat/A2A streaming end to end.
 - **Agent observability** — `RunInfo` → OpenTelemetry spans.
 
 ## Later
 
 - Memory management (summarization, retrieval/RAG); human-in-the-loop pause/resume;
-  x402 live-facilitator conformance and paid remote tools with spend caps; A2A
-  streaming, push notifications, and multi-turn tasks.
+  richer A2A live-stream reconnection (`tasks/resubscribe`) and `input-required`
+  handoffs.
 
 ## Developer experience (ongoing)
 

agent/agent.go

@@ -87,6 +87,16 @@ type agentImpl struct {
 	// Both are surfaced to tool wrappers via ai.RunInfo on the context.
 	runID       string
 	parentRunID string
+
+	// pause records a guardrail approval pause raised during the current
+	// 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.
@@ -150,6 +160,8 @@ func (a *agentImpl) setup() {
 		a.mem = a.opts.Memory
 	case a.ephemeral:
 		a.mem = NewInMemory(a.opts.HistoryLimit)
+	case a.opts.MemoryCompaction.MaxMessages > 0:
+		a.mem = NewCompactingMemory(a.stateStore(), "history", a.opts.MemoryCompaction.MaxMessages, a.opts.MemoryCompaction.KeepRecent)
 	default:
 		a.mem = NewMemory(a.stateStore(), "history", a.opts.HistoryLimit)
 	}
@@ -195,17 +207,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) {
@@ -236,6 +237,7 @@ func (a *agentImpl) askLocked(ctx context.Context, runID, message, parentRunID s
 	a.mem.Add("user", message)
 	a.steps = 0
 	a.calls = map[string]int{}
+	a.pause = nil
 
 	// Correlate this run's tool calls and surface lineage to wrappers.
 	a.runID = runID
@@ -245,17 +247,29 @@ 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
 	}
 	ctx, endRun := a.startRun(ctx, message)
 	defer func() { endRun(err) }()
 
+	messages := a.mem.Messages()
+	if recall, ok := a.mem.(MemoryRecall); ok && a.opts.MemoryRecallLimit > 0 {
+		if recalled := recall.Recall(message, a.opts.MemoryRecallLimit); len(recalled) > 0 {
+			messages = append([]ai.Message{{
+				Role:    "system",
+				Content: "Relevant recalled memory follows; use it as durable prior context without assuming the whole conversation was replayed.",
+			}}, append(recalled, messages...)...)
+		}
+	}
+
 	resp, err := ai.GenerateWithRetry(ctx, a.model, &ai.Request{
 		Prompt:       message,
 		SystemPrompt: a.buildPrompt(),
 		Tools:        toolList,
-		Messages:     a.mem.Messages(),
+		Messages:     messages,
 	}, ai.GeneratePolicy{
 		Timeout:     a.opts.ModelTimeout,
 		MaxAttempts: a.opts.ModelMaxAttempts,
@@ -265,9 +279,28 @@ 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
 	}
+	if a.pause != nil && a.opts.Checkpoint != nil {
+		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
+		if err := a.saveRun(ctx, run); err != nil {
+			return nil, err
+		}
+		return nil, fmt.Errorf("agent run %s paused for approval: %s", run.ID, a.pause.Message)
+	}
 
 	if resp.Reply != "" {
 		a.mem.Add("assistant", resp.Reply)
@@ -296,6 +329,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)
 		}
@@ -200,6 +220,16 @@ func (a *agentImpl) loopWrap(next ai.ToolHandler) ai.ToolHandler {
 }
 
 // approveWrap gates each action before it runs (ApproveTool).
+type approvalPause struct {
+	Tool    string
+	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 {
@@ -208,6 +238,7 @@ func (a *agentImpl) approveWrap(next ai.ToolHandler) ai.ToolHandler {
 				if reason != "" {
 					msg += ": " + reason
 				}
+				a.pause = &approvalPause{Tool: call.Name, Message: msg}
 				return refused(call.ID, ai.RefusedApproval, msg)
 			}
 		}
@@ -226,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,10 +6,15 @@ import (
 	"fmt"
 	"time"
 
+	"go-micro.dev/v6/ai"
 	"go-micro.dev/v6/flow"
 )
 
-const agentAskStep = "ask"
+const (
+	agentAskStep      = "ask"
+	agentApprovalStep = "approval"
+	agentInputStep    = "input-required"
+)
 
 func (a *agentImpl) newCheckpointRun(runID, message, parentRunID string, existing *flow.Run) flow.Run {
 	now := time.Now()
@@ -32,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
 }
@@ -46,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)
@@ -57,6 +74,13 @@ func (a *agentImpl) resume(ctx context.Context, runID string) (*Response, error)
 	if !ok {
 		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
+	}
 	if run.Status == "done" {
 		var resp Response
 		if err := json.Unmarshal(run.State.Data, &resp); err != nil {
@@ -74,6 +98,51 @@ func (a *agentImpl) resume(ctx context.Context, runID string) (*Response, error)
 	return a.askLocked(ctx, run.ID, message, parentID, &run)
 }
 
+// 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)
+}
+
 func (a *agentImpl) pending(ctx context.Context) ([]flow.Run, error) {
 	if a.opts.Checkpoint == nil {
 		return nil, nil
@@ -90,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,58 @@ 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 TestPendingReturnsUnfinishedAgentRuns(t *testing.T) {
 	ctx := context.Background()
 	cp := flow.StoreCheckpoint(store.NewStore(), "pending-agent")
@@ -64,3 +118,129 @@ func TestPendingReturnsUnfinishedAgentRuns(t *testing.T) {
 		t.Fatalf("Pending = %#v, want run-1", runs)
 	}
 }
+
+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")
+	calls := 0
+	fakeGen = func(ctx context.Context, opts ai.Options, req *ai.Request) (*ai.Response, error) {
+		calls++
+		if opts.ToolHandler != nil {
+			opts.ToolHandler(ctx, ai.ToolCall{ID: "call-1", Name: "external.approve", Input: map[string]any{"id": "42"}})
+		}
+		return &ai.Response{Reply: "model saw approval result"}, nil
+	}
+	defer func() { fakeGen = nil }()
+
+	approved := false
+	a := newTestAgent(Name("approval-agent"), WithCheckpoint(cp),
+		WithTool("external.approve", "guarded external action", nil, func(context.Context, map[string]any) (string, error) { return "ok", nil }),
+		ApproveTool(func(tool string, input map[string]any) (bool, string) {
+			return approved, "waiting for operator"
+		}))
+	_, err := a.Ask(ctx, "send the guarded update")
+	if err == nil {
+		t.Fatal("Ask succeeded, want paused approval error")
+	}
+
+	runs, err := Pending(ctx, a)
+	if err != nil {
+		t.Fatalf("Pending: %v", err)
+	}
+	if len(runs) != 1 {
+		t.Fatalf("Pending returned %d runs, want 1: %#v", len(runs), runs)
+	}
+	if runs[0].Status != "paused" || runs[0].State.Stage != agentApprovalStep {
+		t.Fatalf("run status/stage = %s/%s, want paused/%s", runs[0].Status, runs[0].State.Stage, agentApprovalStep)
+	}
+	if got := string(runs[0].State.Data); got != "send the guarded update" {
+		t.Fatalf("paused run data = %q", got)
+	}
+
+	approved = true
+	fakeGen = func(ctx context.Context, opts ai.Options, req *ai.Request) (*ai.Response, error) {
+		calls++
+		if opts.ToolHandler != nil {
+			res := opts.ToolHandler(ctx, ai.ToolCall{ID: "call-2", Name: "external.approve", Input: map[string]any{"id": "42"}})
+			if res.Refused != "" {
+				t.Fatalf("resumed call was refused: %#v", res)
+			}
+		}
+		return &ai.Response{Reply: "done after approval"}, nil
+	}
+	resp, err := Resume(ctx, a, runs[0].ID)
+	if err != nil {
+		t.Fatalf("Resume: %v", err)
+	}
+	if resp.Reply != "done after approval" {
+		t.Fatalf("Resume reply = %q", resp.Reply)
+	}
+	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)
+	}
+	if calls != 2 {
+		t.Fatalf("model calls = %d, want 2", calls)
+	}
+}

agent/integration_test.go

@@ -179,3 +179,46 @@ func TestDelegateRequiresTask(t *testing.T) {
 		t.Errorf("delegate with no task should error; got %q", content)
 	}
 }
+
+func TestCompactingMemorySummarizesAndRecallsArchivedContext(t *testing.T) {
+	var sawSummary, sawRecall bool
+	fakeGen = func(ctx context.Context, opts ai.Options, req *ai.Request) (*ai.Response, error) {
+		for _, msg := range req.Messages {
+			text := msg.Content.(string)
+			if strings.Contains(text, "Conversation memory summary") && strings.Contains(text, "alpha project") {
+				sawSummary = true
+			}
+			if strings.Contains(text, "alpha project budget is 42") {
+				sawRecall = true
+			}
+		}
+		return &ai.Response{Reply: "ok"}, nil
+	}
+	defer func() { fakeGen = nil }()
+
+	a := newTestAgent(Name("memory"), CompactMemory(4, 2), MemoryRecallLimit(3))
+	turns := []string{
+		"alpha project budget is 42",
+		"beta project owner is sam",
+		"gamma project deadline is monday",
+		"delta project status is green",
+		"epsilon project risk is low",
+	}
+	for _, turn := range turns {
+		if _, err := a.Ask(context.Background(), turn); err != nil {
+			t.Fatalf("Ask(%q): %v", turn, err)
+		}
+	}
+	if got := len(a.mem.Messages()); got > 4 {
+		t.Fatalf("compacted memory retained %d messages, want <= 4", got)
+	}
+	if _, err := a.Ask(context.Background(), "what was the alpha budget?"); err != nil {
+		t.Fatalf("Ask recall: %v", err)
+	}
+	if !sawSummary {
+		t.Error("model request did not include a deterministic compacted summary")
+	}
+	if !sawRecall {
+		t.Error("model request did not recall archived matching context")
+	}
+}

agent/memory.go

@@ -2,6 +2,9 @@ package agent
 
 import (
 	"encoding/json"
+	"fmt"
+	"sort"
+	"strings"
 	"sync"
 
 	"go-micro.dev/v6/ai"
@@ -21,6 +24,21 @@ type Memory interface {
 	Clear()
 }
 
+// MemoryCompaction configures deterministic, store-backed context compaction
+// for the default memory implementation. When the retained conversation grows
+// past MaxMessages, older turns are collapsed into a summary message while the
+// newest KeepRecent turns stay verbatim for provider-neutral continuity.
+type MemoryCompaction struct {
+	MaxMessages int
+	KeepRecent  int
+}
+
+// MemoryRecall is implemented by memory backends that can retrieve durable
+// prior context relevant to a new turn without replaying every stored message.
+type MemoryRecall interface {
+	Recall(query string, limit int) []ai.Message
+}
+
 // NewMemory returns the default store-backed memory: an in-process
 // conversation buffer (truncated to limit) that persists to the store
 // under key, so an agent picks up where it left off after a restart.
@@ -31,6 +49,33 @@ func NewMemory(s store.Store, key string, limit int) Memory {
 	return m
 }
 
+// NewCompactingMemory returns store-backed memory with explicit compaction and
+// retrieval controls. It keeps all messages in the backing store, compacts older
+// turns into a deterministic summary when the conversation exceeds maxMessages,
+// and lets callers recall relevant prior turns with Recall.
+func NewCompactingMemory(s store.Store, key string, maxMessages, keepRecent int) Memory {
+	if keepRecent <= 0 {
+		keepRecent = maxMessages / 2
+	}
+	if keepRecent < 1 {
+		keepRecent = 1
+	}
+	m := &storeMemory{
+		store: s,
+		key:   key,
+		// Use an unlimited buffer here; compaction, not truncation, decides
+		// what remains in active context so a summary can preserve older turns.
+		hist: ai.NewHistory(0),
+		compaction: MemoryCompaction{
+			MaxMessages: maxMessages,
+			KeepRecent:  keepRecent,
+		},
+	}
+	m.load()
+	m.compact()
+	return m
+}
+
 // NewInMemory returns conversation memory that is not persisted.
 func NewInMemory(limit int) Memory {
 	return &storeMemory{hist: ai.NewHistory(limit)}
@@ -39,16 +84,19 @@ func NewInMemory(limit int) Memory {
 // storeMemory is the default Memory: an ai.History buffer optionally
 // persisted to a store.
 type storeMemory struct {
-	mu    sync.Mutex
-	store store.Store
-	key   string
-	hist  *ai.History
+	mu         sync.Mutex
+	store      store.Store
+	key        string
+	hist       *ai.History
+	compaction MemoryCompaction
+	archive    []ai.Message
 }
 
 func (m *storeMemory) Add(role, content string) {
 	m.mu.Lock()
 	m.hist.Add(role, content)
 	m.mu.Unlock()
+	m.compact()
 	m.save()
 }
 
@@ -61,10 +109,49 @@ func (m *storeMemory) Messages() []ai.Message {
 func (m *storeMemory) Clear() {
 	m.mu.Lock()
 	m.hist.Reset()
+	m.archive = nil
 	m.mu.Unlock()
 	m.save()
 }
 
+// Recall returns archived messages whose content contains words from query.
+// It is deterministic and provider-neutral: no embeddings or model calls are
+// required, but semantic/vector stores can replace Memory for richer retrieval.
+func (m *storeMemory) Recall(query string, limit int) []ai.Message {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	if limit <= 0 {
+		limit = 5
+	}
+	terms := recallTerms(query)
+	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]
+		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
+}
+
 func (m *storeMemory) load() {
 	if m.store == nil || m.key == "" {
 		return
@@ -73,12 +160,17 @@ func (m *storeMemory) load() {
 	if err != nil || len(recs) == 0 {
 		return
 	}
-	var msgs []ai.Message
-	if err := json.Unmarshal(recs[0].Value, &msgs); err != nil {
-		return
+	var state memoryState
+	if err := json.Unmarshal(recs[0].Value, &state); err != nil {
+		var msgs []ai.Message
+		if err := json.Unmarshal(recs[0].Value, &msgs); err != nil {
+			return
+		}
+		state.Messages = msgs
 	}
 	m.mu.Lock()
-	for _, msg := range msgs {
+	m.archive = state.Archive
+	for _, msg := range state.Messages {
 		m.hist.Add(msg.Role, msg.Content)
 	}
 	m.mu.Unlock()
@@ -89,10 +181,94 @@ func (m *storeMemory) save() {
 		return
 	}
 	m.mu.Lock()
-	data, err := json.Marshal(m.hist.Messages())
+	data, err := json.Marshal(memoryState{
+		Messages: m.hist.Messages(),
+		Archive:  m.archive,
+	})
 	m.mu.Unlock()
 	if err != nil {
 		return
 	}
 	_ = m.store.Write(&store.Record{Key: m.key, Value: data})
 }
+
+func (m *storeMemory) compact() {
+	if m.compaction.MaxMessages <= 0 {
+		return
+	}
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	msgs := m.hist.Messages()
+	if len(msgs) <= m.compaction.MaxMessages {
+		return
+	}
+	keep := m.compaction.KeepRecent
+	if keep <= 0 || keep >= m.compaction.MaxMessages {
+		keep = m.compaction.MaxMessages - 1
+	}
+	if keep < 1 {
+		keep = 1
+	}
+	cut := len(msgs) - keep
+	older := msgs[:cut]
+	recent := msgs[cut:]
+	m.archive = append(m.archive, older...)
+	summary := ai.Message{
+		Role:    "system",
+		Content: fmt.Sprintf("Conversation memory summary: %s", summarizeMessages(older)),
+	}
+	m.hist.Reset()
+	m.hist.Add(summary.Role, summary.Content)
+	for _, msg := range recent {
+		m.hist.Add(msg.Role, msg.Content)
+	}
+}
+
+func summarizeMessages(msgs []ai.Message) string {
+	var b strings.Builder
+	for i, msg := range msgs {
+		if i > 0 {
+			b.WriteString(" | ")
+		}
+		fmt.Fprintf(&b, "%s: %s", msg.Role, compactText(fmt.Sprint(msg.Content), 120))
+	}
+	return b.String()
+}
+
+func compactText(s string, max int) string {
+	s = strings.Join(strings.Fields(s), " ")
+	if max > 0 && len(s) > max {
+		return s[:max] + "…"
+	}
+	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
+	for _, term := range strings.Fields(strings.ToLower(query)) {
+		term = strings.Trim(term, ".,!?;:\"'()[]{}")
+		if len(term) < 3 || seen[term] {
+			continue
+		}
+		seen[term] = true
+		terms = append(terms, term)
+	}
+	return terms
+}
+
+type memoryState struct {
+	Messages []ai.Message `json:"messages"`
+	Archive  []ai.Message `json:"archive,omitempty"`
+}

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

@@ -60,6 +60,13 @@ type Options struct {
 	// Memory is the agent's conversation memory. Nil = the default
 	// store-backed memory (durable across restarts).
 	Memory Memory
+	// MemoryCompaction enables deterministic compaction/retrieval on the
+	// default store-backed memory. Custom Memory implementations can expose
+	// retrieval by implementing MemoryRecall.
+	MemoryCompaction MemoryCompaction
+	// MemoryRecallLimit bounds recalled archived turns injected into a model
+	// request (0 disables recall injection).
+	MemoryRecallLimit int
 	// Checkpoint persists agent Ask runs so callers can resume by run id
 	// after a restart without replaying a run that already completed.
 	Checkpoint flow.Checkpoint
@@ -215,6 +222,25 @@ func WithMemory(m Memory) Option {
 	return func(o *Options) { o.Memory = m }
 }
 
+// CompactMemory enables deterministic, store-backed memory compaction for the
+// default agent memory. Older turns are summarized once active context exceeds
+// maxMessages, keepRecent newest turns remain verbatim, and recalled archived
+// turns are injected into matching future asks.
+func CompactMemory(maxMessages, keepRecent int) Option {
+	return func(o *Options) {
+		o.MemoryCompaction = MemoryCompaction{MaxMessages: maxMessages, KeepRecent: keepRecent}
+		if o.MemoryRecallLimit == 0 {
+			o.MemoryRecallLimit = 5
+		}
+	}
+}
+
+// MemoryRecallLimit sets how many archived turns a memory backend may inject
+// into a model request for the current Ask. Use 0 to disable retrieval.
+func MemoryRecallLimit(n int) Option {
+	return func(o *Options) { o.MemoryRecallLimit = n }
+}
+
 // WithCheckpoint sets the durability backend for agent Ask runs. The
 // Checkpoint interface is shared with flow so services, agents, and workflows
 // can use one execution history backend. When set, each Ask is saved as a

agent/otel.go

@@ -253,9 +253,40 @@ 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.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

@@ -71,6 +71,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
@@ -115,6 +125,19 @@ func TestAgentOpenTelemetrySpans(t *testing.T) {
 	}
 }
 
+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 {

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"
@@ -91,13 +92,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 +151,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/model.go

@@ -116,14 +116,18 @@ const (
 // the agent package. Flows also attach their name and current step so
 // tools and agents called from a workflow can be tied back to the
 // services → agents → workflows lifecycle that invoked them. Per-call
-// detail (tool name, id) is on the ToolCall; attempt counts are naturally
-// counted by the wrapper itself.
+// detail (tool name, id) is on the ToolCall. Attempt and MaxAttempts are
+// set while a model Generate call is in progress, so tools and wrappers can
+// tell which provider attempt produced the call and whether it is part of a
+// retry budget. They are zero when no model-attempt context is known.
 type RunInfo struct {
-	RunID    string // correlation id for this agent or flow run
-	ParentID string // the run that delegated to this one, if any
-	Agent    string // the agent's name
-	Flow     string // the flow's name, when the call is part of a workflow
-	Step     string // the flow step currently executing, when known
+	RunID       string // correlation id for this agent or flow run
+	ParentID    string // the run that delegated to this one, if any
+	Agent       string // the agent's name
+	Flow        string // the flow's name, when the call is part of a workflow
+	Step        string // the flow step currently executing, when known
+	Attempt     int    // current model Generate attempt, starting at 1 when known
+	MaxAttempts int    // configured model Generate attempt budget when known
 }
 
 type runInfoKey struct{}

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,9 +13,34 @@ 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
+
+const (
+	ErrorKindUnknown     ErrorKind = "unknown"
+	ErrorKindCanceled    ErrorKind = "canceled"
+	ErrorKindTimeout     ErrorKind = "timeout"
+	ErrorKindRateLimited ErrorKind = "rate_limited"
+	ErrorKindUnavailable ErrorKind = "unavailable"
+	ErrorKindProvider    ErrorKind = "provider"
+)
+
+// ClassifiedError is implemented by errors that expose a stable ErrorKind.
+type ClassifiedError interface {
+	ErrorKind() ErrorKind
+}
+
 // RetryError is returned when Generate is retried and still fails.
 type RetryError struct {
 	Attempts int
+	Kind     ErrorKind
 	Err      error
 }
 
@@ -23,7 +48,7 @@ func (e *RetryError) Error() string {
 	if e == nil {
 		return ""
 	}
-	return fmt.Sprintf("ai generate failed after %d attempt(s): %v", e.Attempts, e.Err)
+	return fmt.Sprintf("ai generate failed after %d attempt(s) (%s): %v", e.Attempts, e.ErrorKind(), e.Err)
 }
 
 func (e *RetryError) Unwrap() error {
@@ -33,6 +58,13 @@ func (e *RetryError) Unwrap() error {
 	return e.Err
 }
 
+func (e *RetryError) ErrorKind() ErrorKind {
+	if e == nil || e.Kind == "" {
+		return ErrorKindUnknown
+	}
+	return e.Kind
+}
+
 // GeneratePolicy controls timeout and retry behavior for a model call.
 type GeneratePolicy struct {
 	Timeout     time.Duration
@@ -60,6 +92,11 @@ func GenerateWithRetry(ctx context.Context, m Model, req *Request, policy Genera
 		if policy.Timeout > 0 {
 			callCtx, cancel = context.WithTimeout(ctx, policy.Timeout)
 		}
+		if info, ok := RunInfoFrom(callCtx); ok {
+			info.Attempt = attempt
+			info.MaxAttempts = policy.MaxAttempts
+			callCtx = WithRunInfo(callCtx, info)
+		}
 		resp, err := m.Generate(callCtx, req, opts...)
 		cancel()
 		if err == nil {
@@ -71,9 +108,10 @@ func GenerateWithRetry(ctx context.Context, m Model, req *Request, policy Genera
 		if ctx.Err() != nil {
 			return nil, ctx.Err()
 		}
-		if attempt == policy.MaxAttempts || !IsTransientError(err) {
-			if attempt > 1 || IsTransientError(err) {
-				return nil, &RetryError{Attempts: attempt, Err: err}
+		transient := IsTransientError(err)
+		if attempt == policy.MaxAttempts || !transient {
+			if attempt > 1 || transient {
+				return nil, &RetryError{Attempts: attempt, Kind: ClassifyError(err), Err: err}
 			}
 			return nil, err
 		}
@@ -81,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():
@@ -101,25 +130,81 @@ func GenerateWithRetry(ctx context.Context, m Model, req *Request, policy Genera
 		case <-t.C:
 		}
 	}
-	return nil, &RetryError{Attempts: policy.MaxAttempts, Err: last}
+	return nil, &RetryError{Attempts: policy.MaxAttempts, Kind: ClassifyError(last), Err: last}
 }
 
-// IsTransientError reports whether err is worth retrying at the provider boundary.
-func IsTransientError(err error) bool {
+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 {
-		return false
+		return ""
+	}
+	var classified ClassifiedError
+	if errors.As(err, &classified) {
+		if kind := classified.ErrorKind(); kind != "" {
+			return kind
+		}
 	}
 	if errors.Is(err, context.Canceled) {
-		return false
+		return ErrorKindCanceled
 	}
 	if errors.Is(err, context.DeadlineExceeded) {
-		return true
+		return ErrorKindTimeout
 	}
 	var sc StatusCoder
 	if errors.As(err, &sc) {
 		code := sc.StatusCode()
-		return code == 429 || code >= 500
+		switch {
+		case code == 429:
+			return ErrorKindRateLimited
+		case code >= 500:
+			return ErrorKindUnavailable
+		case code > 0:
+			return ErrorKindProvider
+		}
 	}
 	msg := strings.ToLower(err.Error())
-	return strings.Contains(msg, "rate limit") || strings.Contains(msg, "too many requests") || strings.Contains(msg, "timeout") || strings.Contains(msg, "temporar")
+	switch {
+	case strings.Contains(msg, "rate limit") || strings.Contains(msg, "too many requests"):
+		return ErrorKindRateLimited
+	case strings.Contains(msg, "timeout") || strings.Contains(msg, "deadline"):
+		return ErrorKindTimeout
+	case strings.Contains(msg, "temporar") || strings.Contains(msg, "unavailable"):
+		return ErrorKindUnavailable
+	default:
+		return ErrorKindUnknown
+	}
+}
+
+// IsTransientError reports whether err is worth retrying at the provider boundary.
+func IsTransientError(err error) bool {
+	switch ClassifyError(err) {
+	case ErrorKindTimeout, ErrorKindRateLimited, ErrorKindUnavailable:
+		return true
+	default:
+		return false
+	}
 }

ai/retry_test.go

@@ -94,3 +94,130 @@ func TestGenerateWithRetryHonorsPerAttemptTimeout(t *testing.T) {
 		t.Fatalf("attempts = %d, want 2", attempts)
 	}
 }
+
+func TestGenerateWithRetryAddsAttemptMetadataToRunInfo(t *testing.T) {
+	var got []RunInfo
+	model := retryModel{generate: func(ctx context.Context, _ *Request, _ ...GenerateOption) (*Response, error) {
+		info, ok := RunInfoFrom(ctx)
+		if !ok {
+			t.Fatal("RunInfo missing from attempt context")
+		}
+		got = append(got, info)
+		if info.Attempt == 1 {
+			return nil, errors.New("temporary provider outage")
+		}
+		return &Response{Reply: "ok"}, nil
+	}}
+
+	ctx := WithRunInfo(context.Background(), RunInfo{RunID: "run-1", Agent: "worker"})
+	_, err := GenerateWithRetry(ctx, model, &Request{Prompt: "hi"}, GeneratePolicy{
+		MaxAttempts: 2,
+		Backoff:     time.Millisecond,
+	})
+	if err != nil {
+		t.Fatalf("GenerateWithRetry returned error: %v", err)
+	}
+	if len(got) != 2 {
+		t.Fatalf("attempt contexts = %d, want 2", len(got))
+	}
+	for i, info := range got {
+		wantAttempt := i + 1
+		if info.Attempt != wantAttempt {
+			t.Fatalf("attempt %d RunInfo.Attempt = %d, want %d", i, info.Attempt, wantAttempt)
+		}
+		if info.MaxAttempts != 2 {
+			t.Fatalf("attempt %d RunInfo.MaxAttempts = %d, want 2", i, info.MaxAttempts)
+		}
+		if info.RunID != "run-1" || info.Agent != "worker" {
+			t.Fatalf("attempt %d RunInfo identity = (%q, %q), want (run-1, worker)", i, info.RunID, info.Agent)
+		}
+	}
+}
+
+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
+		err  error
+		want ErrorKind
+	}{
+		{name: "canceled", err: context.Canceled, want: ErrorKindCanceled},
+		{name: "timeout", err: context.DeadlineExceeded, want: ErrorKindTimeout},
+		{name: "rate limit status", err: statusErr(429), want: ErrorKindRateLimited},
+		{name: "unavailable status", err: statusErr(503), want: ErrorKindUnavailable},
+		{name: "provider status", err: statusErr(400), want: ErrorKindProvider},
+		{name: "rate limit text", err: errors.New("rate limit exceeded"), want: ErrorKindRateLimited},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			if got := ClassifyError(tt.err); got != tt.want {
+				t.Fatalf("ClassifyError() = %q, want %q", got, tt.want)
+			}
+		})
+	}
+}
+
+func TestGenerateWithRetryExposesRetryErrorKind(t *testing.T) {
+	model := retryModel{generate: func(context.Context, *Request, ...GenerateOption) (*Response, error) {
+		return nil, statusErr(429)
+	}}
+
+	_, err := GenerateWithRetry(context.Background(), model, &Request{Prompt: "hi"}, GeneratePolicy{
+		MaxAttempts: 2,
+		Backoff:     time.Millisecond,
+	})
+	var retryErr *RetryError
+	if !errors.As(err, &retryErr) {
+		t.Fatalf("error = %T %[1]v, want RetryError", err)
+	}
+	if retryErr.ErrorKind() != ErrorKindRateLimited {
+		t.Fatalf("retry kind = %q, want %q", retryErr.ErrorKind(), ErrorKindRateLimited)
+	}
+	if !errors.Is(err, statusErr(429)) {
+		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)
+	}
+}

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")
+	}
+}

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

@@ -20,9 +20,9 @@
 //
 // Scope of this version: the JSON-RPC binding — `message/send`
 // (returns a completed Task), `message/stream` (SSE with the completed
-// Task event), `tasks/get`, and Agent Card discovery. Multi-turn
-// `input-required`, `tasks/resubscribe`, and push notifications are
-// advertised as unsupported and are follow-ups.
+// Task event), `tasks/get`, multi-turn task continuation, push
+// notification delivery, input-required handoffs, `tasks/resubscribe`,
+// and Agent Card discovery.
 package a2a
 
 import (
@@ -225,11 +225,21 @@ type Task struct {
 	Kind      string     `json:"kind"` // "task"
 }
 
+// PushNotificationConfig tells the gateway where to POST task updates for a
+// task. The gateway stores one config per task and delivers best-effort JSON
+// task snapshots whenever that task changes.
+type PushNotificationConfig struct {
+	URL            string            `json:"url"`
+	Token          string            `json:"token,omitempty"`
+	Authentication map[string]string `json:"authentication,omitempty"`
+}
+
 // 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.
@@ -334,7 +344,7 @@ func Card(name, url, description string, services []string) AgentCard {
 		URL:             url,
 		Version:         "1.0.0",
 		ProtocolVersion: protocolVersion,
-		Capabilities:    Capabilities{Streaming: true, PushNotifications: false},
+		Capabilities:    Capabilities{Streaming: true, PushNotifications: true},
 		// The agent converses over a single Chat endpoint; advertise that
 		// as one skill, tagged with the services it manages.
 		DefaultInputModes:  []string{"text/plain"},
@@ -415,12 +425,16 @@ func (g *Gateway) handleRPC(w http.ResponseWriter, r *http.Request) {
 // retains recent tasks for tasks/get. It is shared by the gateway (one
 // per registry) and embedded agents (one per agent).
 type dispatcher struct {
-	mu    sync.Mutex
-	tasks map[string]*Task
-	order []string // task ids in insertion order, for bounded eviction
+	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{}} }
+func newDispatcher() *dispatcher {
+	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) {
 	d.serveWithStream(w, r, invoke, nil)
@@ -448,11 +462,15 @@ func (d *dispatcher) serveWithStream(w http.ResponseWriter, r *http.Request, inv
 		d.stream(requestContext(r.Context()), w, req, invoke)
 	case "tasks/get":
 		d.get(w, req)
+	case "tasks/pushNotificationConfig/set":
+		d.setPushConfig(w, req)
+	case "tasks/pushNotificationConfig/get":
+		d.getPushConfig(w, req)
 	case "tasks/cancel":
 		// 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})
 	}
@@ -541,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()
 	}
@@ -561,8 +580,12 @@ 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 := taskFromReply(p.Message, reply, state)
+	task := d.taskFromReply(p.Message, reply, state)
 	d.store(task)
 	return task, nil
 }
@@ -571,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 == "" {
@@ -587,6 +653,47 @@ func (d *dispatcher) get(w http.ResponseWriter, req rpcRequest) {
 	writeRPC(w, req.ID, task, nil)
 }
 
+type pushConfigParams struct {
+	ID                     string                 `json:"id"`
+	PushNotificationConfig PushNotificationConfig `json:"pushNotificationConfig"`
+}
+
+func (d *dispatcher) setPushConfig(w http.ResponseWriter, req rpcRequest) {
+	var p pushConfigParams
+	if err := json.Unmarshal(req.Params, &p); err != nil || p.ID == "" || p.PushNotificationConfig.URL == "" {
+		writeRPC(w, req.ID, nil, &rpcError{Code: errInvalidParams, Message: "invalid params"})
+		return
+	}
+	d.mu.Lock()
+	task := d.tasks[p.ID]
+	if task != nil {
+		d.pushConfigs[p.ID] = p.PushNotificationConfig
+	}
+	d.mu.Unlock()
+	if task == nil {
+		writeRPC(w, req.ID, nil, &rpcError{Code: errTaskNotFound, Message: "task not found"})
+		return
+	}
+	writeRPC(w, req.ID, map[string]any{"id": p.ID, "pushNotificationConfig": p.PushNotificationConfig}, nil)
+	go d.deliverPush(p.ID, task)
+}
+
+func (d *dispatcher) getPushConfig(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
+	}
+	d.mu.Lock()
+	cfg, ok := d.pushConfigs[p.ID]
+	d.mu.Unlock()
+	if !ok {
+		writeRPC(w, req.ID, nil, &rpcError{Code: errTaskNotFound, Message: "push notification config not found"})
+		return
+	}
+	writeRPC(w, req.ID, map[string]any{"id": p.ID, "pushNotificationConfig": cfg}, nil)
+}
+
 // ---------------------------------------------------------------------------
 // agent RPC
 // ---------------------------------------------------------------------------
@@ -615,30 +722,96 @@ func (g *Gateway) callAgent(ctx context.Context, name, message string) (string,
 
 func (d *dispatcher) store(t *Task) {
 	d.mu.Lock()
-	defer d.mu.Unlock()
+	_, 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 taskFromReply(input Message, reply, state string) *Task {
+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
+	var history []Message
+	if taskID != "" {
+		d.mu.Lock()
+		prev := d.tasks[taskID]
+		if prev != nil {
+			contextID = prev.ContextID
+			history = append(history, prev.History...)
+		}
+		d.mu.Unlock()
+	}
+	if taskID == "" {
+		taskID = uuid.New().String()
+	}
 	if contextID == "" {
 		contextID = uuid.New().String()
 	}
-	return taskFromReplyWithIDs(input, reply, state, uuid.New().String(), contextID)
+	return taskFromReplyWithIDsAndHistory(input, reply, state, taskID, contextID, history)
 }
 
 func taskFromReplyWithIDs(input Message, reply, state, taskID, contextID string) *Task {
+	return taskFromReplyWithIDsAndHistory(input, reply, state, taskID, contextID, nil)
+}
+
+func taskFromReplyWithIDsAndHistory(input Message, reply, state, taskID, contextID string, history []Message) *Task {
+	input.TaskID = taskID
+	input.ContextID = contextID
+	if input.Kind == "" {
+		input.Kind = "message"
+	}
 	task := &Task{
 		ID:        taskID,
 		ContextID: contextID,
 		Kind:      "task",
-		History:   []Message{input},
+		History:   append(append([]Message{}, history...), input),
 		Status:    TaskStatus{State: state, Timestamp: time.Now().UTC().Format(time.RFC3339)},
 		Artifacts: []Artifact{textArtifact(reply)},
 	}
@@ -653,6 +826,33 @@ func taskFromReplyWithIDs(input Message, reply, state, taskID, contextID string)
 	return task
 }
 
+func (d *dispatcher) deliverPush(taskID string, task *Task) {
+	d.mu.Lock()
+	cfg, ok := d.pushConfigs[taskID]
+	d.mu.Unlock()
+	if !ok || cfg.URL == "" || task == nil {
+		return
+	}
+	body, err := json.Marshal(task)
+	if err != nil {
+		return
+	}
+	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+	defer cancel()
+	req, err := http.NewRequestWithContext(ctx, http.MethodPost, cfg.URL, strings.NewReader(string(body)))
+	if err != nil {
+		return
+	}
+	req.Header.Set("Content-Type", "application/json")
+	if cfg.Token != "" {
+		req.Header.Set("Authorization", "Bearer "+cfg.Token)
+	}
+	resp, err := http.DefaultClient.Do(req)
+	if err == nil && resp.Body != nil {
+		_ = resp.Body.Close()
+	}
+}
+
 func textOf(parts []Part) string {
 	var b strings.Builder
 	for _, p := range parts {

gateway/a2a/a2a_test.go

@@ -4,6 +4,7 @@ import (
 	"bytes"
 	"context"
 	"encoding/json"
+	"errors"
 	"fmt"
 	"io"
 	"net/http"
@@ -120,6 +121,89 @@ func TestMessageSendAndGet(t *testing.T) {
 	}
 }
 
+func TestMessageSendContinuesExistingTask(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":"first"}]}}}`, func(_ context.Context, text string) (string, error) {
+		return "reply to " + text, nil
+	})
+
+	secondBody := 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":"second"}]}}}`, first.ID, first.ContextID)
+	second := rpcTaskFromBody(t, d, secondBody, func(_ context.Context, text string) (string, error) {
+		return "reply to " + text, nil
+	})
+
+	if second.ID != first.ID || second.ContextID != first.ContextID {
+		t.Fatalf("continued task identity = %s/%s, want %s/%s", second.ID, second.ContextID, first.ID, first.ContextID)
+	}
+	if len(second.History) != 4 {
+		t.Fatalf("continued history len = %d, want 4: %+v", len(second.History), second.History)
+	}
+	if textOf(second.History[0].Parts) != "first" || textOf(second.History[2].Parts) != "second" {
+		t.Fatalf("continued history did not preserve turns: %+v", second.History)
+	}
+
+	got := rpcTaskFromDispatcher(t, d, first.ID)
+	if got.ID != first.ID || len(got.History) != 4 {
+		t.Fatalf("stored continued task = %+v", got)
+	}
+}
+
+func TestPushNotificationConfigDeliversTaskUpdates(t *testing.T) {
+	d := newDispatcher()
+	updates := make(chan Task, 2)
+	push := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		if got := r.Header.Get("Authorization"); got != "Bearer secret" {
+			t.Errorf("authorization = %q, want bearer token", got)
+		}
+		var task Task
+		if err := json.NewDecoder(r.Body).Decode(&task); err != nil {
+			t.Errorf("decode push task: %v", err)
+			return
+		}
+		updates <- task
+		w.WriteHeader(http.StatusAccepted)
+	}))
+	defer push.Close()
+
+	task := rpcTaskFromBody(t, d, `{
+		"jsonrpc":"2.0","id":1,"method":"message/send",
+		"params":{"message":{"role":"user","kind":"message","messageId":"m1",
+			"parts":[{"kind":"text","text":"ping"}]}}}`, func(_ context.Context, text string) (string, error) {
+		return "pong", nil
+	})
+
+	body := fmt.Sprintf(`{"jsonrpc":"2.0","id":2,"method":"tasks/pushNotificationConfig/set","params":{"id":"%s","pushNotificationConfig":{"url":"%s","token":"secret"}}}`, task.ID, push.URL)
+	var setResp struct {
+		Result struct {
+			ID                     string                 `json:"id"`
+			PushNotificationConfig PushNotificationConfig `json:"pushNotificationConfig"`
+		} `json:"result"`
+		Error *rpcError `json:"error"`
+	}
+	rpcDispatcher(t, d, body, nil, &setResp)
+	if setResp.Error != nil {
+		t.Fatalf("set push config error: %+v", setResp.Error)
+	}
+	if setResp.Result.ID != task.ID || setResp.Result.PushNotificationConfig.URL != push.URL {
+		t.Fatalf("set push config result = %+v", setResp.Result)
+	}
+
+	select {
+	case got := <-updates:
+		if got.ID != task.ID || got.Status.State != stateCompleted || textOf(got.Artifacts[0].Parts) != "pong" {
+			t.Fatalf("push update = %+v, want completed task", got)
+		}
+	case <-time.After(2 * time.Second):
+		t.Fatal("timed out waiting for push update")
+	}
+}
+
 func TestMessageSendUsesRequestContext(t *testing.T) {
 	d := newDispatcher()
 	ctx, cancel := context.WithCancel(context.Background())
@@ -276,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)
@@ -295,6 +566,29 @@ func rpcTaskFromDispatcher(t *testing.T, d *dispatcher, id string) Task {
 	return resp.Result
 }
 
+func rpcTaskFromBody(t *testing.T, d *dispatcher, body string, invoke Invoke) Task {
+	t.Helper()
+	var resp struct {
+		Result Task      `json:"result"`
+		Error  *rpcError `json:"error"`
+	}
+	rpcDispatcher(t, d, body, invoke, &resp)
+	if resp.Error != nil {
+		t.Fatalf("rpc error: %+v", resp.Error)
+	}
+	return resp.Result
+}
+
+func rpcDispatcher(t *testing.T, d *dispatcher, body string, invoke Invoke, v any) {
+	t.Helper()
+	req := httptest.NewRequest(http.MethodPost, "/", bytes.NewBufferString(body))
+	rr := httptest.NewRecorder()
+	d.serve(rr, req, invoke)
+	if err := json.NewDecoder(rr.Result().Body).Decode(v); err != nil {
+		t.Fatalf("decode dispatcher response: %v", err)
+	}
+}
+
 func TestUnknownMethod(t *testing.T) {
 	ts, cleanup := newGatewayWithAgent(t)
 	defer cleanup()
@@ -302,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/a2a/client.go

@@ -60,15 +60,38 @@ func (c *Client) Card(ctx context.Context) (*AgentCard, error) {
 // If the agent returns a task that isn't yet terminal, Send polls
 // tasks/get until it completes or ctx is done.
 func (c *Client) Send(ctx context.Context, text string) (string, error) {
-	res, err := c.call(ctx, "message/send", sendParams{Message: Message{
+	task, err := c.SendMessage(ctx, Message{
 		Role:      "user",
 		Kind:      "message",
 		MessageID: uuid.New().String(),
 		Parts:     []Part{{Kind: "text", Text: text}},
-	}})
+	})
 	if err != nil {
 		return "", err
 	}
+	if task.Status.State != stateCompleted {
+		return "", fmt.Errorf("remote task %s ended in state %q", task.ID, task.Status.State)
+	}
+	return artifactsText(task.Artifacts), nil
+}
+
+// SendMessage sends an A2A message and returns the resulting terminal task.
+// To continue a multi-turn task, pass a Message with TaskID and ContextID set
+// to a prior task's id and context id.
+func (c *Client) SendMessage(ctx context.Context, message Message) (*Task, error) {
+	if message.MessageID == "" {
+		message.MessageID = uuid.New().String()
+	}
+	if message.Kind == "" {
+		message.Kind = "message"
+	}
+	if message.Role == "" {
+		message.Role = "user"
+	}
+	res, err := c.call(ctx, "message/send", sendParams{Message: message})
+	if err != nil {
+		return nil, err
+	}
 
 	// The result is a Message or a Task; the "kind" field disambiguates.
 	var probe struct {
@@ -79,33 +102,61 @@ func (c *Client) Send(ctx context.Context, text string) (string, error) {
 	if probe.Kind == "message" {
 		var m Message
 		if err := json.Unmarshal(res, &m); err != nil {
-			return "", err
+			return nil, err
 		}
-		return textOf(m.Parts), nil
+		return &Task{
+			ID:        m.TaskID,
+			ContextID: m.ContextID,
+			Kind:      "task",
+			Status:    TaskStatus{State: stateCompleted, Timestamp: time.Now().UTC().Format(time.RFC3339)},
+			Artifacts: []Artifact{textArtifact(textOf(m.Parts))},
+			History:   []Message{m},
+		}, nil
 	}
 
 	var task Task
 	if err := json.Unmarshal(res, &task); err != nil {
-		return "", err
+		return nil, err
 	}
 	for !terminal(task.Status.State) {
 		select {
 		case <-ctx.Done():
-			return "", ctx.Err()
+			return nil, ctx.Err()
 		case <-time.After(300 * time.Millisecond):
 		}
 		got, err := c.call(ctx, "tasks/get", getParams{ID: task.ID})
 		if err != nil {
-			return "", err
+			return nil, err
 		}
 		if err := json.Unmarshal(got, &task); err != nil {
-			return "", err
+			return nil, err
 		}
 	}
-	if task.Status.State != stateCompleted {
-		return "", fmt.Errorf("remote task %s ended in state %q", task.ID, task.Status.State)
+	return &task, nil
+}
+
+// SetPushNotificationConfig asks the remote agent to POST updates for taskID to cfg.URL.
+func (c *Client) SetPushNotificationConfig(ctx context.Context, taskID string, cfg PushNotificationConfig) error {
+	_, err := c.call(ctx, "tasks/pushNotificationConfig/set", pushConfigParams{
+		ID:                     taskID,
+		PushNotificationConfig: cfg,
+	})
+	return err
+}
+
+// PushNotificationConfig returns the remote push notification config for taskID.
+func (c *Client) PushNotificationConfig(ctx context.Context, taskID string) (PushNotificationConfig, error) {
+	res, err := c.call(ctx, "tasks/pushNotificationConfig/get", getParams{ID: taskID})
+	if err != nil {
+		return PushNotificationConfig{}, err
 	}
-	return artifactsText(task.Artifacts), nil
+	var out struct {
+		PushNotificationConfig PushNotificationConfig `json:"pushNotificationConfig"`
+	}
+	if err := json.Unmarshal(res, &out); err != nil {
+		return PushNotificationConfig{}, err
+	}
+	return out.PushNotificationConfig, nil
 }
 
 // call performs one JSON-RPC request and returns the raw result.

gateway/a2a/client_test.go

@@ -2,8 +2,11 @@ package a2a
 
 import (
 	"context"
+	"encoding/json"
+	"net/http"
 	"net/http/httptest"
 	"testing"
+	"time"
 )
 
 // An agent that embeds NewAgentHandler is directly A2A-queryable — no
@@ -48,3 +51,62 @@ func TestClientSendAndCard(t *testing.T) {
 		t.Errorf("Send reply = %q, want pong", reply)
 	}
 }
+
+func TestClientContinuesTaskAndConfiguresPush(t *testing.T) {
+	card := Card("solo", "http://localhost:4000", "", []string{"task"})
+	h := NewAgentHandler(card, func(_ context.Context, text string) (string, error) {
+		return "echo:" + text, nil
+	})
+	ts := httptest.NewServer(h)
+	defer ts.Close()
+
+	updates := make(chan Task, 1)
+	push := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		var task Task
+		if err := json.NewDecoder(r.Body).Decode(&task); err != nil {
+			t.Errorf("decode push task: %v", err)
+			return
+		}
+		updates <- task
+		w.WriteHeader(http.StatusAccepted)
+	}))
+	defer push.Close()
+
+	cl := NewClient(ts.URL)
+	first, err := cl.SendMessage(context.Background(), Message{
+		Parts: []Part{{Kind: "text", Text: "one"}},
+	})
+	if err != nil {
+		t.Fatalf("first SendMessage: %v", err)
+	}
+	second, err := cl.SendMessage(context.Background(), Message{
+		TaskID:    first.ID,
+		ContextID: first.ContextID,
+		Parts:     []Part{{Kind: "text", Text: "two"}},
+	})
+	if err != nil {
+		t.Fatalf("second SendMessage: %v", err)
+	}
+	if second.ID != first.ID || second.ContextID != first.ContextID || len(second.History) != 4 {
+		t.Fatalf("continued task = %+v, first %+v", second, first)
+	}
+	cfg := PushNotificationConfig{URL: push.URL}
+	if err := cl.SetPushNotificationConfig(context.Background(), second.ID, cfg); err != nil {
+		t.Fatalf("SetPushNotificationConfig: %v", err)
+	}
+	got, err := cl.PushNotificationConfig(context.Background(), second.ID)
+	if err != nil {
+		t.Fatalf("PushNotificationConfig: %v", err)
+	}
+	if got.URL != push.URL {
+		t.Fatalf("PushNotificationConfig URL = %q, want %q", got.URL, push.URL)
+	}
+	select {
+	case update := <-updates:
+		if update.ID != second.ID {
+			t.Fatalf("push update ID = %q, want %q", update.ID, second.ID)
+		}
+	case <-time.After(2 * time.Second):
+		t.Fatal("timed out waiting for push update")
+	}
+}

internal/docs/AGENT_CONFORMANCE.md

@@ -28,3 +28,12 @@ consistent across providers.
 
 The companion `TestAgentProviderConformanceFakeError` keeps provider error
 propagation covered locally without relying on external credentials.
+
+## Scheduled CI
+
+The daily/manual `Harness (E2E)` workflow runs the same matrix with
+`GO_MICRO_AGENT_CONFORMANCE_LIVE=1` and the provider secrets exported. Providers
+whose keys are absent still skip cleanly, while any configured provider must pass
+the shared tool-calling scenario. This keeps scheduled conformance key-gated: PR
+checks stay deterministic and no-key environments remain green, but maintained
+provider credentials exercise the live matrix regularly.

internal/docs/PRIORITIES.md

@@ -19,22 +19,10 @@ redirect the loop; direction always wins.
 items the loop can auto-merge): brand/positioning copy, breaking public-API
 changes, architectural rewrites. Those go to the human.
 
-## Now (ranked)
+## Developer experience (ranked)
 
-1. **End-to-end agent streaming** (#3200) — complete the roadmap streaming slice:
-   usable `ai.Stream` beyond the current OpenAI path, A2A `message/stream` chunk
-   delivery, cancellation/error semantics, and docs so chat and long-task UX are
-   real across the harness boundary. Roadmap → *Next* streaming, but it now ranks
-   first because durable resume, provider conformance, registry readiness, 0→hero,
-   and agent OpenTelemetry have shipped; streaming is the largest remaining seam
-   in the services → agents → workflows interaction loop.
-2. **Execution lifecycle hooks & metadata** (#2980) — before/after-tool, retry,
-   and failure hooks; first reconcile the issue with the shipped `AgentWrapTool`,
-   structured refusal reasons, `RunInfo`, and OpenTelemetry spans, then close it or
-   scope only a CI-verifiable gap that wrappers plus tracing cannot express.
-   Roadmap → resilience/operability, but ranked after streaming because most of
-   the originally requested lifecycle metadata now exists and the remaining value
-   is validation/scoping rather than a missing harness primitive.
+1. **Expose run inspection in the CLI inner loop** ([#3297](https://github.com/micro/go-micro/issues/3297)) — the canon promises scaffold → run → chat → inspect → deploy, and recent work added run timelines, trace correlation, and a `micro runs` foothold; with conformance scheduling, failure/backoff hardening, the deploy contract, and the support reference now shipped, this is the highest-value remaining DX seam in the Now/ongoing roadmap until local agent/flow activity is documented and CI-tested as an actionable inspect step.
+2. **Add durable agent run checkpoint and resume** ([#3306](https://github.com/micro/go-micro/issues/3306)) — once the remaining inspection seam is closed, the highest-value Next-roadmap item is making agent loops resumable like flows so long-running work can survive restarts without unsafe replay or hidden state loss.
 
 _Seeded by Claude Code from the roadmap + open issues; thereafter maintained by the
 architecture-review pass._

internal/harness/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,17 @@
+# 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.

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/SECURITY_MIGRATION.md

@@ -2,58 +2,64 @@
 
 ## Overview
 
-This document provides guidance for migrating to secure TLS certificate verification in go-micro v5.
+Go Micro v6 verifies TLS certificates by default. This guide is for teams
+upgrading from v5, where TLS verification was disabled by default for backward
+compatibility.
 
-## Current Status (v5)
+## Current Status (v6)
 
-**Default Behavior**: TLS certificate verification is **disabled** by default (`InsecureSkipVerify: true`)
+**Default Behavior**: TLS certificate verification is **enabled** by default
+(`InsecureSkipVerify: false`).
 
-**Reason**: Backward compatibility with existing deployments to avoid breaking production systems during routine upgrades.
+**What changed from v5**: v5 allowed `MICRO_TLS_SECURE=true` to opt into
+certificate verification. In v6, secure verification is the default and
+`MICRO_TLS_SECURE` is no longer used.
 
-**Security Risk**: The default behavior is vulnerable to man-in-the-middle (MITM) attacks.
+**Development escape hatch**: for local self-signed certificates only, set
+`MICRO_TLS_INSECURE=true` or provide an explicit insecure TLS config.
 
-## Migration Path
+## Migration Path from v5
 
-### Option 1: Enable Secure Mode (RECOMMENDED)
+### 1. Remove the old opt-in flag
 
-Set the environment variable to enable certificate verification:
+Delete any use of the v5-only environment variable:
 
 ```bash
-export MICRO_TLS_SECURE=true
+unset MICRO_TLS_SECURE
 ```
 
-This enables proper TLS certificate verification while maintaining compatibility with v5.
+No replacement is required for production: verification is already on in v6.
 
-### Option 2: Use SecureConfig Directly
+### 2. Use the default secure config
 
-In your code, explicitly use the secure configuration:
+Most services need no TLS-specific code. If you configure TLS explicitly, use a standard `crypto/tls` config with verification enabled:
 
 ```go
 import (
+    "crypto/tls"
     "go-micro.dev/v6/broker"
-    mls "go-micro.dev/v6/util/tls"
 )
 
-// Create broker with secure TLS config
+// Create broker with certificate verification enabled.
 b := broker.NewHttpBroker(
-    broker.TLSConfig(mls.SecureConfig()),
+    broker.TLSConfig(&tls.Config{MinVersion: tls.VersionTLS12}),
 )
 ```
 
-### Option 3: Provide Custom TLS Configuration
+### 3. Provide a custom trust root when needed
 
-For fine-grained control, provide your own TLS configuration:
+For private CAs, provide your own TLS configuration:
 
 ```go
 import (
     "crypto/tls"
     "crypto/x509"
     "go-micro.dev/v6/broker"
-    "io/ioutil"
+    "os"
 )
 
 // Load CA certificates
-caCert, err := ioutil.ReadFile("/path/to/ca-cert.pem")
+caCert, err := os.ReadFile("/path/to/ca-cert.pem")
 if err != nil {
     log.Fatal(err)
 }
@@ -73,52 +79,61 @@ b := broker.NewHttpBroker(
 )
 ```
 
+### 4. Use insecure mode only for local development
+
+If a development environment still uses self-signed certificates that are not in
+your trust store, opt out explicitly:
+
+```bash
+export MICRO_TLS_INSECURE=true
+```
+
+or in code:
+
+```go
+broker.TLSConfig(&tls.Config{InsecureSkipVerify: true, MinVersion: tls.VersionTLS12})
+```
+
+Do not use insecure mode in production.
+
 ## Production Deployment Strategy
 
 ### Rolling Upgrade Considerations
 
-The current implementation maintains backward compatibility, allowing safe rolling upgrades:
+The default changed at the v6 major-version boundary. Before rolling v6 into a
+fleet that uses TLS, verify that:
 
-1. **Mixed Version Deployments**: v5 instances can communicate regardless of TLS security settings
-2. **No Immediate Breaking Changes**: Systems continue working with existing behavior
-3. **Gradual Migration**: Enable security incrementally across your infrastructure
+1. All services present certificates trusted by their peers.
+2. Private or self-signed CAs are installed consistently on every host.
+3. Certificates include the DNS names or IP subject alternative names used by
+   clients.
+4. Any deliberate development-only insecure settings are excluded from
+   production manifests.
 
 ### Recommended Approach
 
-1. **Test in Staging**:
-   ```bash
-   # In staging environment
-   export MICRO_TLS_SECURE=true
-   ```
-   
-2. **Deploy with Feature Flag**: Use environment-based configuration for gradual rollout
-
-3. **Monitor for Issues**: Watch for TLS handshake failures or certificate validation errors
-
-4. **Full Production Rollout**: Once validated, enable across all services
+1. **Test in Staging** with the same certificate chain and service names used in
+   production.
+2. **Remove v5 flags** such as `MICRO_TLS_SECURE`; they no longer control v6.
+3. **Monitor for Issues**: watch for TLS handshake failures or certificate
+   validation errors.
+4. **Use explicit insecure mode only in dev** when a short-lived environment
+   cannot yet provide trusted certificates.
 
 ### Multi-Host/Multi-Process Considerations
 
-**Certificate Trust**: When enabling secure mode, ensure:
+**Certificate Trust**: With secure mode as the default, ensure:
 
-1. All hosts trust the same root CAs
-2. Self-signed certificates are properly distributed if used
-3. Certificate validity periods are monitored
-4. Certificate chains are complete
+1. All hosts trust the same root CAs.
+2. Self-signed certificates are properly distributed if used.
+3. Certificate validity periods are monitored.
+4. Certificate chains are complete.
 
 **Service Mesh Alternative**: Consider using a service mesh (Istio, Linkerd, etc.) for:
 - Automatic mTLS between services
 - Certificate management and rotation
 - No application code changes required
 
-## Future Changes (v6)
-
-In go-micro v6, the default will change to **secure by default**:
-
-- `InsecureSkipVerify: false` (certificate verification enabled)
-- Breaking change requiring major version bump
-- Migration completed before v6 release avoids disruption
-
 ## Testing Your Migration
 
 ### Verify Secure Mode is Active
@@ -127,14 +142,12 @@ In go-micro v6, the default will change to **secure by default**:
 package main
 
 import (
+    "crypto/tls"
     "fmt"
-    mls "go-micro.dev/v6/util/tls"
-    "os"
 )
 
 func main() {
-    os.Setenv("MICRO_TLS_SECURE", "true")
-    config := mls.Config()
+    config := &tls.Config{MinVersion: tls.VersionTLS12}
     fmt.Printf("InsecureSkipVerify: %v (should be false)\n", config.InsecureSkipVerify)
 }
 ```
@@ -155,7 +168,7 @@ Create a test service and verify it:
 **Solution**:
 1. Add the CA certificate to the trusted root CAs
 2. Use a properly signed certificate
-3. For development only: Use `InsecureConfig()` explicitly
+3. For development only: use `MICRO_TLS_INSECURE=true` or an explicit insecure TLS config
 
 ### Issue: "x509: certificate has expired"
 
@@ -166,24 +179,17 @@ Create a test service and verify it:
 2. Implement certificate rotation
 3. Monitor certificate expiry dates
 
-### Issue: Services can't communicate after enabling secure mode
+### Issue: Services can't communicate after upgrading to v6
 
-**Cause**: Mixed certificate authorities or missing certificates
+**Cause**: Certificates that v5 accepted by default are now verified.
 
 **Solution**:
-1. Ensure all services use certificates from the same CA
+1. Ensure all services use certificates from a trusted CA
 2. Distribute CA certificates to all nodes
 3. Verify certificate SANs match service addresses
+4. Use insecure mode only as a temporary local-development workaround
 
 ## Questions?
 
-For issues or questions about TLS security migration, please:
-- Open an issue on GitHub
-- Check the documentation at https://go-micro.dev/docs/
-- Review the security guidelines
-
-## Security Resources
-
-- [OWASP TLS Cheat Sheet](https://cheatsheetseries.owasp.org/cheatsheets/Transport_Layer_Protection_Cheat_Sheet.html)
-- [Go TLS Documentation](https://pkg.go.dev/crypto/tls)
-- [Certificate Best Practices](https://www.ssl.com/guide/ssl-best-practices/)
+For issues or questions about TLS security migration, open an issue on GitHub or
+check the documentation at https://go-micro.dev/docs/.

internal/website/docs/TLS_SECURITY_UPDATE.md

@@ -2,66 +2,36 @@
 
 ## What Changed
 
-The TLS configuration in go-micro now includes a security deprecation warning.
+Go Micro v6 verifies TLS certificates by default. This completes the v5 security
+migration where verification was opt-in.
 
-## Current Behavior (v5.x)
+## Current Behavior (v6.x)
 
-**Default**: TLS certificate verification is **disabled** for backward compatibility
-- This maintains existing behavior to avoid breaking production deployments
-- A deprecation warning is logged once per process startup
+**Default**: TLS certificate verification is **enabled**.
+- `MICRO_TLS_SECURE` was a v5 opt-in flag and is no longer used.
+- For local development with untrusted self-signed certificates, opt out
+  explicitly with `MICRO_TLS_INSECURE=true` or an explicit insecure TLS config.
 
-**Why**: Changing the default to secure would be a **breaking change** that could disrupt:
-- Production systems during routine upgrades
-- Distributed systems with mixed versions
-- Services using self-signed certificates
+## Production Recommendation
 
-## How to Enable Security (Recommended)
-
-### Option 1: Environment Variable
-
-```bash
-export MICRO_TLS_SECURE=true
-```
-
-### Option 2: Use SecureConfig
-
-```go
-import (
-    "go-micro.dev/v6/broker"
-    mls "go-micro.dev/v6/util/tls"
-)
-
-broker := broker.NewHttpBroker(
-    broker.TLSConfig(mls.SecureConfig()),
-)
-```
+For production deployments:
+1. Use CA-signed certificates or distribute your private CA to every host.
+2. Remove old `MICRO_TLS_SECURE` settings from v5-era manifests.
+3. Do not set `MICRO_TLS_INSECURE=true` in production.
+4. Consider service mesh mTLS (Istio, Linkerd) if certificate lifecycle should be
+   managed outside the application.
 
 ## Migration Timeline
 
-- **v5.x (Current)**: Insecure by default, opt-in security via `MICRO_TLS_SECURE=true`
-- **v6.x (Future)**: Secure by default (breaking change with major version bump)
-
-## Why This Approach?
-
-This addresses the concerns raised about:
-
-1. **Major version requirements**: No breaking change in v5, deferred to v6
-2. **Cross-host compatibility**: All hosts use same default behavior
-3. **Production safety**: Existing deployments continue working during upgrades
-4. **Migration path**: Clear opt-in path with documentation
+- **v5.x**: Insecure by default, opt-in security via `MICRO_TLS_SECURE=true`.
+- **v6.x current**: Secure by default; use `MICRO_TLS_INSECURE=true` only for an
+  explicit development opt-out.
 
 ## Documentation
 
-See [SECURITY_MIGRATION.md](SECURITY_MIGRATION.html) for detailed migration guide.
-
-## Security Recommendation
-
-For production deployments:
-1. Test with `MICRO_TLS_SECURE=true` in staging
-2. Use proper CA-signed certificates
-3. Consider service mesh (Istio, Linkerd) for automatic mTLS
-4. Plan migration before v6 release
+See [SECURITY_MIGRATION.md](SECURITY_MIGRATION.html) for the detailed migration
+guide.
 
 ## Questions?
 
-Open an issue on GitHub or check the documentation at https://go-micro.dev/docs/
+Open an issue on GitHub or check the documentation at https://go-micro.dev/docs/.

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

@@ -66,7 +66,7 @@ A card looks like:
   "url": "https://agents.example.com/agents/task-mgr",
   "version": "1.0.0",
   "protocolVersion": "0.3.0",
-  "capabilities": { "streaming": false, "pushNotifications": false },
+  "capabilities": { "streaming": true, "pushNotifications": true },
   "defaultInputModes": ["text/plain"],
   "defaultOutputModes": ["text/plain"],
   "skills": [{ "id": "chat", "name": "Chat", "tags": ["task", "project"] }]
@@ -100,7 +100,40 @@ curl -s https://agents.example.com/agents/task-mgr \
 }
 ```
 
-Retrieve a task later with `tasks/get` (`params: { "id": "…" }`).
+Retrieve a task later with `tasks/get` (`params: { "id": "…" }`). To continue
+the same piece of work, send another `message/send` with the previous `taskId`
+and `contextId`. The gateway preserves the task id, context id, and prior
+history, then appends the new user turn and agent reply. That makes a remote
+A2A task fit the Go Micro lifecycle: services are still invoked through the
+agent's normal tools, the agent keeps task context across turns, and a workflow
+can poll one task id as the conversation progresses.
+
+## Push notifications
+
+Operators can register a task callback with
+`tasks/pushNotificationConfig/set`:
+
+```bash
+curl -s https://agents.example.com/agents/task-mgr \
+  -H 'content-type: application/json' \
+  -d '{
+    "jsonrpc": "2.0", "id": 2,
+    "method": "tasks/pushNotificationConfig/set",
+    "params": {
+      "id": "task-id",
+      "pushNotificationConfig": {
+        "url": "https://workflow.example.com/a2a/tasks",
+        "token": "optional-bearer-token"
+      }
+    }
+  }'
+```
+
+The gateway stores one callback per retained task and POSTs the latest task
+snapshot to that URL whenever the task changes. Delivery is best effort: failures
+do not fail the agent turn, and there is no retry queue in the in-memory gateway.
+Use `tasks/get` as the source of truth after a missed callback or receiver
+outage. If a token is configured, it is sent as `Authorization: Bearer <token>`.
 
 ## Calling out to other agents
 
@@ -132,22 +165,18 @@ yet terminal, it polls `tasks/get` until it completes.
 
 ## Scope
 
-This is the JSON-RPC binding for completed-task execution:
+This is the JSON-RPC binding for task execution:
 
 - **`message/send`** runs the agent and returns a completed `Task`.
 - **`message/stream`** streams the completed `Task` as an SSE `data:` event, giving A2A clients a streaming-compatible path while the underlying agent call remains synchronous.
 - **`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.
-- Push notifications.
-
-These are the natural follow-ups; the completed-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/x402-payments.md

@@ -100,7 +100,22 @@ c := &x402.Client{
 resp, err := c.Do(req) // a 402 is paid and retried; over-budget calls error instead
 ```
 
-`Payer` is an interface (`Pay(ctx, Requirements) (payment string, error)`) — the consumer counterpart to `Facilitator`. The budget accumulates across calls, so a long-running agent can be handed a fixed allowance for a task. (The agent-level `AgentMaxSpend` option, wiring this into the agent loop next to `MaxSteps`/`ApproveTool`, is the next step.)
+`Payer` is an interface (`Pay(ctx, Requirements) (payment string, error)`) — the consumer counterpart to `Facilitator`. The budget accumulates across calls, so a long-running agent can be handed a fixed allowance for a task. Budget is reserved before payment is created, which means parallel paid calls cannot race past the cap; if payment creation or verification fails, the reservation is released. (The agent-level `AgentMaxSpend` option, wiring this into the agent loop next to `MaxSteps`/`ApproveTool`, is the next step.)
+
+### Live facilitator conformance
+
+The regular test suite uses in-process facilitators and does not need network credentials. To smoke-test a hosted facilitator, run the opt-in live conformance test with a real payment payload and matching requirements:
+
+```sh
+GO_MICRO_X402_LIVE_FACILITATOR_URL=https://facilitator.example \
+GO_MICRO_X402_LIVE_PAYMENT='...' \
+GO_MICRO_X402_LIVE_PAY_TO=0xYourAddress \
+GO_MICRO_X402_LIVE_NETWORK=base \
+GO_MICRO_X402_LIVE_AMOUNT=1 \
+go test ./wrapper/x402 -run TestLiveFacilitatorConformance -count=1
+```
+
+Leave those variables unset in normal CI; the live test skips unless the facilitator URL, payment payload, and pay-to address are all provided.
 
 ## Notes
 

internal/website/docs/roadmap.md

@@ -13,7 +13,7 @@ The foundation is in place:
 - **Services** — register, discover, RPC, events; every endpoint is automatically an MCP tool.
 - **Agents** — a model with memory and tools that manages services, with `plan`, `delegate`, and guardrails (`MaxSteps`, `LoopLimit`, `ApproveTool`) built in, plus tool-execution middleware (`WrapTool`) and run metadata.
 - **Flows** — durable, event-driven workflows: ordered steps that checkpoint and resume after a crash.
-- **Interop** — the MCP gateway (services as tools) and the A2A gateway (agents as agents, both directions), both generated from the registry; x402 for paid tools.
+- **Interop** — the MCP gateway (services as tools) and the A2A gateway (agents as agents, both directions, including A2A streaming, push notifications, and multi-turn continuation), both generated from the registry; x402 for paid tools.
 - **Secure by default** — TLS verification on, state scoped per component.
 
 ## Principles
@@ -37,15 +37,14 @@ The priority is that what exists works everywhere, under real conditions.
 ## Next — agentic depth
 
 - **Durable agent loop.** Flows resume; the agent's own loop does not yet. Reuse `Checkpoint` so a long-running agent survives a restart and continues.
-- **Streaming.** `ai.Stream` is stubbed across providers; real chat and long-task UX need it, end to end through A2A `message/stream`.
+- **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.
-- **x402** — conformance against a live facilitator; paid remote tools as agent tools with spend caps.
-- **A2A** — streaming, push notifications, multi-turn tasks.
+- **A2A** — richer live-stream reconnection (`tasks/resubscribe`) and `input-required` handoffs.
 
 ## Developer experience (ongoing)
 

micro.go

@@ -110,18 +110,36 @@ func AgentApproveTool(fn ApproveFunc) AgentOption { return agent.ApproveTool(fn)
 // Memory is an agent's pluggable conversation memory.
 type Memory = agent.Memory
 
+// MemoryRecall is implemented by memory backends that retrieve prior context.
+type MemoryRecall = agent.MemoryRecall
+
 // ToolFunc handles a custom agent tool call.
 type ToolFunc = agent.ToolFunc
 
 // NewMemory returns the default store-backed agent memory.
 func NewMemory(s store.Store, key string, limit int) Memory { return agent.NewMemory(s, key, limit) }
 
+// NewCompactingMemory returns store-backed memory with deterministic
+// summarization and retrieval controls.
+func NewCompactingMemory(s store.Store, key string, maxMessages, keepRecent int) Memory {
+	return agent.NewCompactingMemory(s, key, maxMessages, keepRecent)
+}
+
 // NewInMemory returns non-persistent agent memory.
 func NewInMemory(limit int) Memory { return agent.NewInMemory(limit) }
 
 // AgentMemory sets the agent's conversation memory (default: store-backed).
 func AgentMemory(m Memory) AgentOption { return agent.WithMemory(m) }
 
+// AgentCompactMemory enables deterministic default-memory compaction and
+// retrieval for long-running agents.
+func AgentCompactMemory(maxMessages, keepRecent int) AgentOption {
+	return agent.CompactMemory(maxMessages, keepRecent)
+}
+
+// AgentMemoryRecallLimit bounds recalled archived turns injected per Ask.
+func AgentMemoryRecallLimit(n int) AgentOption { return agent.MemoryRecallLimit(n) }
+
 // AgentTool registers a custom tool the agent can call, beyond its services.
 func AgentTool(name, description string, properties map[string]any, handler ToolFunc) AgentOption {
 	return agent.WithTool(name, description, properties, handler)

wrapper/x402/client.go

@@ -83,7 +83,9 @@ func (c *Client) Do(req *http.Request) (*http.Response, error) {
 	reqd := ch.Accepts[0]
 	amount, _ := strconv.ParseInt(reqd.MaxAmountRequired, 10, 64)
 
-	// Spend cap: refuse before paying if this would exceed the budget.
+	// Spend cap: reserve before paying so concurrent calls cannot all pass
+	// the check and overspend the caller's allowance. Roll the reservation
+	// back if payment construction, replay, or verification fails.
 	c.mu.Lock()
 	if c.Budget > 0 && c.spent+amount > c.Budget {
 		spent := c.spent
@@ -91,13 +93,26 @@ func (c *Client) Do(req *http.Request) (*http.Response, error) {
 		return nil, fmt.Errorf("x402: paying %d for %s would exceed budget (spent %d of %d)",
 			amount, reqd.Resource, spent, c.Budget)
 	}
+	c.spent += amount
 	c.mu.Unlock()
+	reserved := true
+	rollback := func() {
+		if !reserved {
+			return
+		}
+		c.mu.Lock()
+		c.spent -= amount
+		c.mu.Unlock()
+		reserved = false
+	}
 
 	if c.Payer == nil {
+		rollback()
 		return nil, fmt.Errorf("x402: payment required for %s but no Payer configured", reqd.Resource)
 	}
 	payment, err := c.Payer.Pay(req.Context(), reqd)
 	if err != nil {
+		rollback()
 		return nil, fmt.Errorf("x402: pay: %w", err)
 	}
 
@@ -108,6 +123,7 @@ func (c *Client) Do(req *http.Request) (*http.Response, error) {
 	}
 	retry, err := http.NewRequestWithContext(req.Context(), req.Method, req.URL.String(), rbody)
 	if err != nil {
+		rollback()
 		return nil, err
 	}
 	for k, v := range req.Header {
@@ -117,14 +133,14 @@ func (c *Client) Do(req *http.Request) (*http.Response, error) {
 
 	resp2, err := c.httpClient().Do(retry)
 	if err != nil {
+		rollback()
 		return nil, err
 	}
 	if resp2.StatusCode == http.StatusPaymentRequired {
+		rollback()
 		return resp2, fmt.Errorf("x402: payment for %s was rejected", reqd.Resource)
 	}
 
-	c.mu.Lock()
-	c.spent += amount
-	c.mu.Unlock()
+	reserved = false
 	return resp2, nil
 }

wrapper/x402/client_test.go

@@ -118,3 +118,62 @@ func TestClientFreeEndpoint(t *testing.T) {
 		t.Errorf("free call should spend nothing, got %d", c.Spent())
 	}
 }
+
+// Concurrent callers reserve budget before paying, so a spend cap cannot be
+// overshot by parallel tool calls that all observe the same pre-spend balance.
+func TestClientBudgetReservedConcurrently(t *testing.T) {
+	srv := paidServer("10000")
+	defer srv.Close()
+
+	c := &Client{Payer: &mockPayer{}, Budget: 10000}
+	errCh := make(chan error, 2)
+	for i := 0; i < 2; i++ {
+		go func() {
+			req, _ := http.NewRequest(http.MethodGet, srv.URL, nil)
+			resp, err := c.Do(req)
+			if resp != nil {
+				resp.Body.Close()
+			}
+			errCh <- err
+		}()
+	}
+
+	var paid, refused int
+	for i := 0; i < 2; i++ {
+		if err := <-errCh; err != nil {
+			refused++
+		} else {
+			paid++
+		}
+	}
+	if paid != 1 || refused != 1 {
+		t.Fatalf("paid=%d refused=%d, want one paid and one refused", paid, refused)
+	}
+	if c.Spent() != 10000 {
+		t.Errorf("spent = %d, want 10000", c.Spent())
+	}
+}
+
+// A reserved budget slot is released when payment cannot be produced, so a
+// transient payer failure does not permanently consume an agent's allowance.
+func TestClientBudgetReservationRollsBackOnPayError(t *testing.T) {
+	srv := paidServer("10000")
+	defer srv.Close()
+
+	c := &Client{Payer: payerFunc(func(context.Context, Requirements) (string, error) {
+		return "", context.Canceled
+	}), Budget: 10000}
+	req, _ := http.NewRequest(http.MethodGet, srv.URL, nil)
+	if _, err := c.Do(req); err == nil {
+		t.Fatal("expected pay error")
+	}
+	if c.Spent() != 0 {
+		t.Errorf("failed payment should roll back spend, got %d", c.Spent())
+	}
+}
+
+type payerFunc func(context.Context, Requirements) (string, error)
+
+func (f payerFunc) Pay(ctx context.Context, req Requirements) (string, error) {
+	return f(ctx, req)
+}

wrapper/x402/live_facilitator_test.go

@@ -0,0 +1,52 @@
+package x402
+
+import (
+	"context"
+	"os"
+	"testing"
+	"time"
+)
+
+// TestLiveFacilitatorConformance is an opt-in smoke test for hosted x402
+// facilitators. It is skipped by default so local and CI runs never need live
+// credentials, but operators can run it against Coinbase, Alchemy, or a
+// self-hosted facilitator by providing a real payment payload and matching
+// requirements.
+func TestLiveFacilitatorConformance(t *testing.T) {
+	url := os.Getenv("GO_MICRO_X402_LIVE_FACILITATOR_URL")
+	payment := os.Getenv("GO_MICRO_X402_LIVE_PAYMENT")
+	payTo := os.Getenv("GO_MICRO_X402_LIVE_PAY_TO")
+	if url == "" || payment == "" || payTo == "" {
+		t.Skip("set GO_MICRO_X402_LIVE_FACILITATOR_URL, GO_MICRO_X402_LIVE_PAYMENT, and GO_MICRO_X402_LIVE_PAY_TO to run live x402 facilitator conformance")
+	}
+
+	network := getenv("GO_MICRO_X402_LIVE_NETWORK", "base")
+	amount := getenv("GO_MICRO_X402_LIVE_AMOUNT", "1")
+	asset := os.Getenv("GO_MICRO_X402_LIVE_ASSET")
+	resource := getenv("GO_MICRO_X402_LIVE_RESOURCE", "go-micro-x402-live-conformance")
+
+	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
+	defer cancel()
+	res, err := (&HTTPFacilitator{URL: url}).Verify(ctx, payment, Requirements{
+		Scheme:            "exact",
+		Network:           network,
+		MaxAmountRequired: amount,
+		Resource:          resource,
+		PayTo:             payTo,
+		Asset:             asset,
+		MaxTimeoutSeconds: 60,
+	})
+	if err != nil {
+		t.Fatalf("live facilitator verify: %v", err)
+	}
+	if !res.Valid {
+		t.Fatalf("live facilitator rejected payment: %s", res.Reason)
+	}
+}
+
+func getenv(key, fallback string) string {
+	if v := os.Getenv(key); v != "" {
+		return v
+	}
+	return fallback
+}