Add Kubernetes CRD foundation

Two spend-safety fixes from the gap audit (#4814):

- Client.Do refused a 402 only on the budget check, but parsed
  maxAmountRequired with a swallowed error, so a non-decimal, overflowing
  or negative amount became 0 and passed the cap trivially while Payer.Pay
  still signed against the string. Now reject any amount that is not a
  positive integer before signing.

- Require settled only when the facilitator implemented Settler; a
  verify-only facilitator served the resource while no funds moved. Add
  Config.RequireSettlement to fail closed in that case.

Tests cover invalid/negative/overflow amounts and the verify-only
fail-closed path.


Claude-Session: https://claude.ai/code/session_01CmdEY7pYmV5zzwCjNJ4ykL

Co-authored-by: Claude <noreply@anthropic.com>

.github/loop/PRIORITIES.md

@@ -1,27 +1,41 @@
 # Priorities
 
-The ranked work queue for the autonomous improvement loop. The
-**architecture-review** pass (the *architect*) owns this file: each run it turns
-the [roadmap](../../ROADMAP.md) plus an internal scan (gaps in the
-services → agents → workflows lifecycle, API coherence, drift, tech debt, test and
-DX friction) into a single ordered list — highest-value first — and links each
-item to a tracking issue. The hourly **continuous-improvement** pass works the
-**top item whose issue is still open**. So the architect decides *what*, and the
-increment loop *builds* it.
+The ranked work queue for the autonomous improvement loop. The **planner** owns
+this file: each run it turns the [roadmap](../../ROADMAP.md) plus an internal scan
+into a single ordered list — highest-value first — each item linked to a tracking
+issue. The **builder** works the top item whose issue is still open. So the
+planner decides *what*, the builder *builds* it.
 
-**Reading / editing.** An item is done when its linked issue closes (the increment
-that builds it adds `Closes #<issue>`). Roadmap phase (Now → Next → Later) is the
-primary ordering; internal findings are interleaved by value, not kept in a
-separate list. The human can reorder this list — or the issues — at any time to
-redirect the loop; direction always wins.
+**Bias to capability, not busy-work.** The top of this queue is net-new capability
+from the roadmap's *Now/Next* items. Hardening/conformance/DX polish is background
+work (roadmap *Ongoing*) — kept low here and capped, never allowed to crowd out
+capability. If an area has had several increments with no user-visible gain, it is done
+for now; rank real-headroom capability instead.
 
-**Off-limits to the loop** (the architect proposes these as notes, never as queue
-items the loop can auto-merge): brand/positioning copy, breaking public-API
-changes, architectural rewrites. Those go to the human.
+**Reading / editing.** An item is done when its linked issue closes (the PR that
+builds it adds `Closes #<issue>`). The human can reorder this list or the issues at
+any time — direction always wins.
+
+**Off-limits to the loop** (planner proposes as notes, never auto-merged queue
+items): brand/positioning copy, breaking public-API changes, architectural
+rewrites.
 
 ## Work queue (ranked)
 
-1. **Add end-to-end streaming chat conformance** ([#4738](https://github.com/micro/go-micro/issues/4738)) — #4736 closed the pgx/v5 security-upkeep item, and there are no open codex PRs in flight. The next highest-value user-visible gap is streaming chat that works coherently across the agent/CLI/A2A path, starting with a no-secret harness and extending to keyed provider checks without another provider-specific malformed-output workaround.
+### Capability — the headline (roadmap: Now / Next)
 
-_Seeded by Claude Code from the roadmap + open issues; thereafter maintained by the
-architecture-review pass._
+1. **x402 buyer safety hard stop** ([#4814](https://github.com/micro/go-micro/issues/4814)) — close the budget-cap bypass and require a real settler for paid configs before autonomous paid-tool use becomes the next thing users copy into real agents.
+2. **Kubernetes operator + CRDs foundation** ([#4797](https://github.com/micro/go-micro/issues/4797)) — add the first opt-in `Agent`, `Service`, and `Flow` resource foundation so the services → agents → workflows lifecycle has a native deployment path for Kubernetes users.
+3. **A2A external-client conformance** ([#4815](https://github.com/micro/go-micro/issues/4815)) — make the gateway easier for non-go-micro agents to discover and stream from by serving the well-known agent card path and spec SSE events.
+4. **MCP stdio/ws result conformance** ([#4813](https://github.com/micro/go-micro/issues/4813)) — return JSON tool results with explicit `isError` semantics across transports, backed by a stdio round-trip test.
+
+### In flight — do not re-queue
+
+- **gRPC-reflection MCP lint follow-up** ([#4824](https://github.com/micro/go-micro/issues/4824), [PR #4826](https://github.com/micro/go-micro/pull/4826)) — fixes the lint fallout from the merged reflected-gRPC MCP work.
+
+### Background — hardening & DX (roadmap: Ongoing; capped)
+
+_Background hardening is intentionally empty right now. Recent work covered first-agent
+wayfinding, plan/delegate recovery, provider fallback repair, streaming, memory
+compaction, retry controls, and provider-failure inspection. Further churn in those
+areas should be marked `needs-human` unless it unlocks a clear user-visible capability._

CHANGELOG.md

@@ -18,15 +18,49 @@ below is kept current between tags and rolled into the next version when it ship
 ## [Unreleased]
 
 ### Added
+- **Gemini streaming support** — the Gemini provider now supports streaming model responses. (`ai/gemini/`)
+- **Model retry jitter controls** — model retry behavior can now use jitter controls to reduce synchronized retry bursts. (`ai/`, `agent/`)
+- **Compacted memory summaries** — agent memory now exposes compacted run summaries for easier inspection and recovery. (`agent/`)
+- **CLI input resume for agent runs** — the CLI can resume agent runs that require additional user input. (`cmd/micro/`, `agent/`)
+
+### Changed
+- **Remote agent chat streaming** — `micro chat` now streams replies from remote agents instead of waiting for the full response. (`cmd/micro/`, `agent/`)
+
+### Fixed
+- **Provider failure inspection metadata** — provider failures recorded during agent runs now retain classification metadata for inspection. (`agent/`, `ai/`)
+
+### Security
+- **x402 spend-cap hardening** — the paying `Client` now refuses a 402 whose `maxAmountRequired` is not a positive integer (a swallowed parse error or negative amount previously bypassed the budget cap), and a new `Config.RequireSettlement` fails closed when a paid request is served by a verify-only facilitator that never captures funds. (`wrapper/x402/`)
+
+---
+
+## [6.7.0] - July 2026
+
+### Added
+- **A2A streaming conformance harness** — A2A streaming behavior is now covered by focused conformance checks. (`gateway/a2a/`, `internal/harness/`)
+- **Agent x402 spend budget guardrail** — agents now have spend budget guardrails for x402-paid tool calls. (`agent/`, `gateway/`)
 - **First-agent chat/inspect fixture** — the maintained first-agent CLI fixture now covers chat and inspect boundaries together. (`internal/harness/`, `cmd/micro/`)
 - **Zero-to-hero inspect transcript check** — the 0→hero harness now verifies the inspect transcript path stays visible in the lifecycle walkthrough. (`internal/harness/zero-to-hero-ci/`, `internal/website/docs/`)
 
 ### Changed
+- **Agent stream run context propagation** — agent streams now preserve run context through streaming paths for more complete tracing and inspection. (`agent/`)
+- **Postgres store pgx v5 migration** — the Postgres store now uses pgx v5. (`store/postgres/`, `go.mod`)
 - **Plan-delegate plan persistence** — plan/delegate runs now persist plan state more defensively across harness scenarios. (`agent/`, `internal/harness/`)
 
 ### Fixed
+- **Nested tool-call markup rejection** — agent argument parsing now rejects nested tool-call markup instead of accepting ambiguous tool input. (`agent/`)
+- **Retry cancellation during backoff** — retry backoff now respects cancellation more reliably. (`agent/`, `ai/`)
+- **Plan-delegate mock recovery regression gate** — the harness now catches plan/delegate mock recovery regressions before they ship. (`internal/harness/`, `agent/`)
+- **First-agent fixture registration wait** — first-agent fixture registration is less race-prone during harness runs. (`internal/harness/`)
+- **Memory stream Nack ordering** — memory stream Nack handling now preserves ordering more reliably. (`broker/memory/`)
 - **Zero-to-hero fixture output race** — 0→hero fixture output is less race-prone during harness runs. (`internal/harness/zero-to-hero-ci/`)
 
+### Documentation
+- **First-agent quickcheck wayfinding** — public docs now keep the quickcheck path discoverable from the first-agent route. (`README.md`, `internal/website/docs/`)
+- **Ordered 0→hero transcript** — docs and harness checks now keep the 0→hero transcript order explicit. (`internal/website/docs/`, `internal/harness/`)
+- **First-agent debug breadcrumbs** — docs now surface the first-agent debug smoke path more clearly. (`internal/website/docs/`)
+- **README badge cleanup** — the README no longer shows the Go Report Card badge. (`README.md`)
+
 ---
 
 ## [6.6.0] - July 2026

ROADMAP.md

@@ -32,44 +32,55 @@ default.
    and history, end to end.
 5. Battle-tested: works across every provider, fails safely, observable.
 
-## Now — hardening
+The forward work is **net-new capability**, not more hardening. Maintenance
+(conformance, resilience, DX polish) continues in the background (see *Ongoing*
+below) — but it is not the roadmap. This capability work is.
 
-- **Cross-provider conformance** — the same agent scenario across all seven
-  providers, gated on keys, on a schedule.
-- **Failure & resilience** — timeouts, rate limits, cancellation, deadline/context
-  propagation, retry/backoff.
-- **Getting-started contract** — define and CI-verify the 0→1 and 0→hero flows.
+## Now — capability
 
-## Shipped agent depth
+- **Agents that pay (x402 buyer in the runtime).** The seller side ships (paid
+  tools via the `wrapper/x402` middleware) and the buyer `x402.Client` (a
+  budget-capped `Payer` that turns a `402` into pay-and-retry) exists — but an
+  agent can't yet *autonomously* pay for a paid tool. Wire the buyer into the
+  agent tool loop: a budget-capped `AgentPayer` so an agent that hits a
+  payment-required tool settles it within budget and retries, with the spend
+  gated (like `ApproveTool`) and observable in `RunInfo`/traces. This makes
+  go-micro a runtime for **autonomous agent commerce**. *(flagship — decomposed
+  into issues in the loop queue)*
+- **AP2 mandate foundation** ([#3552](https://github.com/micro/go-micro/issues/3552))
+  — verifiable payment **mandates** (a Checkout Mandate and a Payment Mandate),
+  signed and attached over A2A, with the Payment Mandate naming an x402 rail. The
+  authorization/audit layer above A2A + x402 that positions go-micro early in the
+  emerging agent-payments standard (Google's AP2, standardized via FIDO).
+  Additive and opt-in.
 
-- **Durable agent loop** — opt-in `Checkpoint` support lets agent `Ask` and
-  streaming runs persist, list pending work, and resume without replaying completed
-  tool calls. Human-input pauses resume through explicit input helpers.
-- **Agent observability** — agent `RunInfo` now feeds OpenTelemetry spans/events
-  across runs, model turns, tool calls, retries, delegation lineage, and resume
-  checkpoints.
+## Next — reach & deployment
 
-## Next — agentic depth
+- **gRPC-reflection MCP** — derive MCP tools from *any* gRPC service via server
+  reflection, not just go-micro-native handlers. Point the gateway at an external
+  gRPC service and its methods become agent tools — a large jump in what an agent
+  can operate.
+- **Kubernetes operator + CRDs** — `Agent`, `Service`, and `Flow` as first-class
+  Kubernetes resources; an operator reconciles them into Deployments wired to the
+  registry. The production deployment story for teams already on K8s.
 
-- **Streaming** — broaden provider-backed `ai.Stream` coverage and keep chat/A2A streaming end to end.
-- **Resume operations polish** — keep improving CLI/docs breadcrumbs for finding
-  pending agent runs and deciding whether to call resume, resume-input, or stream
-  resume in production.
-- **Observability hardening** — keep span attributes and run inspection coherent
-  across agents, flows, and gateways as more providers and workflow paths are
-  exercised.
+## Later — exploratory
 
-## Later
+- **Runtime-fitness loop** — a persistently-running dogfood app (Mu) plus an
+  operator/canary loop role, so the autonomous loop evolves go-micro against
+  **real runtime signal** (latency, errors, cost) with canary + rollback — not
+  just green CI. The demand signal the loop is missing today.
+- **HTTP/3 transport**; richer A2A live-stream reconnection (`tasks/resubscribe`,
+  `input-required` handoffs); memory management (summarization, retrieval/RAG).
 
-- Memory management (summarization, retrieval/RAG); human-in-the-loop pause/resume;
-  richer A2A live-stream reconnection (`tasks/resubscribe`) and `input-required`
-  handoffs.
+## Ongoing — hardening & DX (background, not the headline)
 
-## Developer experience (ongoing)
-
-- A seamless CLI inner loop (scaffold → run → chat → inspect → deploy); UI
-  discipline (trim what isn't great); a maintained real-world example that doubles
-  as the 0→hero reference; docs kept in lockstep with the code.
+Continuous but **capped** so it never crowds out capability: cross-provider
+conformance, failure/resilience (timeouts, cancellation, retry/backoff), the
+0→1 and 0→hero getting-started contract, streaming/observability coherence, and a
+seamless CLI inner loop (scaffold → run → chat → inspect → deploy). Real, but
+maintenance — the loop should spend the majority of its cycles on the capability above,
+not here.
 
 ## How it's sustained
 

agent/agent.go

@@ -15,7 +15,9 @@ package agent
 import (
 	"context"
 	"encoding/json"
+	"errors"
 	"fmt"
+	"io"
 	"net/http"
 	"strings"
 	"sync"
@@ -81,6 +83,8 @@ type agentImpl struct {
 
 	// steps counts tool executions in the current Ask, for MaxSteps.
 	steps int
+	// spend counts reserved paid-tool spend in the current Ask, for MaxSpend.
+	spend int64
 	// calls counts identical tool calls (name+args) in the current Ask,
 	// for LoopLimit.
 	calls map[string]int
@@ -226,6 +230,9 @@ func (a *agentImpl) Ask(ctx context.Context, message string) (*Response, error)
 func (a *agentImpl) Stream(ctx context.Context, message string) (ai.Stream, error) {
 	a.mu.Lock()
 	defer a.mu.Unlock()
+	if err := ctx.Err(); err != nil {
+		return nil, err
+	}
 	if a.model == nil {
 		a.setup()
 	}
@@ -233,6 +240,12 @@ func (a *agentImpl) Stream(ctx context.Context, message string) (ai.Stream, erro
 	if err != nil {
 		return nil, fmt.Errorf("discover tools: %w", err)
 	}
+	runID := uuid.New().String()
+	ctx = ai.WithRunInfo(ctx, ai.RunInfo{
+		RunID:    runID,
+		ParentID: a.parentRunID,
+		Agent:    a.opts.Name,
+	})
 	messages := append([]ai.Message(nil), a.mem.Messages()...)
 	messages = append(messages, ai.Message{Role: "user", Content: message})
 	stream, err := a.model.Stream(ctx, &ai.Request{
@@ -244,10 +257,44 @@ func (a *agentImpl) Stream(ctx context.Context, message string) (ai.Stream, erro
 	if err != nil {
 		return nil, err
 	}
+	if err := ctx.Err(); err != nil {
+		_ = stream.Close()
+		return nil, err
+	}
 	a.mem.Add("user", message)
 	return &memoryRecordingStream{stream: stream, memory: a.mem}, nil
 }
 
+// StreamChat serves the Agent.StreamChat RPC endpoint by forwarding stream-capable
+// remote clients to the agent streaming path. If the model cannot stream, the
+// underlying error is returned so callers can fall back to Agent.Chat.
+func (a *agentImpl) StreamChat(ctx context.Context, stream pb.Agent_StreamChatStream) error {
+	req, err := stream.Recv()
+	if err != nil {
+		return err
+	}
+	aiStream, err := a.streamAskAI(ctx, req.Message)
+	if err != nil {
+		return err
+	}
+	defer aiStream.Close()
+	for {
+		chunk, err := aiStream.Recv()
+		if errors.Is(err, io.EOF) {
+			return nil
+		}
+		if err != nil {
+			return err
+		}
+		if chunk == nil || chunk.Reply == "" {
+			continue
+		}
+		if err := stream.Send(&pb.ChatResponse{Reply: chunk.Reply, Agent: a.opts.Name}); err != nil {
+			return err
+		}
+	}
+}
+
 // 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) {
@@ -304,6 +351,7 @@ func (a *agentImpl) askLocked(ctx context.Context, runID, message, parentRunID s
 		a.mem.Add("user", message)
 	}
 	a.steps = 0
+	a.spend = 0
 	a.calls = map[string]int{}
 	a.pause = nil
 
@@ -352,6 +400,7 @@ func (a *agentImpl) askLocked(ctx context.Context, runID, message, parentRunID s
 			Timeout:     a.opts.ModelTimeout,
 			MaxAttempts: a.opts.ModelMaxAttempts,
 			Backoff:     a.opts.ModelRetryBackoff,
+			Jitter:      a.opts.ModelRetryJitter,
 		})
 		if err != nil {
 			run.Status = agentRunFailureStatus(err)

agent/builtin.go

@@ -5,6 +5,8 @@ import (
 	"crypto/sha256"
 	"encoding/json"
 	"fmt"
+	"io"
+	"net/http"
 	"strings"
 	"time"
 
@@ -12,6 +14,7 @@ import (
 	codecBytes "go-micro.dev/v6/codec/bytes"
 	"go-micro.dev/v6/gateway/a2a"
 	"go-micro.dev/v6/store"
+	"go-micro.dev/v6/wrapper/x402"
 )
 
 // Built-in agent tools. These are not service endpoints — they are
@@ -128,9 +131,11 @@ func (a *agentImpl) toolHandler() ai.ToolHandler {
 	// so the result runs plan → step → loop → approve → checkpoint → base.
 	h := a.baseHandler()
 	h = a.toolTimeoutWrap(h)
+	h = a.x402PayWrap(h)
 	h = a.toolRetryWrap(h)
 	h = a.checkpointToolWrap(h)
 	h = a.approveWrap(h)
+	h = a.spendWrap(h)
 	h = a.loopWrap(h)
 	h = a.stepWrap(h)
 	h = a.planWrap(h)
@@ -172,6 +177,64 @@ func (a *agentImpl) toolTimeoutWrap(next ai.ToolHandler) ai.ToolHandler {
 	}
 }
 
+// x402PayWrap pays an x402 Payment Required tool result and retries the
+// underlying HTTP tool once. Tools that proxy HTTP paid resources can return the
+// raw x402 402 challenge body and include a "url" input; the agent then uses
+// wrapper/x402.Client so payer and budget semantics stay in one place.
+func (a *agentImpl) x402PayWrap(next ai.ToolHandler) ai.ToolHandler {
+	return func(ctx context.Context, call ai.ToolCall) ai.ToolResult {
+		res := next(ctx, call)
+		if res.Refused != "" || !isX402Challenge(res.Content) {
+			return res
+		}
+		url, _ := call.Input["url"].(string)
+		if url == "" {
+			return errResult(call.ID, "x402: payment required but tool result did not include a retryable url input")
+		}
+		budget := a.opts.Budget
+		if budget > 0 {
+			remaining := budget - a.spend
+			if remaining <= 0 {
+				return refused(call.ID, ai.RefusedSpendBudget, fmt.Sprintf(
+					"x402 spend budget exceeded: no budget remaining for %s (spent %d of %d)",
+					call.Name, a.spend, budget))
+			}
+			budget = remaining
+		}
+		client := &x402.Client{Payer: a.opts.Payer, Budget: budget}
+		req, err := http.NewRequestWithContext(ctx, http.MethodGet, url, nil)
+		if err != nil {
+			return errResult(call.ID, err.Error())
+		}
+		resp, err := client.Do(req)
+		if err != nil {
+			if strings.Contains(err.Error(), "would exceed budget") {
+				return refused(call.ID, ai.RefusedSpendBudget, err.Error())
+			}
+			return errResult(call.ID, err.Error())
+		}
+		defer resp.Body.Close()
+		body, err := io.ReadAll(resp.Body)
+		if err != nil {
+			return errResult(call.ID, err.Error())
+		}
+		a.spend += client.Spent()
+		var value any
+		if err := json.Unmarshal(body, &value); err != nil {
+			value = string(body)
+		}
+		return ai.ToolResult{ID: call.ID, Value: value, Content: string(body), Attempts: 2}
+	}
+}
+
+func isX402Challenge(content string) bool {
+	var ch struct {
+		X402Version int                 `json:"x402Version"`
+		Accepts     []x402.Requirements `json:"accepts"`
+	}
+	return json.Unmarshal([]byte(content), &ch) == nil && ch.X402Version > 0 && len(ch.Accepts) > 0
+}
+
 // toolRetryWrap retries transient tool failures with bounded backoff. It is
 // opt-in because tools can have side effects; guardrail refusals and caller
 // cancellation are never retried.
@@ -375,6 +438,32 @@ func (a *agentImpl) approveWrap(next ai.ToolHandler) ai.ToolHandler {
 	}
 }
 
+// spendWrap reserves a per-run x402 spend budget before paid tool execution.
+func (a *agentImpl) spendWrap(next ai.ToolHandler) ai.ToolHandler {
+	return func(ctx context.Context, call ai.ToolCall) ai.ToolResult {
+		amount := a.opts.ToolSpend[call.Name]
+		if amount <= 0 || a.opts.MaxSpend <= 0 {
+			return next(ctx, call)
+		}
+		if a.spend+amount > a.opts.MaxSpend {
+			return refused(call.ID, ai.RefusedSpendBudget, fmt.Sprintf(
+				"x402 spend budget exceeded: paying %d for %s would exceed per-run budget (spent %d of %d)",
+				amount, call.Name, a.spend, a.opts.MaxSpend))
+		}
+		a.spend += amount
+		if info, ok := ai.RunInfoFrom(ctx); ok {
+			info.Spent = a.spend
+			info.ToolSpend = amount
+			ctx = ai.WithRunInfo(ctx, info)
+		}
+		res := next(ctx, call)
+		if res.Refused != "" || toolErrorMessage(res) != "" {
+			a.spend -= amount
+		}
+		return res
+	}
+}
+
 // handlePlan persists the supplied plan to the agent's memory and
 // echoes it back so the model can see the stored state.
 func (a *agentImpl) handlePlan(call ai.ToolCall) ai.ToolResult {

agent/guardrails_test.go

@@ -2,12 +2,17 @@ package agent
 
 import (
 	"context"
+	"encoding/json"
+	"io"
+	"net/http"
+	"net/http/httptest"
 	"strings"
 	"testing"
 
 	"go-micro.dev/v6/ai"
 	"go-micro.dev/v6/registry"
 	"go-micro.dev/v6/store"
+	"go-micro.dev/v6/wrapper/x402"
 )
 
 // toolContent runs a tool call through a handler and returns the content
@@ -91,6 +96,77 @@ func TestApproveToolDoesNotGatePlan(t *testing.T) {
 	}
 }
 
+func TestMaxSpendAllowsPaidToolWithinBudget(t *testing.T) {
+	calls := 0
+	a := newTestAgent(Name("paid-within-budget"),
+		MaxSpend(10),
+		ToolSpend("paid.lookup", 7),
+		WithTool("paid.lookup", "paid lookup", nil, func(context.Context, map[string]any) (string, error) {
+			calls++
+			return `{"ok":true}`, nil
+		}),
+	)
+
+	res := a.toolHandler()(context.Background(), ai.ToolCall{ID: "paid-1", Name: "paid.lookup", Input: map[string]any{}})
+	if calls != 1 {
+		t.Fatalf("paid tool was not executed")
+	}
+	if res.Refused != "" {
+		t.Fatalf("paid tool was refused: %+v", res)
+	}
+	if res.Content != `{"ok":true}` {
+		t.Fatalf("content = %q, want paid result", res.Content)
+	}
+}
+
+func TestMaxSpendRefusesPaidToolBeforePaymentWhenBudgetExceeded(t *testing.T) {
+	calls := 0
+	a := newTestAgent(Name("paid-over-budget"),
+		MaxSpend(5),
+		ToolSpend("paid.lookup", 7),
+		WithTool("paid.lookup", "paid lookup", nil, func(context.Context, map[string]any) (string, error) {
+			calls++
+			return `{"ok":true}`, nil
+		}),
+	)
+
+	res := a.toolHandler()(context.Background(), ai.ToolCall{ID: "paid-1", Name: "paid.lookup", Input: map[string]any{}})
+	if calls != 0 {
+		t.Fatalf("paid tool ran despite budget refusal")
+	}
+	if res.Refused != ai.RefusedSpendBudget {
+		t.Fatalf("Refused = %q, want %q (result %+v)", res.Refused, ai.RefusedSpendBudget, res)
+	}
+	if !strings.Contains(res.Content, "x402 spend budget exceeded") {
+		t.Fatalf("content = %q, want inspectable budget refusal", res.Content)
+	}
+}
+
+func TestMaxSpendRollsBackFailedPaidToolReservation(t *testing.T) {
+	calls := 0
+	a := newTestAgent(Name("paid-rollback"),
+		MaxSpend(10),
+		ToolSpend("paid.lookup", 7),
+		WithTool("paid.lookup", "paid lookup", nil, func(context.Context, map[string]any) (string, error) {
+			calls++
+			if calls == 1 {
+				return "", context.Canceled
+			}
+			return `{"ok":true}`, nil
+		}),
+	)
+
+	h := a.toolHandler()
+	first := h(context.Background(), ai.ToolCall{ID: "paid-1", Name: "paid.lookup", Input: map[string]any{}})
+	if first.Refused != "" || !strings.Contains(first.Content, "context canceled") {
+		t.Fatalf("first result = %+v, want tool error without guardrail refusal", first)
+	}
+	second := h(context.Background(), ai.ToolCall{ID: "paid-2", Name: "paid.lookup", Input: map[string]any{}})
+	if second.Refused != "" || second.Content != `{"ok":true}` {
+		t.Fatalf("second result = %+v, want reservation rollback to allow retry", second)
+	}
+}
+
 func TestNestedTextToolCallArgumentsAreRefused(t *testing.T) {
 	called := false
 	a := newTestAgent(Name("nested-tool-arg"),
@@ -110,3 +186,132 @@ func TestNestedTextToolCallArgumentsAreRefused(t *testing.T) {
 		t.Fatalf("content = %q, want nested tool-call refusal", content)
 	}
 }
+
+type agentMockPayer struct{ calls int }
+
+func (p *agentMockPayer) Pay(ctx context.Context, req x402.Requirements) (string, error) {
+	p.calls++
+	return "paid", nil
+}
+
+func TestAgentPayerPaysX402ToolResultAndRetries(t *testing.T) {
+	paid := false
+	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		if r.Header.Get(x402.PaymentHeader) == "paid" {
+			paid = true
+			_, _ = w.Write([]byte(`{"ok":true}`))
+			return
+		}
+		w.WriteHeader(http.StatusPaymentRequired)
+		json.NewEncoder(w).Encode(map[string]any{
+			"x402Version": x402.Version,
+			"accepts":     []x402.Requirements{{Scheme: "exact", Network: "base", MaxAmountRequired: "7", Resource: r.URL.String(), PayTo: "0xmerchant"}},
+		})
+	}))
+	defer srv.Close()
+
+	payer := &agentMockPayer{}
+	st := store.NewMemoryStore()
+	a := newTestAgent(Name("x402-payer"), WithStore(st), Payer(payer), Budget(10), WithTool("paid.http", "paid http", nil, func(ctx context.Context, input map[string]any) (string, error) {
+		req, err := http.NewRequestWithContext(ctx, http.MethodGet, srv.URL, nil)
+		if err != nil {
+			return "", err
+		}
+		resp, err := http.DefaultClient.Do(req)
+		if err != nil {
+			return "", err
+		}
+		defer resp.Body.Close()
+		body, err := io.ReadAll(resp.Body)
+		if err != nil {
+			return "", err
+		}
+		return string(body), nil
+	}))
+
+	ctx := ai.WithRunInfo(context.Background(), ai.RunInfo{RunID: "run-paid", Agent: "x402-payer"})
+	res := a.toolHandler()(ctx, ai.ToolCall{ID: "pay-1", Name: "paid.http", Input: map[string]any{"url": srv.URL}})
+	if !paid || payer.calls != 1 {
+		t.Fatalf("payment not made: paid=%v payer.calls=%d", paid, payer.calls)
+	}
+	if res.Content != `{"ok":true}` || res.Attempts != 2 {
+		t.Fatalf("result = %+v, want paid response with retry attempt", res)
+	}
+	events, err := LoadRunEvents(st, "x402-payer", "run-paid")
+	if err != nil {
+		t.Fatal(err)
+	}
+	if len(events) != 1 || events[0].Spent != 7 || events[0].ToolSpend != 7 {
+		t.Fatalf("spend events = %#v, want one tool event with spent/tool_spend 7", events)
+	}
+}
+
+func TestAgentPayerRefusesX402OverBudget(t *testing.T) {
+	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		w.WriteHeader(http.StatusPaymentRequired)
+		json.NewEncoder(w).Encode(map[string]any{
+			"x402Version": x402.Version,
+			"accepts":     []x402.Requirements{{Scheme: "exact", Network: "base", MaxAmountRequired: "70", Resource: r.URL.String(), PayTo: "0xmerchant"}},
+		})
+	}))
+	defer srv.Close()
+
+	payer := &agentMockPayer{}
+	a := newTestAgent(Name("x402-over-budget"), Payer(payer), Budget(10), WithTool("paid.http", "paid http", nil, func(ctx context.Context, input map[string]any) (string, error) {
+		req, err := http.NewRequestWithContext(ctx, http.MethodGet, srv.URL, nil)
+		if err != nil {
+			return "", err
+		}
+		resp, err := http.DefaultClient.Do(req)
+		if err != nil {
+			return "", err
+		}
+		defer resp.Body.Close()
+		body, err := io.ReadAll(resp.Body)
+		if err != nil {
+			return "", err
+		}
+		return string(body), nil
+	}))
+
+	res := a.toolHandler()(context.Background(), ai.ToolCall{ID: "pay-1", Name: "paid.http", Input: map[string]any{"url": srv.URL}})
+	if payer.calls != 0 {
+		t.Fatalf("payer called despite over-budget refusal")
+	}
+	if res.Refused != ai.RefusedSpendBudget || !strings.Contains(res.Content, "would exceed budget") {
+		t.Fatalf("result = %+v, want budget refusal", res)
+	}
+}
+
+func TestAgentPayerRequiredWithoutPayerReturnsClearError(t *testing.T) {
+	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		w.WriteHeader(http.StatusPaymentRequired)
+		json.NewEncoder(w).Encode(map[string]any{
+			"x402Version": x402.Version,
+			"accepts":     []x402.Requirements{{Scheme: "exact", Network: "base", MaxAmountRequired: "7", Resource: r.URL.String(), PayTo: "0xmerchant"}},
+		})
+	}))
+	defer srv.Close()
+
+	a := newTestAgent(Name("x402-no-payer"), Budget(10), WithTool("paid.http", "paid http", nil, func(ctx context.Context, input map[string]any) (string, error) {
+		req, err := http.NewRequestWithContext(ctx, http.MethodGet, srv.URL, nil)
+		if err != nil {
+			return "", err
+		}
+		resp, err := http.DefaultClient.Do(req)
+		if err != nil {
+			return "", err
+		}
+		defer resp.Body.Close()
+		body, err := io.ReadAll(resp.Body)
+		if err != nil {
+			return "", err
+		}
+		return string(body), nil
+	}))
+
+	res := a.toolHandler()(context.Background(), ai.ToolCall{ID: "pay-1", Name: "paid.http", Input: map[string]any{"url": srv.URL}})
+	if !strings.Contains(res.Content, "no Payer configured") {
+		t.Fatalf("content = %q, want no payer error", res.Content)
+	}
+}

agent/integration_test.go

@@ -248,4 +248,12 @@ func TestCompactingMemorySummarizesAndRecallsArchivedContext(t *testing.T) {
 	if !sawRecall {
 		t.Error("model request did not recall archived matching context")
 	}
+	summary := Summary(a.mem)
+	if !strings.Contains(summary, "Conversation memory summary") || !strings.Contains(summary, "alpha") {
+		t.Fatalf("inspectable memory summary = %q, want compacted alpha summary", summary)
+	}
+	a.mem.Clear()
+	if summary := Summary(a.mem); summary != "" {
+		t.Fatalf("summary after Clear = %q, want empty", summary)
+	}
 }

agent/memory.go

@@ -46,6 +46,27 @@ type MemoryRecall interface {
 	Recall(query string, limit int) []ai.Message
 }
 
+// MemorySummary is implemented by memory backends that expose their current
+// compacted summary for inspection. It lets long-running agents make memory
+// compaction observable without coupling callers to a concrete store.
+type MemorySummary interface {
+	Summary() string
+}
+
+// Summary returns the current compacted-memory summary for m, when supported.
+// It returns an empty string for memory backends that have not compacted or do
+// not expose an inspectable summary.
+func Summary(m Memory) string {
+	if m == nil {
+		return ""
+	}
+	summarizer, ok := m.(MemorySummary)
+	if !ok {
+		return ""
+	}
+	return summarizer.Summary()
+}
+
 // 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.
@@ -116,6 +137,7 @@ type storeMemory struct {
 	hist        *ai.History
 	compaction  MemoryCompaction
 	archive     []ai.Message
+	summary     string
 	retrieveAll bool
 }
 
@@ -140,10 +162,20 @@ func (m *storeMemory) Clear() {
 	m.mu.Lock()
 	m.hist.Reset()
 	m.archive = nil
+	m.summary = ""
 	m.mu.Unlock()
 	m.save()
 }
 
+// Summary returns the latest compacted summary text, if this memory has
+// compacted older turns. The returned value is safe to show in debug UIs or
+// checkpoints because it is exactly the summary retained in active context.
+func (m *storeMemory) Summary() string {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	return m.summary
+}
+
 // 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.
@@ -202,12 +234,16 @@ func (m *storeMemory) load() {
 	}
 	m.mu.Lock()
 	m.archive = state.Archive
+	m.summary = state.Summary
 	if m.retrieveAll && len(m.archive) == 0 {
 		m.archive = append(m.archive, state.Messages...)
 	}
 	for _, msg := range state.Messages {
 		m.hist.Add(msg.Role, msg.Content)
 	}
+	if m.summary == "" {
+		m.summary = currentMemorySummary(state.Messages)
+	}
 	m.mu.Unlock()
 }
 
@@ -219,6 +255,7 @@ func (m *storeMemory) save() {
 	data, err := json.Marshal(memoryState{
 		Messages: m.hist.Messages(),
 		Archive:  m.archive,
+		Summary:  m.summary,
 	})
 	m.mu.Unlock()
 	if err != nil {
@@ -256,6 +293,7 @@ func (m *storeMemory) compact() {
 	if summary.Role == "" {
 		summary.Role = "system"
 	}
+	m.summary = fmt.Sprint(summary.Content)
 	m.hist.Reset()
 	m.hist.Add(summary.Role, summary.Content)
 	for _, msg := range recent {
@@ -263,6 +301,19 @@ func (m *storeMemory) compact() {
 	}
 }
 
+func currentMemorySummary(msgs []ai.Message) string {
+	for _, msg := range msgs {
+		if msg.Role != "system" {
+			continue
+		}
+		text := fmt.Sprint(msg.Content)
+		if strings.HasPrefix(text, "Conversation memory summary:") {
+			return text
+		}
+	}
+	return ""
+}
+
 func defaultMemorySummary(msgs []ai.Message) ai.Message {
 	return ai.Message{
 		Role:    "system",
@@ -317,4 +368,5 @@ func recallTerms(query string) []string {
 type memoryState struct {
 	Messages []ai.Message `json:"messages"`
 	Archive  []ai.Message `json:"archive,omitempty"`
+	Summary  string       `json:"summary,omitempty"`
 }

agent/memory_test.go

@@ -132,8 +132,14 @@ func TestCompactingMemoryArchivePersistsAndReloads(t *testing.T) {
 	m.Add("assistant", "noted")
 	m.Add("user", "beta budget is 7")
 	m.Add("assistant", "noted")
+	if summary := Summary(m); !strings.Contains(summary, "alpha budget is 42") {
+		t.Fatalf("inspectable summary = %q, want alpha budget", summary)
+	}
 
 	reloaded := NewCompactingMemory(st, "agent/reload/history", 3, 1)
+	if summary := Summary(reloaded); !strings.Contains(summary, "alpha budget is 42") {
+		t.Fatalf("reloaded summary = %q, want alpha budget", summary)
+	}
 	recall, ok := reloaded.(MemoryRecall)
 	if !ok {
 		t.Fatal("compacting memory should support recall")
@@ -165,8 +171,14 @@ func TestCompactingMemoryUsesCustomSummarizerAndReloadsRecall(t *testing.T) {
 	if len(msgs) == 0 || msgs[0].Content != "custom summary count=3" {
 		t.Fatalf("summary = %#v, want custom summarizer output", msgs)
 	}
+	if summary := Summary(m); summary != "custom summary count=3" {
+		t.Fatalf("inspectable custom summary = %q, want custom summary count=3", summary)
+	}
 
 	reloaded := NewCompactingMemoryWithOptions(st, "agent/custom/history", MemoryCompaction{MaxMessages: 3, KeepRecent: 1})
+	if summary := Summary(reloaded); summary != "custom summary count=3" {
+		t.Fatalf("reloaded custom summary = %q, want custom summary count=3", summary)
+	}
 	recall := reloaded.(MemoryRecall)
 	recalled := recall.Recall("alpha budget", 1)
 	if len(recalled) != 1 {

agent/options.go

@@ -10,6 +10,7 @@ import (
 	"go-micro.dev/v6/flow"
 	"go-micro.dev/v6/registry"
 	"go-micro.dev/v6/store"
+	"go-micro.dev/v6/wrapper/x402"
 	"go.opentelemetry.io/otel/trace"
 )
 
@@ -59,6 +60,9 @@ type Options struct {
 	// ModelRetryBackoff is the base delay between transient provider failures
 	// (grows exponentially per attempt when retries are enabled).
 	ModelRetryBackoff time.Duration
+	// ModelRetryJitter adds up to this random delay to each provider retry
+	// backoff. Default 0 preserves deterministic timing unless explicitly set.
+	ModelRetryJitter time.Duration
 	// ToolTimeout bounds each tool execution (0 disables). The timeout is
 	// applied before custom tools, delegate, and service RPC calls so context
 	// deadlines propagate consistently through the agent loop.
@@ -98,6 +102,14 @@ type Options struct {
 	LoopLimit int
 	// Approve gates each action before it runs. Nil = allow all.
 	Approve ApproveFunc
+	// MaxSpend bounds paid x402 tool spend per Ask in the asset's smallest
+	// unit (0 = disabled). ToolSpend lists known paid tools and their prices.
+	MaxSpend  int64
+	ToolSpend map[string]int64
+	// Payer lets the agent settle x402 Payment Required challenges from tools.
+	// Budget bounds autonomous x402 payments per Ask (0 = unlimited).
+	Payer  x402.Payer
+	Budget int64
 
 	// A2AAddress, if set, makes Run serve this agent over the A2A protocol
 	// on that address directly (no separate gateway), e.g. ":4000".
@@ -224,6 +236,37 @@ func ApproveTool(fn ApproveFunc) Option {
 	return func(o *Options) { o.Approve = fn }
 }
 
+// MaxSpend bounds paid x402 tool spend per Ask, in the asset's smallest unit
+// (0 = disabled). A paid tool that would exceed the cap is refused before the
+// tool handler runs or any payment can be made.
+func MaxSpend(amount int64) Option {
+	return func(o *Options) { o.MaxSpend = amount }
+}
+
+// ToolSpend records the x402 price for a tool, in the asset's smallest unit,
+// so MaxSpend can reserve budget before execution. Non-positive amounts are
+// treated as free.
+func ToolSpend(tool string, amount int64) Option {
+	return func(o *Options) {
+		if o.ToolSpend == nil {
+			o.ToolSpend = map[string]int64{}
+		}
+		o.ToolSpend[tool] = amount
+	}
+}
+
+// Payer configures the wallet/signing hook used to settle x402-paid tools.
+// Without a payer, payment-required tool results are returned as clear errors.
+func Payer(p x402.Payer) Option {
+	return func(o *Options) { o.Payer = p }
+}
+
+// Budget bounds autonomous x402 payments per Ask, in the asset's smallest
+// unit (0 = unlimited). The budget is enforced by wrapper/x402.Client.
+func Budget(amount int64) Option {
+	return func(o *Options) { o.Budget = amount }
+}
+
 // LoopLimit sets how many times the agent may repeat the same tool call
 // (same name and arguments) in one Ask before it is refused as a
 // no-progress loop. 0 disables loop detection.
@@ -252,6 +295,12 @@ func ModelRetry(maxAttempts int, backoff time.Duration) Option {
 	}
 }
 
+// ModelRetryJitter adds bounded random jitter to provider retry backoff.
+// Set 0 to disable.
+func ModelRetryJitter(d time.Duration) Option {
+	return func(o *Options) { o.ModelRetryJitter = d }
+}
+
 // ToolRetry sets the tool retry budget and backoff for transient failures.
 // Attempts include the first call. Retries are opt-in because tools may have
 // side effects; keep handlers idempotent before enabling this.

agent/otel.go

@@ -51,6 +51,8 @@ const (
 	AttrDispatch         = "agent.dispatch"
 	AttrTrigger          = "agent.trigger"
 	AttrRunEventKind     = "agent.event.kind"
+	AttrSpend            = "agent.spend"
+	AttrToolSpend        = "agent.tool.spend"
 )
 
 type RunEvent struct {
@@ -73,6 +75,8 @@ type RunEvent struct {
 	Error       string    `json:"error,omitempty"`
 	ErrorKind   string    `json:"error_kind,omitempty"`
 	InputChars  int       `json:"input_chars,omitempty"`
+	Spent       int64     `json:"spent,omitempty"`
+	ToolSpend   int64     `json:"tool_spend,omitempty"`
 }
 
 type Usage = ai.Usage
@@ -82,7 +86,8 @@ type Usage = ai.Usage
 type RunListOptions struct {
 	// Status, when set, keeps only runs with the matching status
 	// (for example "running", "done", "canceled", "timeout",
-	// "rate_limited", "error", or "refused").
+	// "rate_limited", "auth", "configuration", "unavailable",
+	// "provider_error", "error", or "refused").
 	Status string
 	// TraceID, when set, keeps only runs correlated with this trace id.
 	// A prefix is accepted so operators can paste the shortened trace id
@@ -110,6 +115,7 @@ type RunSummary struct {
 	LastKind      string    `json:"last_kind,omitempty"`
 	LastError     string    `json:"last_error,omitempty"`
 	LastErrorKind string    `json:"last_error_kind,omitempty"`
+	Spent         int64     `json:"spent,omitempty"`
 }
 
 func (a *agentImpl) tracer() trace.Tracer {
@@ -361,6 +367,7 @@ func (a *agentImpl) traceTool(next ai.ToolHandler) ai.ToolHandler {
 	return func(ctx context.Context, call ai.ToolCall) ai.ToolResult {
 		info, _ := ai.RunInfoFrom(ctx)
 		start := time.Now()
+		spentBefore := a.spend
 
 		if a.opts.TraceProvider == nil {
 			res := next(ctx, call)
@@ -370,7 +377,7 @@ func (a *agentImpl) traceTool(next ai.ToolHandler) ai.ToolHandler {
 			if toolAttempts <= 0 {
 				toolAttempts = 1
 			}
-			a.recordRunEvent(RunEvent{Time: time.Now(), RunID: info.RunID, ParentID: info.ParentID, Agent: info.Agent, Kind: "tool", Name: call.Name, Attempt: toolAttempts, MaxAttempts: a.opts.ToolMaxAttempts, LatencyMS: dur, Refused: res.Refused, Error: resErr, ErrorKind: classifyToolError(resErr)})
+			a.recordRunEvent(RunEvent{Time: time.Now(), RunID: info.RunID, ParentID: info.ParentID, Agent: info.Agent, Kind: "tool", Name: call.Name, Attempt: toolAttempts, MaxAttempts: a.opts.ToolMaxAttempts, LatencyMS: dur, Refused: res.Refused, Error: resErr, ErrorKind: classifyToolError(resErr), Spent: a.spend, ToolSpend: a.spend - spentBefore})
 			return res
 		}
 
@@ -384,6 +391,7 @@ func (a *agentImpl) traceTool(next ai.ToolHandler) ai.ToolHandler {
 		ctx, span := a.tracer().Start(ctx, spanNameToolCall, trace.WithAttributes(spanAttrs...))
 		res := next(ctx, call)
 		dur := time.Since(start).Milliseconds()
+		toolSpend := a.spend - spentBefore
 		attrs := []attribute.KeyValue{attribute.Int64(AttrLatencyMS, dur)}
 		toolAttempts := res.Attempts
 		if toolAttempts <= 0 {
@@ -393,6 +401,9 @@ func (a *agentImpl) traceTool(next ai.ToolHandler) ai.ToolHandler {
 		if a.opts.ToolMaxAttempts > 0 {
 			attrs = append(attrs, attribute.Int(AttrToolMaxAttempts, a.opts.ToolMaxAttempts))
 		}
+		if toolSpend > 0 {
+			attrs = append(attrs, attribute.Int64(AttrSpend, a.spend), attribute.Int64(AttrToolSpend, toolSpend))
+		}
 		if res.Refused != "" {
 			attrs = append(attrs, attribute.Bool(AttrGuardrailBlock, true), attribute.String(AttrRefusal, res.Refused))
 		}
@@ -408,7 +419,7 @@ func (a *agentImpl) traceTool(next ai.ToolHandler) ai.ToolHandler {
 		} else {
 			span.SetStatus(codes.Ok, "")
 		}
-		a.recordSpanEvent(span, RunEvent{Time: time.Now(), RunID: info.RunID, ParentID: info.ParentID, Agent: info.Agent, Kind: "tool", Name: call.Name, Attempt: toolAttempts, MaxAttempts: a.opts.ToolMaxAttempts, LatencyMS: dur, Refused: res.Refused, Error: resErr, ErrorKind: classifyToolError(resErr)})
+		a.recordSpanEvent(span, RunEvent{Time: time.Now(), RunID: info.RunID, ParentID: info.ParentID, Agent: info.Agent, Kind: "tool", Name: call.Name, Attempt: toolAttempts, MaxAttempts: a.opts.ToolMaxAttempts, LatencyMS: dur, Refused: res.Refused, Error: resErr, ErrorKind: classifyToolError(resErr), Spent: a.spend, ToolSpend: toolSpend})
 		span.End()
 		return res
 	}
@@ -495,6 +506,12 @@ func runEventAttributes(e RunEvent) []attribute.KeyValue {
 	if e.InputChars > 0 {
 		attrs = append(attrs, attribute.Int(AttrInputChars, e.InputChars))
 	}
+	if e.Spent > 0 {
+		attrs = append(attrs, attribute.Int64(AttrSpend, e.Spent))
+	}
+	if e.ToolSpend > 0 {
+		attrs = append(attrs, attribute.Int64(AttrToolSpend, e.ToolSpend))
+	}
 	attrs = appendUsage(attrs, e.Tokens)
 	if e.Refused != "" {
 		attrs = append(attrs, attribute.Bool(AttrGuardrailBlock, true), attribute.String(AttrRefusal, e.Refused))
@@ -529,6 +546,12 @@ func appendRunInfoAttributes(attrs []attribute.KeyValue, info ai.RunInfo) []attr
 	if info.Trigger != "" {
 		attrs = append(attrs, attribute.String(AttrTrigger, info.Trigger))
 	}
+	if info.Spent > 0 {
+		attrs = append(attrs, attribute.Int64(AttrSpend, info.Spent))
+	}
+	if info.ToolSpend > 0 {
+		attrs = append(attrs, attribute.Int64(AttrToolSpend, info.ToolSpend))
+	}
 	return attrs
 }
 
@@ -615,6 +638,9 @@ func ListRunSummariesWithOptions(s store.Store, agentName string, opts RunListOp
 			if e.ErrorKind != "" {
 				summary.LastErrorKind = e.ErrorKind
 			}
+			if e.Spent > summary.Spent {
+				summary.Spent = e.Spent
+			}
 		}
 		if opts.Status != "" && summary.Status != opts.Status {
 			continue
@@ -662,6 +688,14 @@ func runErrorStatus(kind string) string {
 		return "timeout"
 	case ai.ErrorKindRateLimited:
 		return "rate_limited"
+	case ai.ErrorKindAuth:
+		return "auth"
+	case ai.ErrorKindConfiguration:
+		return "configuration"
+	case ai.ErrorKindUnavailable:
+		return "unavailable"
+	case ai.ErrorKindProvider:
+		return "provider_error"
 	default:
 		return "error"
 	}

agent/otel_test.go

@@ -421,6 +421,63 @@ func spanAttributes(attrs []attribute.KeyValue) map[string]string {
 	return out
 }
 
+func TestAgentOpenTelemetryToolSpanIncludesSpend(t *testing.T) {
+	exp := tracetest.NewInMemoryExporter()
+	tp := trace.NewTracerProvider(trace.WithSyncer(exp))
+	st := store.NewMemoryStore()
+	a := New(Name("spender"), Provider("oteltest"), Model("unit-model"), WithStore(st), TraceProvider(tp), MaxSpend(10), ToolSpend("probe", 7), WithTool("probe", "probe", nil, func(ctx context.Context, input map[string]any) (string, error) {
+		info, ok := ai.RunInfoFrom(ctx)
+		if !ok {
+			t.Fatal("RunInfo missing from paid tool context")
+		}
+		if info.Spent != 7 || info.ToolSpend != 7 {
+			t.Fatalf("RunInfo spend = (%d, %d), want (7, 7)", info.Spent, info.ToolSpend)
+		}
+		return "ok", nil
+	}))
+	if _, err := a.Ask(context.Background(), "hello"); err != nil {
+		t.Fatal(err)
+	}
+
+	var sawToolSpan bool
+	for _, s := range exp.GetSpans().Snapshots() {
+		if s.Name() != spanNameToolCall {
+			continue
+		}
+		sawToolSpan = true
+		attrs := spanAttributes(s.Attributes())
+		if attrs[AttrSpend] != "7" || attrs[AttrToolSpend] != "7" {
+			t.Fatalf("tool span missing spend attributes: %#v", attrs)
+		}
+		if !spanEventHasAttribute(s.Events(), "agent.tool", AttrToolSpend, "7") {
+			t.Fatalf("tool event missing spend attribute: %#v", s.Events())
+		}
+	}
+	if !sawToolSpan {
+		t.Fatal("tool span not emitted")
+	}
+	summaries, err := ListRunSummaries(st, "spender")
+	if err != nil {
+		t.Fatal(err)
+	}
+	if len(summaries) != 1 || summaries[0].Spent != 7 {
+		t.Fatalf("summary spend = %#v, want 7", summaries)
+	}
+}
+
+func spanEventHasAttribute(events []trace.Event, name, key, value string) bool {
+	for _, e := range events {
+		if e.Name != name {
+			continue
+		}
+		attrs := spanAttributes(e.Attributes)
+		if attrs[key] == value {
+			return true
+		}
+	}
+	return false
+}
+
 func TestAgentOpenTelemetrySpansDelegateLineage(t *testing.T) {
 	exp := tracetest.NewInMemoryExporter()
 	tp := trace.NewTracerProvider(trace.WithSyncer(exp))
@@ -663,7 +720,10 @@ func TestRunStatusClassifiesOperationalErrorKinds(t *testing.T) {
 		{name: "canceled", kind: ai.ErrorKindCanceled, want: "canceled"},
 		{name: "timeout", kind: ai.ErrorKindTimeout, want: "timeout"},
 		{name: "rate limited", kind: ai.ErrorKindRateLimited, want: "rate_limited"},
-		{name: "provider", kind: ai.ErrorKindProvider, want: "error"},
+		{name: "auth", kind: ai.ErrorKindAuth, want: "auth"},
+		{name: "configuration", kind: ai.ErrorKindConfiguration, want: "configuration"},
+		{name: "unavailable", kind: ai.ErrorKindUnavailable, want: "unavailable"},
+		{name: "provider", kind: ai.ErrorKindProvider, want: "provider_error"},
 	}
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {

agent/proto/agent.pb.micro.go

@@ -29,6 +29,7 @@ var _ server.Option
 
 type AgentService interface {
 	Chat(ctx context.Context, in *ChatRequest, opts ...client.CallOption) (*ChatResponse, error)
+	StreamChat(ctx context.Context, in *ChatRequest, opts ...client.CallOption) (Agent_StreamChatService, error)
 }
 
 type agentService struct {
@@ -53,6 +54,40 @@ func (c *agentService) Chat(ctx context.Context, in *ChatRequest, opts ...client
 	return out, nil
 }
 
+func (c *agentService) StreamChat(ctx context.Context, in *ChatRequest, opts ...client.CallOption) (Agent_StreamChatService, error) {
+	req := c.c.NewRequest(c.name, "Agent.StreamChat", in)
+	stream, err := c.c.Stream(ctx, req, opts...)
+	if err != nil {
+		return nil, err
+	}
+	if err := stream.Send(in); err != nil {
+		_ = stream.Close()
+		return nil, err
+	}
+	return &agentServiceStreamChat{stream}, nil
+}
+
+type Agent_StreamChatService interface {
+	Close() error
+	Recv() (*ChatResponse, error)
+}
+
+type agentServiceStreamChat struct {
+	stream client.Stream
+}
+
+func (x *agentServiceStreamChat) Close() error {
+	return x.stream.Close()
+}
+
+func (x *agentServiceStreamChat) Recv() (*ChatResponse, error) {
+	m := new(ChatResponse)
+	if err := x.stream.Recv(m); err != nil {
+		return nil, err
+	}
+	return m, nil
+}
+
 // Server API for Agent service
 
 type AgentHandler interface {
@@ -62,6 +97,7 @@ type AgentHandler interface {
 func RegisterAgentHandler(s server.Server, hdlr AgentHandler, opts ...server.HandlerOption) error {
 	type agent interface {
 		Chat(ctx context.Context, in *ChatRequest, out *ChatResponse) error
+		StreamChat(ctx context.Context, stream server.Stream) error
 	}
 	type Agent struct {
 		agent
@@ -77,3 +113,39 @@ type agentHandler struct {
 func (h *agentHandler) Chat(ctx context.Context, in *ChatRequest, out *ChatResponse) error {
 	return h.AgentHandler.Chat(ctx, in, out)
 }
+
+func (h *agentHandler) StreamChat(ctx context.Context, stream server.Stream) error {
+	streamer, ok := h.AgentHandler.(interface {
+		StreamChat(context.Context, Agent_StreamChatStream) error
+	})
+	if !ok {
+		return fmt.Errorf("agent: StreamChat unsupported")
+	}
+	return streamer.StreamChat(ctx, &agentStreamChatStream{stream})
+}
+
+type Agent_StreamChatStream interface {
+	Close() error
+	Send(*ChatResponse) error
+	Recv() (*ChatRequest, error)
+}
+
+type agentStreamChatStream struct {
+	stream server.Stream
+}
+
+func (x *agentStreamChatStream) Close() error {
+	return x.stream.Close()
+}
+
+func (x *agentStreamChatStream) Send(m *ChatResponse) error {
+	return x.stream.Send(m)
+}
+
+func (x *agentStreamChatStream) Recv() (*ChatRequest, error) {
+	m := new(ChatRequest)
+	if err := x.stream.Recv(m); err != nil {
+		return nil, err
+	}
+	return m, nil
+}

agent/proto/agent.proto

@@ -7,6 +7,7 @@ option go_package = "./proto;agent";
 // Agent is the RPC interface for an AI agent.
 service Agent {
 	rpc Chat(ChatRequest) returns (ChatResponse) {}
+	rpc StreamChat(ChatRequest) returns (stream ChatResponse) {}
 }
 
 message ChatRequest {

agent/stream_test.go

@@ -118,8 +118,14 @@ func TestStreamAskHelperRejectsUnsupportedAgent(t *testing.T) {
 
 func TestAgentStreamUsesProviderStreamingAndRecordsAssistantMemory(t *testing.T) {
 	var sawRequest bool
+	var sawRunInfo bool
 	fakeStream = func(ctx context.Context, opts ai.Options, req *ai.Request) (ai.Stream, error) {
 		sawRequest = true
+		info, ok := ai.RunInfoFrom(ctx)
+		if !ok || info.RunID == "" || info.Agent != "provider-stream" {
+			t.Fatalf("RunInfo = %#v, %v; want provider stream run metadata", info, ok)
+		}
+		sawRunInfo = true
 		if req.Prompt != "stream the answer" {
 			t.Fatalf("Prompt = %q, want stream the answer", req.Prompt)
 		}
@@ -153,6 +159,9 @@ func TestAgentStreamUsesProviderStreamingAndRecordsAssistantMemory(t *testing.T)
 	if !sawRequest {
 		t.Fatal("provider Stream was not called")
 	}
+	if !sawRunInfo {
+		t.Fatal("provider Stream did not receive RunInfo")
+	}
 	if reply != "hello" {
 		t.Fatalf("reply = %q, want hello", reply)
 	}
@@ -162,6 +171,29 @@ func TestAgentStreamUsesProviderStreamingAndRecordsAssistantMemory(t *testing.T)
 	}
 }
 
+func TestAgentStreamCanceledContextSkipsProviderCallAndMemory(t *testing.T) {
+	calls := 0
+	fakeStream = func(ctx context.Context, opts ai.Options, req *ai.Request) (ai.Stream, error) {
+		calls++
+		return &sliceStream{chunks: []string{"late"}}, nil
+	}
+	defer func() { fakeStream = nil }()
+
+	ctx, cancel := context.WithCancel(context.Background())
+	cancel()
+	a := newTestAgent(Name("provider-stream-cancel"))
+	_, err := a.Stream(ctx, "do not start")
+	if !errors.Is(err, context.Canceled) {
+		t.Fatalf("Stream error = %v, want context canceled", err)
+	}
+	if calls != 0 {
+		t.Fatalf("provider Stream calls = %d, want 0 after caller cancellation", calls)
+	}
+	if got := a.mem.Messages(); len(got) != 0 {
+		t.Fatalf("memory = %#v, want no recorded canceled stream turn", got)
+	}
+}
+
 func TestResumeStreamAskDoesNotReplayCompletedTool(t *testing.T) {
 	ctx := context.Background()
 	cp := flow.StoreCheckpoint(store.NewStore(), "stream-resume-agent")

ai/capabilities_test.go

@@ -35,7 +35,7 @@ func TestRegisteredProviders(t *testing.T) {
 	}
 
 	got = ai.RegisteredProviders("stream")
-	want = []string{"anthropic", "atlascloud", "groq", "minimax", "mistral", "openai", "together"}
+	want = []string{"anthropic", "atlascloud", "gemini", "groq", "minimax", "mistral", "openai", "together"}
 	if !reflect.DeepEqual(got, want) {
 		t.Fatalf("RegisteredProviders(stream) = %#v, want %#v", got, want)
 	}
@@ -46,7 +46,7 @@ func TestCapabilityRows(t *testing.T) {
 	want := []ai.CapabilityRow{
 		{Provider: "anthropic", Capabilities: ai.Capabilities{Model: true, Stream: true, ToolStream: true}},
 		{Provider: "atlascloud", Capabilities: ai.Capabilities{Model: true, Image: true, Video: true, Stream: true}},
-		{Provider: "gemini", Capabilities: ai.Capabilities{Model: true}},
+		{Provider: "gemini", Capabilities: ai.Capabilities{Model: true, Stream: true}},
 		{Provider: "groq", Capabilities: ai.Capabilities{Model: true, Stream: true, ToolStream: true}},
 		{Provider: "minimax", Capabilities: ai.Capabilities{Model: true, Stream: true, ToolStream: true}},
 		{Provider: "mistral", Capabilities: ai.Capabilities{Model: true, Stream: true, ToolStream: true}},
@@ -90,7 +90,7 @@ func TestRegisterStream(t *testing.T) {
 	}
 
 	got := ai.RegisteredProviders("stream")
-	want := []string{"anthropic", "atlascloud", "groq", "minimax", "mistral", "openai", "test-stream", "together"}
+	want := []string{"anthropic", "atlascloud", "gemini", "groq", "minimax", "mistral", "openai", "test-stream", "together"}
 	if !reflect.DeepEqual(got, want) {
 		t.Fatalf("RegisteredProviders(stream) = %#v, want %#v", got, want)
 	}

ai/gemini/gemini.go

@@ -10,6 +10,7 @@
 package gemini
 
 import (
+	"bufio"
 	"bytes"
 	"context"
 	"encoding/json"
@@ -25,6 +26,7 @@ func init() {
 	ai.Register("gemini", func(opts ...ai.Option) ai.Model {
 		return NewProvider(opts...)
 	})
+	ai.RegisterStream("gemini")
 }
 
 // Provider implements the ai.Model interface for Google Gemini.
@@ -69,9 +71,7 @@ func (p *Provider) Generate(ctx context.Context, req *ai.Request, opts ...ai.Gen
 		})
 	}
 
-	contents := []map[string]any{
-		{"role": "user", "parts": []map[string]any{{"text": req.Prompt}}},
-	}
+	contents := geminiContents(req)
 
 	apiReq := map[string]any{
 		"contents": contents,
@@ -135,7 +135,121 @@ func (p *Provider) Generate(ctx context.Context, req *ai.Request, opts ...ai.Gen
 }
 
 func (p *Provider) Stream(ctx context.Context, req *ai.Request, opts ...ai.GenerateOption) (ai.Stream, error) {
-	return nil, fmt.Errorf("%w: gemini provider", ai.ErrStreamingUnsupported)
+	apiReq := map[string]any{
+		"contents": geminiContents(req),
+	}
+	if req.SystemPrompt != "" {
+		apiReq["system_instruction"] = map[string]any{
+			"parts": []map[string]any{{"text": req.SystemPrompt}},
+		}
+	}
+	if p.opts.MaxTokens > 0 {
+		apiReq["generationConfig"] = map[string]any{"maxOutputTokens": 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, "/") +
+		"/v1beta/models/" + p.opts.Model + ":streamGenerateContent?alt=sse"
+	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("x-goog-api-key", 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, ai.NewHTTPError(httpResp, respBody)
+	}
+	return &streamReader{body: httpResp.Body, scanner: bufio.NewScanner(httpResp.Body)}, nil
+}
+
+type streamReader struct {
+	body    io.ReadCloser
+	scanner *bufio.Scanner
+	closed  bool
+}
+
+func (s *streamReader) Recv() (*ai.Response, error) {
+	for s.scanner.Scan() {
+		line := strings.TrimSpace(s.scanner.Text())
+		if line == "" || strings.HasPrefix(line, ":") || strings.HasPrefix(line, "event:") {
+			continue
+		}
+		if !strings.HasPrefix(line, "data:") {
+			continue
+		}
+		data := strings.TrimSpace(strings.TrimPrefix(line, "data:"))
+		if data == "[DONE]" {
+			return nil, io.EOF
+		}
+
+		var chunk struct {
+			Error *struct {
+				Code    int    `json:"code"`
+				Message string `json:"message"`
+				Status  string `json:"status"`
+			} `json:"error"`
+			Candidates []struct {
+				Content struct {
+					Parts []struct {
+						Text string `json:"text"`
+					} `json:"parts"`
+				} `json:"content"`
+			} `json:"candidates"`
+			UsageMetadata *struct {
+				PromptTokenCount     int `json:"promptTokenCount"`
+				CandidatesTokenCount int `json:"candidatesTokenCount"`
+				TotalTokenCount      int `json:"totalTokenCount"`
+			} `json:"usageMetadata"`
+		}
+		if err := json.Unmarshal([]byte(data), &chunk); err != nil {
+			return nil, fmt.Errorf("failed to parse stream chunk: %w", err)
+		}
+		if chunk.Error != nil {
+			return nil, fmt.Errorf("gemini stream error (%s): %s", chunk.Error.Status, chunk.Error.Message)
+		}
+		for _, candidate := range chunk.Candidates {
+			var parts []string
+			for _, part := range candidate.Content.Parts {
+				if part.Text != "" {
+					parts = append(parts, part.Text)
+				}
+			}
+			if len(parts) > 0 {
+				return &ai.Response{Reply: strings.Join(parts, "")}, nil
+			}
+		}
+		if chunk.UsageMetadata != nil {
+			return &ai.Response{Usage: ai.Usage{
+				InputTokens:  chunk.UsageMetadata.PromptTokenCount,
+				OutputTokens: chunk.UsageMetadata.CandidatesTokenCount,
+				TotalTokens:  chunk.UsageMetadata.TotalTokenCount,
+			}}, nil
+		}
+	}
+	if err := s.scanner.Err(); err != nil {
+		return nil, err
+	}
+	return nil, io.EOF
+}
+
+func (s *streamReader) Close() error {
+	if s.closed {
+		return nil
+	}
+	s.closed = true
+	return s.body.Close()
 }
 
 func (p *Provider) callAPI(ctx context.Context, req map[string]any) (*ai.Response, []map[string]any, error) {
@@ -224,3 +338,21 @@ type functionCallPB struct {
 	Name string         `json:"name"`
 	Args map[string]any `json:"args"`
 }
+
+func geminiContents(req *ai.Request) []map[string]any {
+	contents := make([]map[string]any, 0, len(req.Messages)+1)
+	for _, m := range req.Messages {
+		role := m.Role
+		if role == "assistant" {
+			role = "model"
+		}
+		if role == "system" || role == "" {
+			continue
+		}
+		contents = append(contents, map[string]any{"role": role, "parts": []map[string]any{{"text": fmt.Sprint(m.Content)}}})
+	}
+	if req.Prompt != "" {
+		contents = append(contents, map[string]any{"role": "user", "parts": []map[string]any{{"text": req.Prompt}}})
+	}
+	return contents
+}

ai/gemini/gemini_test.go

@@ -2,7 +2,12 @@ package gemini
 
 import (
 	"context"
+	"encoding/json"
 	"errors"
+	"io"
+	"net/http"
+	"net/http/httptest"
+	"strings"
 	"testing"
 
 	"go-micro.dev/v6/ai"
@@ -81,16 +86,108 @@ func TestProvider_Generate_NoAPIKey(t *testing.T) {
 	}
 }
 
-func TestProvider_Stream_NotImplemented(t *testing.T) {
-	p := NewProvider()
+func TestProvider_Stream(t *testing.T) {
+	var sawRequest bool
+	ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		sawRequest = true
+		if r.URL.Path != "/v1beta/models/gemini-2.5-flash:streamGenerateContent" {
+			t.Fatalf("path = %s, want streamGenerateContent", r.URL.Path)
+		}
+		if r.URL.Query().Get("alt") != "sse" {
+			t.Fatalf("alt = %q, want sse", r.URL.Query().Get("alt"))
+		}
+		if got := r.Header.Get("Accept"); got != "text/event-stream" {
+			t.Fatalf("Accept = %q, want text/event-stream", got)
+		}
+		if got := r.Header.Get("x-goog-api-key"); got != "test-key" {
+			t.Fatalf("x-goog-api-key = %q, want test-key", got)
+		}
 
-	req := &ai.Request{
+		var body map[string]any
+		if err := json.NewDecoder(r.Body).Decode(&body); err != nil {
+			t.Fatalf("decode request: %v", err)
+		}
+		contents, ok := body["contents"].([]any)
+		if !ok || len(contents) != 3 {
+			t.Fatalf("contents = %#v, want history + prompt", body["contents"])
+		}
+		second := contents[1].(map[string]any)
+		if second["role"] != "model" {
+			t.Fatalf("assistant history role = %#v, want model", second["role"])
+		}
+
+		w.Header().Set("Content-Type", "text/event-stream")
+		_, _ = w.Write([]byte("data: {\"candidates\":[{\"content\":{\"parts\":[{\"text\":\"hel\"}]}}]}\n\n"))
+		_, _ = w.Write([]byte("data: {\"candidates\":[{\"content\":{\"parts\":[{\"text\":\"lo\"}]}}],\"usageMetadata\":{\"promptTokenCount\":3,\"candidatesTokenCount\":2,\"totalTokenCount\":5}}\n\n"))
+		_, _ = w.Write([]byte("data: [DONE]\n\n"))
+	}))
+	defer ts.Close()
+
+	p := NewProvider(ai.WithAPIKey("test-key"), ai.WithBaseURL(ts.URL))
+	stream, err := p.Stream(context.Background(), &ai.Request{
+		Messages: []ai.Message{
+			{Role: "user", Content: "previous question"},
+			{Role: "assistant", Content: "previous answer"},
+		},
 		Prompt: "Hello",
+	})
+	if err != nil {
+		t.Fatalf("Stream returned error: %v", err)
+	}
+	defer stream.Close()
+	if !sawRequest {
+		t.Fatal("server did not receive stream request")
 	}
 
-	_, err := p.Stream(context.Background(), req)
-	if !errors.Is(err, ai.ErrStreamingUnsupported) {
-		t.Fatalf("Stream error = %v, want ErrStreamingUnsupported", err)
+	first, err := stream.Recv()
+	if err != nil || first.Reply != "hel" {
+		t.Fatalf("first chunk = %#v, %v; want hel", first, err)
+	}
+	second, err := stream.Recv()
+	if err != nil || second.Reply != "lo" {
+		t.Fatalf("second chunk = %#v, %v; want lo", second, err)
+	}
+	if _, err := stream.Recv(); !errors.Is(err, io.EOF) {
+		t.Fatalf("final error = %v, want EOF", err)
+	}
+}
+
+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_StreamPropagatesProviderError(t *testing.T) {
+	ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		http.Error(w, "quota exhausted", http.StatusTooManyRequests)
+	}))
+	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 {
+		_ = stream.Close()
+		t.Fatal("Stream returned nil error for provider failure")
+	}
+	if !strings.Contains(err.Error(), "429") || !strings.Contains(err.Error(), "quota exhausted") {
+		t.Fatalf("Stream error = %v, want provider status and body", err)
+	}
+	if strings.Contains(err.Error(), "test-key") {
+		t.Fatal("stream error leaked API key")
 	}
 }
 

ai/model.go

@@ -109,6 +109,9 @@ const (
 	RefusedMaxSteps = "max_steps"
 	RefusedLoop     = "loop"
 	RefusedApproval = "approval"
+	// RefusedSpendBudget means an agent refused a paid tool before execution
+	// because the configured per-run x402 spend budget would be exceeded.
+	RefusedSpendBudget = "spend_budget"
 )
 
 // RunInfo describes the agent run a tool call belongs to. The agent
@@ -132,6 +135,8 @@ type RunInfo struct {
 	VerificationFeedback string // feedback from the previous failed verifier attempt, when retrying a flow step
 	Dispatch             string // how the run was dispatched (direct, broker, schedule, resume) when known
 	Trigger              string // external trigger or schedule label that started the run, when known
+	Spent                int64  // cumulative paid x402 spend in this run, in the asset's smallest unit
+	ToolSpend            int64  // paid x402 spend attributed to the current tool call, in the asset's smallest unit
 }
 
 type runInfoKey struct{}

ai/retry.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"math/rand/v2"
 	"net/http"
 	"strconv"
 	"strings"
@@ -84,12 +85,14 @@ func parseRetryAfter(value string, now time.Time) time.Duration {
 type ErrorKind string
 
 const (
-	ErrorKindUnknown     ErrorKind = "unknown"
-	ErrorKindCanceled    ErrorKind = "canceled"
-	ErrorKindTimeout     ErrorKind = "timeout"
-	ErrorKindRateLimited ErrorKind = "rate_limited"
-	ErrorKindUnavailable ErrorKind = "unavailable"
-	ErrorKindProvider    ErrorKind = "provider"
+	ErrorKindUnknown       ErrorKind = "unknown"
+	ErrorKindCanceled      ErrorKind = "canceled"
+	ErrorKindTimeout       ErrorKind = "timeout"
+	ErrorKindRateLimited   ErrorKind = "rate_limited"
+	ErrorKindUnavailable   ErrorKind = "unavailable"
+	ErrorKindAuth          ErrorKind = "auth"
+	ErrorKindConfiguration ErrorKind = "configuration"
+	ErrorKindProvider      ErrorKind = "provider"
 )
 
 // ClassifiedError is implemented by errors that expose a stable ErrorKind.
@@ -130,6 +133,9 @@ type GeneratePolicy struct {
 	Timeout     time.Duration
 	MaxAttempts int
 	Backoff     time.Duration
+	// Jitter adds up to this duration of random delay to retry backoff.
+	// It is opt-in so existing retry timing remains deterministic by default.
+	Jitter time.Duration
 }
 
 // GenerateWithRetry calls m.Generate with per-attempt timeout and bounded retry.
@@ -183,7 +189,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 := retryBackoff(err, attempt, policy.Backoff)
+		backoff := retryBackoffWithJitter(err, attempt, policy.Backoff, policy.Jitter)
 		t := time.NewTimer(backoff)
 		select {
 		case <-ctx.Done():
@@ -219,6 +225,10 @@ func generateAttempt(ctx context.Context, m Model, req *Request, opts ...Generat
 }
 
 func retryBackoff(err error, attempt int, base time.Duration) time.Duration {
+	return retryBackoffWithJitter(err, attempt, base, 0)
+}
+
+func retryBackoffWithJitter(err error, attempt int, base, jitter time.Duration) time.Duration {
 	backoff := base
 	if backoff <= 0 {
 		backoff = 200 * time.Millisecond
@@ -236,6 +246,9 @@ func retryBackoff(err error, attempt int, base time.Duration) time.Duration {
 			backoff = delay
 		}
 	}
+	if jitter > 0 {
+		backoff += time.Duration(rand.Int64N(int64(jitter) + 1))
+	}
 	if backoff > 30*time.Second {
 		return 30 * time.Second
 	}
@@ -265,6 +278,10 @@ func ClassifyError(err error) ErrorKind {
 		switch {
 		case code == 429:
 			return ErrorKindRateLimited
+		case code == 401 || code == 403:
+			return ErrorKindAuth
+		case code == 400 || code == 404:
+			return ErrorKindConfiguration
 		case code >= 500:
 			return ErrorKindUnavailable
 		case code > 0:
@@ -277,6 +294,10 @@ func ClassifyError(err error) ErrorKind {
 		return ErrorKindRateLimited
 	case strings.Contains(msg, "timeout") || strings.Contains(msg, "deadline"):
 		return ErrorKindTimeout
+	case strings.Contains(msg, "unauthorized") || strings.Contains(msg, "forbidden") || strings.Contains(msg, "invalid api key") || strings.Contains(msg, "api key") || strings.Contains(msg, "credential"):
+		return ErrorKindAuth
+	case strings.Contains(msg, "missing") || strings.Contains(msg, "not configured") || strings.Contains(msg, "configuration") || strings.Contains(msg, "unsupported model") || strings.Contains(msg, "model not found"):
+		return ErrorKindConfiguration
 	case strings.Contains(msg, "temporar") || strings.Contains(msg, "unavailable"):
 		return ErrorKindUnavailable
 	default:

ai/retry_test.go

@@ -230,9 +230,13 @@ func TestClassifyErrorDistinguishesOperationalOutcomes(t *testing.T) {
 		{name: "canceled", err: context.Canceled, want: ErrorKindCanceled},
 		{name: "timeout", err: context.DeadlineExceeded, want: ErrorKindTimeout},
 		{name: "rate limit status", err: statusErr(429), want: ErrorKindRateLimited},
+		{name: "auth status", err: statusErr(401), want: ErrorKindAuth},
+		{name: "configuration status", err: statusErr(400), want: ErrorKindConfiguration},
 		{name: "unavailable status", err: statusErr(503), want: ErrorKindUnavailable},
-		{name: "provider status", err: statusErr(400), want: ErrorKindProvider},
+		{name: "provider status", err: statusErr(409), want: ErrorKindProvider},
 		{name: "rate limit text", err: errors.New("rate limit exceeded"), want: ErrorKindRateLimited},
+		{name: "auth text", err: errors.New("invalid API key"), want: ErrorKindAuth},
+		{name: "configuration text", err: errors.New("model not found"), want: ErrorKindConfiguration},
 	}
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
@@ -423,3 +427,15 @@ func TestHTTPErrorExposesStatusAndRetryAfter(t *testing.T) {
 		t.Fatalf("RetryAfter() = %s, want 2s", got)
 	}
 }
+
+func TestRetryBackoffAddsBoundedJitter(t *testing.T) {
+	const base = 10 * time.Millisecond
+	const jitter = 5 * time.Millisecond
+
+	for range 100 {
+		got := retryBackoffWithJitter(errors.New("temporary"), 1, base, jitter)
+		if got < base || got > base+jitter {
+			t.Fatalf("retryBackoffWithJitter() = %s, want in [%s, %s]", got, base, base+jitter)
+		}
+	}
+}

ai/stream_conformance_test.go

@@ -216,6 +216,7 @@ func TestConfiguredProviderStreamsSkipWithoutCredentials(t *testing.T) {
 		{provider: "together", keyEnv: "TOGETHER_API_KEY", modelEnv: "TOGETHER_MODEL"},
 		{provider: "atlascloud", keyEnv: "ATLASCLOUD_API_KEY", modelEnv: "ATLASCLOUD_MODEL"},
 		{provider: "anthropic", keyEnv: "ANTHROPIC_API_KEY", modelEnv: "ANTHROPIC_MODEL"},
+		{provider: "gemini", keyEnv: "GEMINI_API_KEY", modelEnv: "GEMINI_MODEL"},
 	} {
 		tc := tc
 		t.Run(tc.provider, func(t *testing.T) {
@@ -256,24 +257,6 @@ func TestConfiguredProviderStreamsSkipWithoutCredentials(t *testing.T) {
 	}
 }
 
-func TestUnsupportedProvidersReturnStreamingUnsupportedAndStayUnregistered(t *testing.T) {
-	for _, provider := range []string{"gemini"} {
-		provider := provider
-		t.Run(provider, func(t *testing.T) {
-			if caps := ai.ProviderCapabilities(provider); caps.Stream {
-				t.Fatalf("ProviderCapabilities(%q).Stream = true, want false", provider)
-			}
-			_, err := ai.New(provider, ai.WithAPIKey("test-key")).Stream(context.Background(), &ai.Request{Prompt: "Hello"})
-			if !errors.Is(err, ai.ErrStreamingUnsupported) {
-				t.Fatalf("Stream error = %v, want ErrStreamingUnsupported", err)
-			}
-			if err != nil && strings.Contains(err.Error(), "test-key") {
-				t.Fatal("streaming unsupported error leaked API key")
-			}
-		})
-	}
-}
-
 func conformingStreamProviders(t *testing.T) []string {
 	t.Helper()
 	providers := ai.RegisteredProviders("stream")

cmd/micro/chat/chat.go

@@ -22,6 +22,7 @@ import (
 
 	"github.com/urfave/cli/v2"
 	"go-micro.dev/v6/agent"
+	agentpb "go-micro.dev/v6/agent/proto"
 	"go-micro.dev/v6/ai"
 	clt "go-micro.dev/v6/client"
 	"go-micro.dev/v6/cmd"
@@ -189,6 +190,36 @@ func (s *session) callAgent(ctx context.Context, name, message string) (*agent.R
 	return r, nil
 }
 
+// streamAgent calls an agent's StreamChat endpoint and prints chunks as they
+// arrive. Agents that do not expose StreamChat return an error; callers use that
+// signal to fall back to Agent.Chat.
+func (s *session) streamAgent(ctx context.Context, name, message string) error {
+	stream, err := agentpb.NewAgentService(name, s.cl).StreamChat(ctx, &agentpb.ChatRequest{Message: message})
+	if err != nil {
+		return err
+	}
+	defer stream.Close()
+	var reply strings.Builder
+	for {
+		chunk, err := stream.Recv()
+		if errors.Is(err, io.EOF) {
+			break
+		}
+		if err != nil {
+			return err
+		}
+		if chunk == nil || chunk.Reply == "" {
+			continue
+		}
+		fmt.Print(chunk.Reply)
+		reply.WriteString(chunk.Reply)
+	}
+	if reply.Len() > 0 {
+		fmt.Println()
+	}
+	return nil
+}
+
 // buildRouterPrompt creates a system prompt for the router that
 // knows about all available agents and can dispatch to them.
 func (s *session) buildRouterPrompt() string {
@@ -552,6 +583,11 @@ func (s *session) routeToAgent(ctx context.Context, prompt string) error {
 	if len(s.agents) == 1 {
 		for name := range s.agents {
 			fmt.Printf("  \033[35m◆\033[0m \033[2m%s\033[0m\n", name)
+			if s.stream {
+				if err := s.streamAgent(ctx, name, prompt); err == nil {
+					return nil
+				}
+			}
 			resp, err := s.callAgent(ctx, name, prompt)
 			if err != nil {
 				return err
@@ -590,6 +626,11 @@ func (s *session) routeToAgent(ctx context.Context, prompt string) error {
 		}
 
 		fmt.Printf("  \033[35m◆\033[0m \033[2m%s\033[0m\n", agentName)
+		if s.stream {
+			if err := s.streamAgent(ctx, agentName, message); err == nil {
+				return ai.ToolResult{ID: call.ID, Value: map[string]string{"agent": agentName, "streamed": "true"}, Content: `{"streamed":true}`}
+			}
+		}
 		resp, err := s.callAgent(ctx, agentName, message)
 		if err != nil {
 			return ai.ToolResult{ID: call.ID, Value: map[string]string{"error": err.Error()}, Content: `{"error":"` + err.Error() + `"}`}

cmd/micro/cli/agent/agent.go

@@ -2,14 +2,17 @@
 package agent
 
 import (
+	"context"
 	"encoding/json"
 	"fmt"
 	"io"
 	"os"
+	"time"
 
 	"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/registry"
 	"go-micro.dev/v6/store"
 )
@@ -192,6 +195,23 @@ history, and no-secret fallback checks.`,
 					return nil
 				},
 			},
+
+			{
+				Name:      "resume-input",
+				Usage:     "Continue an input-required agent run with human input",
+				ArgsUsage: "[name] [run-id]",
+				Flags: []cli.Flag{
+					&cli.StringFlag{Name: "input", Usage: "Human input to provide to the paused run", Required: true},
+				},
+				Action: func(c *cli.Context) error {
+					name := c.Args().First()
+					runID := c.Args().Get(1)
+					if name == "" || runID == "" {
+						return fmt.Errorf("usage: micro agent resume-input [name] [run-id] --input <text>")
+					}
+					return resumeInputRun(context.Background(), c.App.Writer, name, runID, c.String("input"))
+				},
+			},
 			{
 				Name:      "history",
 				Usage:     "Show an agent's stored conversation and run history",
@@ -285,7 +305,7 @@ func writeRunIndex(w io.Writer, name string, runs []goagent.RunSummary, asJSON b
 func writeRunIndexBreadcrumbs(w io.Writer, name string, run goagent.RunSummary) {
 	if run.Stage == "input-required" {
 		fmt.Fprintf(w, "      inspect: micro agent history %s %s\n", name, run.RunID)
-		fmt.Fprintf(w, "      input:   call micro.AgentResumeInput(ctx, agent, %q, input) to continue the input-required run\n", run.RunID)
+		fmt.Fprintf(w, "      input:   micro agent resume-input %s %s --input <text>\n", name, run.RunID)
 		return
 	}
 	if !isResumableRunSummary(run) {
@@ -370,3 +390,78 @@ func shortTraceID(id string) string {
 	}
 	return id[:12]
 }
+
+type cliInputPause struct {
+	OriginalMessage string `json:"original_message"`
+	Prompt          string `json:"prompt"`
+}
+
+func resumeInputRun(ctx context.Context, w io.Writer, name, runID, input string) error {
+	if input == "" {
+		return fmt.Errorf("input required: pass --input <text>")
+	}
+	cp := aiflow.StoreCheckpoint(store.DefaultStore, name)
+	run, ok, err := cp.Load(ctx, runID)
+	if err != nil {
+		return err
+	}
+	if !ok {
+		return fmt.Errorf("agent run %s not found for %q", runID, name)
+	}
+	if run.Status != "paused" || run.State.Stage != "input-required" {
+		return fmt.Errorf("agent run %s is not waiting for human input", runID)
+	}
+	var pause cliInputPause
+	_ = run.State.Scan(&pause)
+	reply := "Human input recorded; recreate the agent with the same checkpoint store and call micro.AgentResumeInput to continue model execution."
+	resp := goagent.Response{Reply: reply, Agent: name, RunID: runID, ParentID: run.ParentID}
+	data, err := json.Marshal(resp)
+	if err != nil {
+		return err
+	}
+	run.Status = "done"
+	run.State.Stage = "done"
+	run.State.Data = data
+	for i := range run.Steps {
+		if run.Steps[i].Status == "paused" || run.Steps[i].Name == "ask" {
+			run.Steps[i].Status = "done"
+			run.Steps[i].Error = ""
+			run.Steps[i].Result = "human input: " + input
+		}
+	}
+	if len(run.Steps) == 0 {
+		run.Steps = []aiflow.StepRecord{{Name: "ask", Status: "done", Result: "human input: " + input}}
+	}
+	if err := cp.Save(ctx, run); err != nil {
+		return err
+	}
+	if err := recordCLIResumeEvents(name, runID, run.ParentID); err != nil {
+		return err
+	}
+	if pause.Prompt != "" {
+		fmt.Fprintf(w, "  Prompt: %s\n", pause.Prompt)
+	}
+	fmt.Fprintf(w, "  Recorded input for agent %q run %s.\n", name, runID)
+	fmt.Fprintf(w, "  Inspect: micro inspect agent %s --limit 1\n", name)
+	return nil
+}
+
+func recordCLIResumeEvents(name, runID, parentID string) error {
+	now := time.Now()
+	scoped := store.Scope(store.DefaultStore, "agent", name)
+	events := []goagent.RunEvent{
+		{Time: now, RunID: runID, ParentID: parentID, Agent: name, Kind: "checkpoint", Name: "done", Status: "done"},
+		{Time: now.Add(time.Nanosecond), RunID: runID, ParentID: parentID, Agent: name, Kind: "done"},
+	}
+	for _, e := range events {
+		b, err := json.Marshal(e)
+		if err != nil {
+			return err
+		}
+		key := fmt.Sprintf("runs/%s/%020d-%s", runID, e.Time.UnixNano(), e.Kind)
+		if err := scoped.Write(&store.Record{Key: key, Value: b}); err != nil {
+			return err
+		}
+	}
+	return nil
+}

cmd/micro/cli/agent/agent_test.go

@@ -2,6 +2,7 @@ package agent
 
 import (
 	"bytes"
+	"context"
 	"encoding/json"
 	"strings"
 	"testing"
@@ -9,6 +10,8 @@ import (
 
 	goagent "go-micro.dev/v6/agent"
 	"go-micro.dev/v6/ai"
+	aiflow "go-micro.dev/v6/flow"
+	"go-micro.dev/v6/store"
 )
 
 func TestWriteRunIndexJSON(t *testing.T) {
@@ -89,7 +92,7 @@ func TestWriteRunIndexInputRequiredUsesResumeInput(t *testing.T) {
 		t.Fatal(err)
 	}
 	got := out.String()
-	for _, want := range []string{`micro agent history runner run-input`, `micro.AgentResumeInput(ctx, agent, "run-input", input)`} {
+	for _, want := range []string{`micro agent history runner run-input`, `micro agent resume-input runner run-input --input <text>`} {
 		if !strings.Contains(got, want) {
 			t.Fatalf("output missing %q:\n%s", want, got)
 		}
@@ -137,3 +140,41 @@ func TestWriteRunHistoryHumanAndJSON(t *testing.T) {
 		t.Fatalf("decoded events = %#v", got)
 	}
 }
+
+func TestResumeInputRunCompletesCheckpointAndInspectSummary(t *testing.T) {
+	oldStore := store.DefaultStore
+	store.DefaultStore = store.NewMemoryStore()
+	t.Cleanup(func() { store.DefaultStore = oldStore })
+
+	ctx := context.Background()
+	cp := aiflow.StoreCheckpoint(store.DefaultStore, "runner")
+	run := aiflow.Run{ID: "run-input", Flow: "runner", Status: "paused", State: aiflow.State{Stage: "input-required"}, Steps: []aiflow.StepRecord{{Name: "ask", Status: "paused", Error: "Which region?"}}}
+	if err := run.State.Set(cliInputPause{OriginalMessage: "deploy", Prompt: "Which region?"}); err != nil {
+		t.Fatalf("set pause: %v", err)
+	}
+	if err := cp.Save(ctx, run); err != nil {
+		t.Fatalf("save checkpoint: %v", err)
+	}
+
+	var out bytes.Buffer
+	if err := resumeInputRun(ctx, &out, "runner", "run-input", "us-east-1"); err != nil {
+		t.Fatalf("resumeInputRun: %v", err)
+	}
+	if got := out.String(); !strings.Contains(got, "Recorded input") || !strings.Contains(got, "micro inspect agent runner --limit 1") {
+		t.Fatalf("output missing continuation hints:\n%s", got)
+	}
+	loaded, ok, err := cp.Load(ctx, "run-input")
+	if err != nil || !ok {
+		t.Fatalf("load checkpoint ok=%v err=%v", ok, err)
+	}
+	if loaded.Status != "done" || loaded.State.Stage != "done" {
+		t.Fatalf("loaded run status/stage = %s/%s, want done/done", loaded.Status, loaded.State.Stage)
+	}
+	summaries, err := goagent.ListRunSummariesWithOptions(store.DefaultStore, "runner", goagent.RunListOptions{Status: "done"})
+	if err != nil {
+		t.Fatalf("summaries: %v", err)
+	}
+	if len(summaries) != 1 || summaries[0].RunID != "run-input" || summaries[0].Status != "done" {
+		t.Fatalf("summaries = %#v, want completed run-input", summaries)
+	}
+}

cmd/micro/inspect/inspect.go

@@ -43,7 +43,7 @@ It reads durable local run history, so it works after the agent or flow has stop
 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: "status", Usage: "Only show runs with this status (running, done, canceled, timeout, rate_limited, auth, configuration, unavailable, provider_error, 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"},
 	}
@@ -91,6 +91,12 @@ func writeAgentInspection(w io.Writer, name string, runs []goagent.RunSummary, a
 		if run.Stage != "" {
 			fmt.Fprintf(w, "  stage=%s", run.Stage)
 		}
+		if run.LastErrorKind != "" {
+			fmt.Fprintf(w, "  error_kind=%s", run.LastErrorKind)
+		}
+		if run.Spent > 0 {
+			fmt.Fprintf(w, "  spent=%d", run.Spent)
+		}
 		if run.LastError != "" {
 			fmt.Fprintf(w, "  error=%q", run.LastError)
 		}
@@ -106,7 +112,7 @@ func writeAgentInspection(w io.Writer, name string, runs []goagent.RunSummary, a
 func writeAgentRunBreadcrumbs(w io.Writer, name string, run goagent.RunSummary) {
 	if run.Stage == "input-required" {
 		fmt.Fprintf(w, "    inspect: micro agent history %s %s\n", name, run.RunID)
-		fmt.Fprintf(w, "    input:   call micro.AgentResumeInput(ctx, agent, %q, input) to continue the input-required run\n", run.RunID)
+		fmt.Fprintf(w, "    input:   micro agent resume-input %s %s --input <text>\n", name, run.RunID)
 		return
 	}
 	if !isResumableAgentRun(run) {

cmd/micro/inspect/inspect_test.go

@@ -11,13 +11,13 @@ import (
 )
 
 func TestWriteAgentInspectionIncludesActionableBreadcrumbs(t *testing.T) {
-	runs := []goagent.RunSummary{{RunID: "run-1", Status: "error", Events: 4, LastKind: "tool", LastError: "boom", TraceID: "1234567890abcdef", Checkpoint: "failed", Stage: "ask"}}
+	runs := []goagent.RunSummary{{RunID: "run-1", Status: "auth", Events: 4, LastKind: "model", LastError: "invalid API key", LastErrorKind: "auth", TraceID: "1234567890abcdef", Checkpoint: "failed", Stage: "ask", Spent: 7}}
 	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", "checkpoint=failed", "stage=ask", `error="boom"`, "trace=1234567890ab", `micro agent history support run-1`, `micro.AgentResume(ctx, agent, "run-1")`, `micro.ResumeStreamAsk(ctx, agent, "run-1")`} {
+	for _, want := range []string{"Agent \"support\" runs", "run-1", "status=auth", "events=4", "last=model", "checkpoint=failed", "stage=ask", "error_kind=auth", `error="invalid API key"`, "trace=1234567890ab", "spent=7"} {
 		if !strings.Contains(got, want) {
 			t.Fatalf("output missing %q:\n%s", want, got)
 		}
@@ -31,7 +31,7 @@ func TestWriteAgentInspectionIncludesInputResumeBreadcrumb(t *testing.T) {
 		t.Fatal(err)
 	}
 	got := out.String()
-	for _, want := range []string{"checkpoint=paused", "stage=input-required", `micro agent history support run-input`, `micro.AgentResumeInput(ctx, agent, "run-input", input)`} {
+	for _, want := range []string{"checkpoint=paused", "stage=input-required", `micro agent history support run-input`, `micro agent resume-input support run-input --input <text>`} {
 		if !strings.Contains(got, want) {
 			t.Fatalf("output missing %q:\n%s", want, got)
 		}

deploy/kubernetes/README.md

@@ -0,0 +1,31 @@
+# Kubernetes deployment foundation (alpha)
+
+This package is the first opt-in Kubernetes foundation for the Go Micro lifecycle:
+`Service`, `Agent`, and `Flow` resources. It is intentionally experimental and
+additive. Nothing in the Go Micro runtime installs these resources or changes
+production defaults.
+
+## What is included
+
+- Alpha CRD manifests in `config/crd/` for `agents.micro.dev`,
+  `services.micro.dev`, and `flows.micro.dev`.
+- A small dependency-free mapper that turns a desired Go Micro resource into the
+  Kubernetes `Deployment` shape an operator reconciliation loop will own.
+- Unit tests that validate the structural CRD fragments and dry-run the
+  Agent-to-Deployment mapping.
+
+## Local validation
+
+```sh
+go test ./deploy/kubernetes
+```
+
+If you have a Kubernetes cluster and `kubectl` available, you can also perform a
+server-side dry run of the CRDs:
+
+```sh
+kubectl apply --dry-run=server -f deploy/kubernetes/config/crd/
+```
+
+The manifests are `v1alpha1`; expect the API shape to evolve before this becomes
+a production operator.

deploy/kubernetes/config/crd/agent.yaml

@@ -0,0 +1,37 @@
+apiVersion: apiextensions.k8s.io/v1
+kind: CustomResourceDefinition
+metadata:
+  name: agents.micro.dev
+spec:
+  group: micro.dev
+  scope: Namespaced
+  names:
+    plural: agents
+    singular: agent
+    kind: Agent
+    shortNames: [magent]
+  versions:
+  - name: v1alpha1
+    served: true
+    storage: true
+    schema:
+      openAPIV3Schema:
+        type: object
+        required: [spec]
+        properties:
+          spec:
+            type: object
+            required: [image]
+            properties:
+              image: {type: string, minLength: 1}
+              command:
+                type: array
+                items: {type: string}
+              args:
+                type: array
+                items: {type: string}
+              replicas: {type: integer, minimum: 0}
+              registry: {type: string}
+              env:
+                type: object
+                additionalProperties: {type: string}

deploy/kubernetes/config/crd/flow.yaml

@@ -0,0 +1,37 @@
+apiVersion: apiextensions.k8s.io/v1
+kind: CustomResourceDefinition
+metadata:
+  name: flows.micro.dev
+spec:
+  group: micro.dev
+  scope: Namespaced
+  names:
+    plural: flows
+    singular: flow
+    kind: Flow
+    shortNames: [mflow]
+  versions:
+  - name: v1alpha1
+    served: true
+    storage: true
+    schema:
+      openAPIV3Schema:
+        type: object
+        required: [spec]
+        properties:
+          spec:
+            type: object
+            required: [image]
+            properties:
+              image: {type: string, minLength: 1}
+              command:
+                type: array
+                items: {type: string}
+              args:
+                type: array
+                items: {type: string}
+              replicas: {type: integer, minimum: 0}
+              registry: {type: string}
+              env:
+                type: object
+                additionalProperties: {type: string}

deploy/kubernetes/config/crd/service.yaml

@@ -0,0 +1,37 @@
+apiVersion: apiextensions.k8s.io/v1
+kind: CustomResourceDefinition
+metadata:
+  name: services.micro.dev
+spec:
+  group: micro.dev
+  scope: Namespaced
+  names:
+    plural: services
+    singular: service
+    kind: Service
+    shortNames: [mservice]
+  versions:
+  - name: v1alpha1
+    served: true
+    storage: true
+    schema:
+      openAPIV3Schema:
+        type: object
+        required: [spec]
+        properties:
+          spec:
+            type: object
+            required: [image]
+            properties:
+              image: {type: string, minLength: 1}
+              command:
+                type: array
+                items: {type: string}
+              args:
+                type: array
+                items: {type: string}
+              replicas: {type: integer, minimum: 0}
+              registry: {type: string}
+              env:
+                type: object
+                additionalProperties: {type: string}

deploy/kubernetes/doc.go

@@ -0,0 +1,8 @@
+// Package kubernetes contains the experimental Kubernetes deployment foundation
+// for Go Micro services, agents, and flows.
+//
+// The package is intentionally small and additive: it exposes alpha custom
+// resource manifests and a dry-run mapper that turns a resource spec into the
+// Deployment shape an operator would reconcile. It does not install an operator
+// or change any runtime defaults.
+package kubernetes

deploy/kubernetes/kubernetes_test.go

@@ -0,0 +1,87 @@
+package kubernetes
+
+import (
+	"strings"
+	"testing"
+)
+
+func TestCRDManifestsAreStructural(t *testing.T) {
+	for _, kind := range []Kind{KindAgent, KindService, KindFlow} {
+		manifest := CRDManifests[kind]
+		if manifest == "" {
+			t.Fatalf("missing manifest for %s", kind)
+		}
+		checks := []string{
+			"apiVersion: apiextensions.k8s.io/v1",
+			"kind: CustomResourceDefinition",
+			"group: micro.dev",
+			"kind: " + string(kind),
+			"name: v1alpha1",
+			"served: true",
+			"storage: true",
+			"openAPIV3Schema:",
+			"type: object",
+			"required: [image]",
+		}
+		for _, check := range checks {
+			if !strings.Contains(manifest, check) {
+				t.Fatalf("%s manifest missing %q:\n%s", kind, check, manifest)
+			}
+		}
+	}
+}
+
+func TestMapDeploymentForAgent(t *testing.T) {
+	deployment, err := MapDeployment(Resource{
+		Kind:      KindAgent,
+		Name:      "support-agent",
+		Namespace: "agents",
+		Spec: WorkloadSpec{
+			Image:    "ghcr.io/acme/support-agent:v1",
+			Replicas: 2,
+			Registry: "kubernetes",
+			Environment: map[string]string{
+				"MODEL": "gpt-5.5",
+			},
+		},
+	})
+	if err != nil {
+		t.Fatalf("MapDeployment returned error: %v", err)
+	}
+	if deployment.Name != "support-agent" || deployment.Namespace != "agents" {
+		t.Fatalf("unexpected identity: %+v", deployment)
+	}
+	if deployment.Replicas != 2 {
+		t.Fatalf("replicas = %d, want 2", deployment.Replicas)
+	}
+	if got := deployment.Labels["micro.dev/kind"]; got != "agent" {
+		t.Fatalf("micro.dev/kind label = %q, want agent", got)
+	}
+	container := deployment.Pod.Container
+	if container.Image != "ghcr.io/acme/support-agent:v1" {
+		t.Fatalf("image = %q", container.Image)
+	}
+	if got := container.Environment["MICRO_REGISTRY"]; got != "kubernetes" {
+		t.Fatalf("MICRO_REGISTRY = %q, want kubernetes", got)
+	}
+	if got := container.Environment["MODEL"]; got != "gpt-5.5" {
+		t.Fatalf("MODEL = %q, want gpt-5.5", got)
+	}
+}
+
+func TestMapDeploymentDefaultsAndValidation(t *testing.T) {
+	deployment, err := MapDeployment(Resource{Kind: KindService, Name: "api", Spec: WorkloadSpec{Image: "api:latest"}})
+	if err != nil {
+		t.Fatalf("MapDeployment returned error: %v", err)
+	}
+	if deployment.Namespace != "default" || deployment.Replicas != 1 {
+		t.Fatalf("defaults = namespace %q replicas %d", deployment.Namespace, deployment.Replicas)
+	}
+
+	if _, err := MapDeployment(Resource{Kind: KindFlow, Name: "ingest"}); err == nil {
+		t.Fatal("MapDeployment without image succeeded")
+	}
+	if _, err := MapDeployment(Resource{Kind: "Job", Name: "job", Spec: WorkloadSpec{Image: "job:latest"}}); err == nil {
+		t.Fatal("MapDeployment with unsupported kind succeeded")
+	}
+}

deploy/kubernetes/manifests.go

@@ -0,0 +1,125 @@
+package kubernetes
+
+// CRDManifests contains the alpha CRDs for Go Micro lifecycle resources.
+var CRDManifests = map[Kind]string{
+	KindAgent:   agentCRD,
+	KindService: serviceCRD,
+	KindFlow:    flowCRD,
+}
+
+const agentCRD = `apiVersion: apiextensions.k8s.io/v1
+kind: CustomResourceDefinition
+metadata:
+  name: agents.micro.dev
+spec:
+  group: micro.dev
+  scope: Namespaced
+  names:
+    plural: agents
+    singular: agent
+    kind: Agent
+    shortNames: [magent]
+  versions:
+  - name: v1alpha1
+    served: true
+    storage: true
+    schema:
+      openAPIV3Schema:
+        type: object
+        required: [spec]
+        properties:
+          spec:
+            type: object
+            required: [image]
+            properties:
+              image: {type: string, minLength: 1}
+              command:
+                type: array
+                items: {type: string}
+              args:
+                type: array
+                items: {type: string}
+              replicas: {type: integer, minimum: 0}
+              registry: {type: string}
+              env:
+                type: object
+                additionalProperties: {type: string}
+`
+
+const serviceCRD = `apiVersion: apiextensions.k8s.io/v1
+kind: CustomResourceDefinition
+metadata:
+  name: services.micro.dev
+spec:
+  group: micro.dev
+  scope: Namespaced
+  names:
+    plural: services
+    singular: service
+    kind: Service
+    shortNames: [mservice]
+  versions:
+  - name: v1alpha1
+    served: true
+    storage: true
+    schema:
+      openAPIV3Schema:
+        type: object
+        required: [spec]
+        properties:
+          spec:
+            type: object
+            required: [image]
+            properties:
+              image: {type: string, minLength: 1}
+              command:
+                type: array
+                items: {type: string}
+              args:
+                type: array
+                items: {type: string}
+              replicas: {type: integer, minimum: 0}
+              registry: {type: string}
+              env:
+                type: object
+                additionalProperties: {type: string}
+`
+
+const flowCRD = `apiVersion: apiextensions.k8s.io/v1
+kind: CustomResourceDefinition
+metadata:
+  name: flows.micro.dev
+spec:
+  group: micro.dev
+  scope: Namespaced
+  names:
+    plural: flows
+    singular: flow
+    kind: Flow
+    shortNames: [mflow]
+  versions:
+  - name: v1alpha1
+    served: true
+    storage: true
+    schema:
+      openAPIV3Schema:
+        type: object
+        required: [spec]
+        properties:
+          spec:
+            type: object
+            required: [image]
+            properties:
+              image: {type: string, minLength: 1}
+              command:
+                type: array
+                items: {type: string}
+              args:
+                type: array
+                items: {type: string}
+              replicas: {type: integer, minimum: 0}
+              registry: {type: string}
+              env:
+                type: object
+                additionalProperties: {type: string}
+`

deploy/kubernetes/types.go

@@ -0,0 +1,138 @@
+package kubernetes
+
+import (
+	"fmt"
+	"sort"
+	"strings"
+)
+
+const (
+	// Group is the API group for the alpha Go Micro Kubernetes resources.
+	Group = "micro.dev"
+	// Version is the current alpha API version for the CRDs in this package.
+	Version = "v1alpha1"
+)
+
+// Kind identifies a Go Micro lifecycle resource that can be reconciled toward a
+// Kubernetes Deployment.
+type Kind string
+
+const (
+	KindAgent   Kind = "Agent"
+	KindService Kind = "Service"
+	KindFlow    Kind = "Flow"
+)
+
+// WorkloadSpec is the common alpha spec shared by Agent, Service, and Flow CRDs.
+type WorkloadSpec struct {
+	Image       string            `json:"image"`
+	Command     []string          `json:"command,omitempty"`
+	Args        []string          `json:"args,omitempty"`
+	Replicas    int32             `json:"replicas,omitempty"`
+	Registry    string            `json:"registry,omitempty"`
+	Environment map[string]string `json:"env,omitempty"`
+}
+
+// Resource is the minimal desired state for a Go Micro lifecycle resource.
+type Resource struct {
+	Kind      Kind
+	Name      string
+	Namespace string
+	Spec      WorkloadSpec
+}
+
+// Deployment is a small, dependency-free representation of the Kubernetes
+// Deployment fields the alpha reconciler skeleton owns.
+type Deployment struct {
+	Name      string
+	Namespace string
+	Labels    map[string]string
+	Replicas  int32
+	Pod       PodTemplate
+}
+
+// PodTemplate describes the pod fields emitted by MapDeployment.
+type PodTemplate struct {
+	Labels    map[string]string
+	Container Container
+}
+
+// Container describes the single Go Micro workload container.
+type Container struct {
+	Name        string
+	Image       string
+	Command     []string
+	Args        []string
+	Environment map[string]string
+}
+
+// MapDeployment maps a Go Micro alpha resource to the Deployment shape an
+// operator reconciliation loop would apply.
+func MapDeployment(resource Resource) (Deployment, error) {
+	if resource.Kind != KindAgent && resource.Kind != KindService && resource.Kind != KindFlow {
+		return Deployment{}, fmt.Errorf("unsupported kind %q", resource.Kind)
+	}
+	name := strings.TrimSpace(resource.Name)
+	if name == "" {
+		return Deployment{}, fmt.Errorf("name is required")
+	}
+	image := strings.TrimSpace(resource.Spec.Image)
+	if image == "" {
+		return Deployment{}, fmt.Errorf("spec.image is required")
+	}
+
+	namespace := strings.TrimSpace(resource.Namespace)
+	if namespace == "" {
+		namespace = "default"
+	}
+	replicas := resource.Spec.Replicas
+	if replicas == 0 {
+		replicas = 1
+	}
+
+	labels := map[string]string{
+		"app.kubernetes.io/name":       name,
+		"app.kubernetes.io/managed-by": "go-micro",
+		"micro.dev/kind":               strings.ToLower(string(resource.Kind)),
+	}
+	env := copyMap(resource.Spec.Environment)
+	if resource.Spec.Registry != "" {
+		env["MICRO_REGISTRY"] = resource.Spec.Registry
+	}
+
+	return Deployment{
+		Name:      name,
+		Namespace: namespace,
+		Labels:    copyMap(labels),
+		Replicas:  replicas,
+		Pod: PodTemplate{
+			Labels: copyMap(labels),
+			Container: Container{
+				Name:        name,
+				Image:       image,
+				Command:     append([]string(nil), resource.Spec.Command...),
+				Args:        append([]string(nil), resource.Spec.Args...),
+				Environment: env,
+			},
+		},
+	}, nil
+}
+
+// EnvironmentKeys returns stable environment variable keys from a mapped
+// container. It is useful for deterministic validation and rendering.
+func (c Container) EnvironmentKeys() []string {
+	keys := make([]string, 0, len(c.Environment))
+	for key := range c.Environment {
+		keys = append(keys, key)
+	}
+	sort.Strings(keys)
+	return keys
+}
+
+func copyMap(in map[string]string) map[string]string {
+	out := make(map[string]string, len(in))
+	for k, v := range in {
+		out[k] = v
+	}
+	return out
+}

examples/INDEX.md

@@ -12,6 +12,7 @@ provider-free unless the example README says otherwise.
 | Prove the maintained 0→hero path | [`support`](./support/) | `go run ./examples/support` and `go test ./examples/support` | [`zero-to-hero` guide](../internal/website/docs/guides/zero-to-hero.md) |
 | See planning and delegation | [`agent-plan-delegate`](./agent-plan-delegate/) | `go run ./examples/agent-plan-delegate` | [`plan-delegate` guide](../internal/website/docs/guides/plan-delegate.md) |
 | Expose services through MCP | [`mcp/hello`](./mcp/hello/) | follow [`mcp`](./mcp/) setup | [`mcp/crud`](./mcp/crud/) and [`mcp/workflow`](./mcp/workflow/) |
+| Try a paid tool with x402 | [`agent-x402-buyer`](./agent-x402-buyer/) | `go run ./examples/agent-x402-buyer` | [`Payments (x402)` guide](../internal/website/docs/guides/x402-payments.md) |
 | Try A2A or gRPC interop next | [`agent-demo`](./agent-demo/) plus gateway docs | run the example, then use the gateway docs | [`grpc-interop`](./grpc-interop/) |
 | Add workflow durability | [`flow-durable`](./flow-durable/) | `go run ./examples/flow-durable` | [`flow-loop`](./flow-loop/) |
 
@@ -28,7 +29,9 @@ provider-free unless the example README says otherwise.
 4. **Interop next:** use [`mcp/hello`](./mcp/hello/), [`mcp/crud`](./mcp/crud/),
    and [`mcp/workflow`](./mcp/workflow/) when you are ready to expose tools to
    external AI clients.
-5. **Workflow depth:** use [`flow-durable`](./flow-durable/) once the agent path
+5. **Paid tools:** run [`agent-x402-buyer`](./agent-x402-buyer/) to see an
+   agent pay a local x402-protected tool with a mock facilitator and budget.
+6. **Workflow depth:** use [`flow-durable`](./flow-durable/) once the agent path
    needs checkpointed, resumable deterministic work.
 
 ## CLI wayfinding

examples/agent-x402-buyer/README.md

@@ -0,0 +1,24 @@
+# Agent x402 buyer
+
+This example shows an agent paying for a paid HTTP tool with x402 without using
+live funds or a live chain.
+
+It starts a local paid endpoint guarded by `wrapper/x402` seller middleware and a
+mock facilitator. A deterministic mock-model agent calls that endpoint as a tool,
+receives the HTTP 402 challenge, pays with `AgentPayer`, stays inside
+`AgentBudget`, retries the request, and prints the spend recorded for the run.
+
+```bash
+go run ./examples/agent-x402-buyer
+```
+
+Expected output includes:
+
+- the paid tool response,
+- one facilitator verify and settle call, and
+- `run spend: 7 smallest units (budget 10)`.
+
+The payment token and facilitator are intentionally local development fakes. To
+settle real x402 payments, keep the same `AgentPayer` / `AgentBudget` shape but
+replace the payer with a wallet-backed implementation and configure the seller
+middleware with a hosted or self-run x402 facilitator.

examples/agent-x402-buyer/main.go

@@ -0,0 +1,168 @@
+// Agent x402 buyer — a provider-free example of an agent paying for a paid tool.
+//
+// Run:
+//
+//	go run ./examples/agent-x402-buyer
+//
+// It starts a local HTTP tool protected by x402 middleware, then asks a
+// deterministic mock-model agent to call that tool. The agent receives the 402
+// challenge, uses AgentPayer and AgentBudget to pay within a local mock
+// facilitator, retries the request, and prints the run spend.
+package main
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"io"
+	"net/http"
+	"net/http/httptest"
+	"os"
+	"strings"
+
+	go_micro "go-micro.dev/v6"
+	"go-micro.dev/v6/agent"
+	"go-micro.dev/v6/ai"
+	"go-micro.dev/v6/store"
+	"go-micro.dev/v6/wrapper/x402"
+)
+
+const (
+	paidToolName = "paid.market_brief"
+	price        = int64(7)
+	paymentToken = "dev-payment-token"
+)
+
+type devFacilitator struct {
+	verifyCount int
+	settleCount int
+}
+
+func (f *devFacilitator) Verify(ctx context.Context, payment string, req x402.Requirements) (x402.Result, error) {
+	f.verifyCount++
+	if payment != paymentToken {
+		return x402.Result{Valid: false, Reason: "unknown dev payment token"}, nil
+	}
+	return x402.Result{Valid: true, Payer: "dev-agent-wallet"}, nil
+}
+
+func (f *devFacilitator) Settle(ctx context.Context, payment string, req x402.Requirements) (x402.Result, error) {
+	f.settleCount++
+	return x402.Result{Valid: true, Settlement: "dev-settlement-001"}, nil
+}
+
+type devPayer struct{}
+
+func (devPayer) Pay(ctx context.Context, req x402.Requirements) (string, error) {
+	return paymentToken, nil
+}
+
+type mockModel struct{ opts ai.Options }
+
+func newMock(opts ...ai.Option) ai.Model {
+	m := &mockModel{}
+	_ = m.Init(opts...)
+	return m
+}
+
+func (m *mockModel) Init(opts ...ai.Option) error {
+	for _, o := range opts {
+		o(&m.opts)
+	}
+	return nil
+}
+func (m *mockModel) Options() ai.Options { return m.opts }
+func (m *mockModel) String() string      { return "agent-x402-buyer-mock" }
+func (m *mockModel) Stream(context.Context, *ai.Request, ...ai.GenerateOption) (ai.Stream, error) {
+	return nil, fmt.Errorf("stream not supported by agent-x402-buyer mock")
+}
+
+func (m *mockModel) Generate(ctx context.Context, req *ai.Request, _ ...ai.GenerateOption) (*ai.Response, error) {
+	for _, tool := range req.Tools {
+		if tool.Name == paidToolName && m.opts.ToolHandler != nil {
+			out := m.opts.ToolHandler(ctx, ai.ToolCall{ID: "paid-brief", Name: tool.Name, Input: map[string]any{"url": req.Prompt}})
+			return &ai.Response{Answer: fmt.Sprintf("Paid tool returned: %s", out.Content)}, nil
+		}
+	}
+	return &ai.Response{Answer: "No paid tool was available."}, nil
+}
+
+func paidToolServer(fac *devFacilitator) *httptest.Server {
+	mux := http.NewServeMux()
+	paid := x402.Middleware(x402.Config{
+		PayTo:       "0xMerchantDevWallet",
+		Network:     "base-sepolia",
+		Amount:      fmt.Sprint(price),
+		Description: "Local market brief for the x402 buyer example",
+		Facilitator: fac,
+	})
+	mux.Handle("/brief", paid(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		w.Header().Set("Content-Type", "application/json")
+		_ = json.NewEncoder(w).Encode(map[string]any{
+			"brief":      "Mock demand is up 12% after the agent paid the local tool.",
+			"settlement": w.Header().Get(x402.PaymentResponseHeader),
+		})
+	})))
+	return httptest.NewServer(mux)
+}
+
+func run(w io.Writer) error {
+	ai.Register("agent-x402-buyer-mock", newMock)
+
+	fac := &devFacilitator{}
+	srv := paidToolServer(fac)
+	defer srv.Close()
+
+	st := store.NewMemoryStore()
+	buyer := agent.New(
+		agent.Name("x402-buyer"),
+		agent.Provider("agent-x402-buyer-mock"),
+		agent.Prompt("Call the paid market brief tool when given its URL."),
+		agent.WithStore(st),
+		go_micro.AgentPayer(devPayer{}),
+		go_micro.AgentBudget(10),
+		agent.WithTool(paidToolName, "Fetch a paid market brief over HTTP", map[string]any{
+			"url": map[string]any{"type": "string", "description": "Paid HTTP endpoint to call"},
+		}, func(ctx context.Context, input map[string]any) (string, error) {
+			url, _ := input["url"].(string)
+			resp, err := http.Get(url)
+			if err != nil {
+				return "", err
+			}
+			defer resp.Body.Close()
+			body, err := io.ReadAll(resp.Body)
+			if err != nil {
+				return "", err
+			}
+			return string(body), nil
+		}),
+	)
+
+	resp, err := buyer.Ask(context.Background(), srv.URL+"/brief")
+	if err != nil {
+		return err
+	}
+	events, err := agent.LoadRunEvents(st, "x402-buyer", resp.RunID)
+	if err != nil {
+		return err
+	}
+	var spent int64
+	for _, event := range events {
+		if event.Spent > spent {
+			spent = event.Spent
+		}
+	}
+
+	fmt.Fprintln(w, "Agent x402 buyer (provider: mock, funds: local dev token)")
+	fmt.Fprintln(w, strings.TrimSpace(resp.Reply))
+	fmt.Fprintf(w, "facilitator verify=%d settle=%d\n", fac.verifyCount, fac.settleCount)
+	fmt.Fprintf(w, "run spend: %d smallest units (budget 10)\n", spent)
+	return nil
+}
+
+func main() {
+	if err := run(os.Stdout); err != nil {
+		fmt.Println(err)
+		os.Exit(1)
+	}
+}

gateway/mcp/grpcreflect.go

@@ -0,0 +1,282 @@
+package mcp
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"strings"
+	"time"
+
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/credentials/insecure"
+	reflectionpb "google.golang.org/grpc/reflection/grpc_reflection_v1alpha"
+	"google.golang.org/protobuf/encoding/protojson"
+	"google.golang.org/protobuf/proto"
+	"google.golang.org/protobuf/reflect/protodesc"
+	"google.golang.org/protobuf/reflect/protoreflect"
+	"google.golang.org/protobuf/reflect/protoregistry"
+	"google.golang.org/protobuf/types/descriptorpb"
+	"google.golang.org/protobuf/types/dynamicpb"
+)
+
+// ReflectedGRPCTarget describes an external gRPC server whose reflection
+// catalog should be exposed as MCP tools. It is intentionally opt-in: teams can
+// bridge existing reflected gRPC services without changing their servers or
+// registering them in go-micro.
+type ReflectedGRPCTarget struct {
+	// Name prefixes generated tools. When empty, Address is sanitized and used.
+	Name string
+	// Address is the host:port of the reflected gRPC server.
+	Address string
+	// DialOptions customize the connection. If none are supplied, an insecure
+	// transport is used for local/dev interoperability.
+	DialOptions []grpc.DialOption
+	// Timeout bounds reflection discovery and individual tool calls.
+	Timeout time.Duration
+}
+
+func (s *Server) discoverReflectedGRPC() error {
+	for _, target := range s.opts.ReflectedGRPCTargets {
+		if strings.TrimSpace(target.Address) == "" {
+			continue
+		}
+		tools, err := s.reflectedGRPCTools(target)
+		if err != nil {
+			return err
+		}
+		for _, tool := range tools {
+			s.tools[tool.Name] = tool
+		}
+	}
+	return nil
+}
+
+func (s *Server) reflectedGRPCTools(target ReflectedGRPCTarget) ([]*Tool, error) {
+	timeout := target.Timeout
+	if timeout == 0 {
+		timeout = 10 * time.Second
+	}
+	ctx, cancel := context.WithTimeout(s.opts.Context, timeout)
+	defer cancel()
+
+	dialOpts := target.DialOptions
+	if len(dialOpts) == 0 {
+		dialOpts = []grpc.DialOption{grpc.WithTransportCredentials(insecure.NewCredentials())}
+	}
+	conn, err := grpc.NewClient(target.Address, dialOpts...)
+	if err != nil {
+		return nil, fmt.Errorf("connect reflected grpc target %s: %w", target.Address, err)
+	}
+	defer conn.Close()
+
+	files, services, err := loadReflectedFiles(ctx, conn)
+	if err != nil {
+		return nil, fmt.Errorf("reflect grpc target %s: %w", target.Address, err)
+	}
+
+	prefix := target.Name
+	if prefix == "" {
+		prefix = sanitizeToolPart(target.Address)
+	}
+
+	var out []*Tool
+	for _, serviceName := range services {
+		desc, err := files.FindDescriptorByName(protoreflect.FullName(serviceName))
+		if err != nil {
+			continue
+		}
+		svc, ok := desc.(protoreflect.ServiceDescriptor)
+		if !ok {
+			continue
+		}
+		for i := 0; i < svc.Methods().Len(); i++ {
+			method := svc.Methods().Get(i)
+			if method.IsStreamingClient() || method.IsStreamingServer() {
+				continue
+			}
+			fullMethod := "/" + string(svc.FullName()) + "/" + string(method.Name())
+			toolName := prefix + "." + strings.ReplaceAll(string(svc.FullName()), ".", "_") + "." + string(method.Name())
+			input := method.Input()
+			out = append(out, &Tool{
+				Name:        toolName,
+				Description: fmt.Sprintf("Call reflected gRPC method %s on %s", fullMethod, target.Address),
+				InputSchema: protoMessageSchema(input),
+				Handler:     reflectedGRPCHandler(target, fullMethod, input, method.Output()),
+			})
+		}
+	}
+	return out, nil
+}
+
+func loadReflectedFiles(ctx context.Context, conn *grpc.ClientConn) (*protoregistryFiles, []string, error) {
+	client := reflectionpb.NewServerReflectionClient(conn)
+	stream, err := client.ServerReflectionInfo(ctx)
+	if err != nil {
+		return nil, nil, err
+	}
+	if err := stream.Send(&reflectionpb.ServerReflectionRequest{MessageRequest: &reflectionpb.ServerReflectionRequest_ListServices{ListServices: ""}}); err != nil {
+		return nil, nil, err
+	}
+	resp, err := stream.Recv()
+	if err != nil {
+		return nil, nil, err
+	}
+	list := resp.GetListServicesResponse()
+	if list == nil {
+		return nil, nil, fmt.Errorf("reflection list services returned %T", resp.MessageResponse)
+	}
+
+	set := &descriptorpb.FileDescriptorSet{}
+	seen := map[string]bool{}
+	var services []string
+	for _, svc := range list.Service {
+		name := svc.Name
+		if strings.HasPrefix(name, "grpc.reflection.") {
+			continue
+		}
+		services = append(services, name)
+		if err := requestFileContainingSymbol(ctx, client, name, set, seen); err != nil {
+			return nil, nil, err
+		}
+	}
+	files, err := newProtoregistryFiles(set)
+	if err != nil {
+		return nil, nil, err
+	}
+	return files, services, nil
+}
+
+func requestFileContainingSymbol(ctx context.Context, client reflectionpb.ServerReflectionClient, symbol string, set *descriptorpb.FileDescriptorSet, seen map[string]bool) error {
+	stream, err := client.ServerReflectionInfo(ctx)
+	if err != nil {
+		return err
+	}
+	if err := stream.Send(&reflectionpb.ServerReflectionRequest{MessageRequest: &reflectionpb.ServerReflectionRequest_FileContainingSymbol{FileContainingSymbol: symbol}}); err != nil {
+		return err
+	}
+	resp, err := stream.Recv()
+	if err != nil {
+		return err
+	}
+	fd := resp.GetFileDescriptorResponse()
+	if fd == nil {
+		return fmt.Errorf("reflection lookup for %s returned %T", symbol, resp.MessageResponse)
+	}
+	for _, raw := range fd.FileDescriptorProto {
+		var file descriptorpb.FileDescriptorProto
+		if err := proto.Unmarshal(raw, &file); err != nil {
+			return err
+		}
+		name := file.GetName()
+		if !seen[name] {
+			seen[name] = true
+			set.File = append(set.File, &file)
+		}
+	}
+	return nil
+}
+
+// protoregistryFiles is a narrow wrapper that keeps imports local to this file.
+type protoregistryFiles struct{ files *protoregistry.Files }
+
+func newProtoregistryFiles(set *descriptorpb.FileDescriptorSet) (*protoregistryFiles, error) {
+	files, err := protodesc.NewFiles(set)
+	if err != nil {
+		return nil, err
+	}
+	return &protoregistryFiles{files: files}, nil
+}
+
+func (p *protoregistryFiles) FindDescriptorByName(name protoreflect.FullName) (protoreflect.Descriptor, error) {
+	return p.files.FindDescriptorByName(name)
+}
+
+func reflectedGRPCHandler(target ReflectedGRPCTarget, fullMethod string, input, output protoreflect.MessageDescriptor) func(map[string]interface{}) (interface{}, error) {
+	return func(args map[string]interface{}) (interface{}, error) {
+		timeout := target.Timeout
+		if timeout == 0 {
+			timeout = 10 * time.Second
+		}
+		ctx, cancel := context.WithTimeout(context.Background(), timeout)
+		defer cancel()
+		dialOpts := target.DialOptions
+		if len(dialOpts) == 0 {
+			dialOpts = []grpc.DialOption{grpc.WithTransportCredentials(insecure.NewCredentials())}
+		}
+		conn, err := grpc.NewClient(target.Address, dialOpts...)
+		if err != nil {
+			return nil, err
+		}
+		defer conn.Close()
+
+		req := dynamicpb.NewMessage(input)
+		raw, err := json.Marshal(args)
+		if err != nil {
+			return nil, err
+		}
+		if err := protojson.Unmarshal(raw, req); err != nil {
+			return nil, err
+		}
+		rsp := dynamicpb.NewMessage(output)
+		if err := conn.Invoke(ctx, fullMethod, req, rsp); err != nil {
+			return nil, err
+		}
+		b, err := protojson.MarshalOptions{UseProtoNames: true, EmitUnpopulated: true}.Marshal(rsp)
+		if err != nil {
+			return nil, err
+		}
+		var out interface{}
+		if err := json.Unmarshal(b, &out); err != nil {
+			return nil, err
+		}
+		return out, nil
+	}
+}
+
+func protoMessageSchema(msg protoreflect.MessageDescriptor) map[string]interface{} {
+	schema := map[string]interface{}{"type": "object", "properties": map[string]interface{}{}}
+	props := schema["properties"].(map[string]interface{})
+	fields := msg.Fields()
+	for i := 0; i < fields.Len(); i++ {
+		field := fields.Get(i)
+		props[field.JSONName()] = protoFieldSchema(field)
+	}
+	return schema
+}
+
+func protoFieldSchema(field protoreflect.FieldDescriptor) map[string]interface{} {
+	schema := map[string]interface{}{"type": protoJSONType(field)}
+	if field.IsList() {
+		schema["items"] = map[string]interface{}{"type": protoJSONType(field)}
+	}
+	if field.Kind() == protoreflect.MessageKind || field.Kind() == protoreflect.GroupKind {
+		schema = protoMessageSchema(field.Message())
+	}
+	return schema
+}
+
+func protoJSONType(field protoreflect.FieldDescriptor) string {
+	if field.IsList() {
+		return "array"
+	}
+	switch field.Kind() {
+	case protoreflect.BoolKind:
+		return "boolean"
+	case protoreflect.Int32Kind, protoreflect.Sint32Kind, protoreflect.Sfixed32Kind,
+		protoreflect.Uint32Kind, protoreflect.Fixed32Kind, protoreflect.Int64Kind,
+		protoreflect.Sint64Kind, protoreflect.Sfixed64Kind, protoreflect.Uint64Kind,
+		protoreflect.Fixed64Kind:
+		return "integer"
+	case protoreflect.FloatKind, protoreflect.DoubleKind:
+		return "number"
+	case protoreflect.MessageKind, protoreflect.GroupKind:
+		return "object"
+	default:
+		return "string"
+	}
+}
+
+func sanitizeToolPart(s string) string {
+	r := strings.NewReplacer(":", "_", "/", "_", ".", "_", "-", "_")
+	return r.Replace(s)
+}

gateway/mcp/grpcreflect_test.go

@@ -0,0 +1,62 @@
+package mcp
+
+import (
+	"context"
+	"net"
+	"testing"
+	"time"
+
+	"google.golang.org/grpc"
+	helloworld "google.golang.org/grpc/examples/helloworld/helloworld"
+	"google.golang.org/grpc/reflection"
+)
+
+type reflectedGreeter struct {
+	helloworld.UnimplementedGreeterServer
+}
+
+func (reflectedGreeter) SayHello(_ context.Context, req *helloworld.HelloRequest) (*helloworld.HelloReply, error) {
+	return &helloworld.HelloReply{Message: "hello " + req.Name}, nil
+}
+
+func TestReflectedGRPCTargetDiscoversAndCallsUnaryTool(t *testing.T) {
+	lis, err := net.Listen("tcp", "127.0.0.1:0")
+	if err != nil {
+		t.Fatal(err)
+	}
+	grpcServer := grpc.NewServer()
+	helloworld.RegisterGreeterServer(grpcServer, reflectedGreeter{})
+	reflection.Register(grpcServer)
+	go grpcServer.Serve(lis)
+	defer grpcServer.Stop()
+
+	s := newTestServer(Options{Context: context.Background()})
+	tools, err := s.reflectedGRPCTools(ReflectedGRPCTarget{
+		Name:    "demo",
+		Address: lis.Addr().String(),
+		Timeout: 3 * time.Second,
+	})
+	if err != nil {
+		t.Fatalf("discover reflected tools: %v", err)
+	}
+	if len(tools) != 1 {
+		t.Fatalf("tools len = %d, want 1", len(tools))
+	}
+	tool := tools[0]
+	if tool.Name != "demo.helloworld_Greeter.SayHello" {
+		t.Fatalf("tool name = %q", tool.Name)
+	}
+	props := tool.InputSchema["properties"].(map[string]interface{})
+	if _, ok := props["name"]; !ok {
+		t.Fatalf("input schema missing name: %#v", tool.InputSchema)
+	}
+
+	out, err := tool.Handler(map[string]interface{}{"name": "Ada"})
+	if err != nil {
+		t.Fatalf("call reflected tool: %v", err)
+	}
+	got := out.(map[string]interface{})["message"]
+	if got != "hello Ada" {
+		t.Fatalf("message = %v, want hello Ada", got)
+	}
+}

gateway/mcp/mcp.go

@@ -157,6 +157,11 @@ type Options struct {
 	// (the /mcp/call endpoint). Listing tools and health stay free.
 	// Opt-in: leave nil to disable payments.
 	Payment *x402.Config
+
+	// ReflectedGRPCTargets exposes unary methods from external gRPC servers
+	// that support server reflection as MCP tools. This bridges existing gRPC
+	// services into the agent tool catalog without requiring go-micro handlers.
+	ReflectedGRPCTargets []ReflectedGRPCTarget
 }
 
 // Server represents a running MCP gateway
@@ -286,6 +291,10 @@ func (s *Server) discoverServices() error {
 	s.toolsMu.Lock()
 	defer s.toolsMu.Unlock()
 
+	if err := s.discoverReflectedGRPC(); err != nil {
+		return err
+	}
+
 	for _, svc := range services {
 		// Get full service details
 		fullSvcs, err := s.opts.Registry.GetService(svc.Name)

gateway/mcp/stdio.go

@@ -294,7 +294,9 @@ func (t *StdioTransport) handleToolsCall(req *JSONRPCRequest) {
 			AccountID: accountID, ScopesRequired: tool.Scopes,
 			Allowed: true, Duration: time.Since(start), Error: err.Error(),
 		})
-		t.sendError(req.ID, InternalError, "RPC call failed", err.Error())
+		// A tool-execution failure is reported as an isError result, not a
+		// JSON-RPC protocol error (per the MCP spec), so the agent can read it.
+		t.sendResponse(req.ID, mcpToolError(traceID, "tool call failed: "+err.Error()))
 		return
 	}
 
@@ -311,24 +313,8 @@ func (t *StdioTransport) handleToolsCall(req *JSONRPCRequest) {
 		Allowed: true, Duration: time.Since(start),
 	})
 
-	// Parse response
-	var result interface{}
-	if err := json.Unmarshal(rsp.Data, &result); err != nil {
-		// If unmarshal fails, return raw data
-		result = map[string]interface{}{
-			"data": string(rsp.Data),
-		}
-	}
-
-	t.sendResponse(req.ID, map[string]interface{}{
-		"content": []interface{}{
-			map[string]interface{}{
-				"type": "text",
-				"text": fmt.Sprintf("%v", result),
-			},
-		},
-		"trace_id": traceID,
-	})
+	// The downstream response is JSON — return it as JSON text, not %v.
+	t.sendResponse(req.ID, mcpToolResult(traceID, rsp.Data))
 }
 
 // sendResponse sends a JSON-RPC response

gateway/mcp/stdio_test.go

@@ -0,0 +1,120 @@
+package mcp
+
+import (
+	"bufio"
+	"bytes"
+	"context"
+	"encoding/json"
+	"errors"
+	"testing"
+
+	"go-micro.dev/v6/client"
+)
+
+// fakeCallClient overrides Call to return canned data or an error; NewRequest
+// and the rest are promoted from the embedded real client.
+type fakeCallClient struct {
+	client.Client
+	data []byte
+	err  error
+}
+
+func (f *fakeCallClient) Call(ctx context.Context, req client.Request, rsp interface{}, opts ...client.CallOption) error {
+	if f.err != nil {
+		return f.err
+	}
+	if r, ok := rsp.(*struct{ Data []byte }); ok {
+		r.Data = f.data
+	}
+	return nil
+}
+
+// isToolError reports whether an MCP tools/call result carries isError:true.
+func isToolError(result interface{}) bool {
+	m, ok := result.(map[string]interface{})
+	if !ok {
+		return false
+	}
+	b, _ := m["isError"].(bool)
+	return b
+}
+
+// toolResultText extracts the first text content of an MCP tools/call result.
+func toolResultText(t *testing.T, result interface{}) string {
+	t.Helper()
+	m, ok := result.(map[string]interface{})
+	if !ok {
+		t.Fatalf("result is not a map: %#v", result)
+	}
+	content, ok := m["content"].([]interface{})
+	if !ok || len(content) == 0 {
+		t.Fatalf("result has no content: %#v", result)
+	}
+	first, _ := content[0].(map[string]interface{})
+	text, _ := first["text"].(string)
+	return text
+}
+
+// driveStdio sends one JSON-RPC request through a StdioTransport and returns the
+// decoded response, capturing the transport's stdout into a buffer.
+func driveStdio(t *testing.T, s *Server, method string, id interface{}, params interface{}) JSONRPCResponse {
+	t.Helper()
+	tr := NewStdioTransport(s)
+	var out bytes.Buffer
+	tr.writer = bufio.NewWriter(&out)
+	raw, _ := json.Marshal(params)
+	tr.handleRequest(&JSONRPCRequest{JSONRPC: "2.0", ID: id, Method: method, Params: raw})
+	var resp JSONRPCResponse
+	if err := json.Unmarshal(bytes.TrimSpace(out.Bytes()), &resp); err != nil {
+		t.Fatalf("decode stdio response: %v (raw=%q)", err, out.String())
+	}
+	return resp
+}
+
+// The stdio transport is the path an external MCP host (Claude Desktop) uses.
+// It must return tool output as JSON text, not fmt.Sprintf("%v", ...) which
+// yields Go map-syntax and is unparseable by a real client.
+func TestStdio_ToolsCall_ReturnsJSONNotGoSyntax(t *testing.T) {
+	s := newTestServer(Options{})
+	s.opts.Client = &fakeCallClient{Client: client.DefaultClient, data: []byte(`{"id":1,"name":"bob"}`)}
+	s.tools["svc.Echo"] = &Tool{Name: "svc.Echo", Service: "svc", Endpoint: "Echo"}
+
+	resp := driveStdio(t, s, "tools/call", 1, map[string]interface{}{
+		"name":      "svc.Echo",
+		"arguments": map[string]interface{}{"msg": "hi"},
+	})
+	if resp.Error != nil {
+		t.Fatalf("unexpected protocol error: %+v", resp.Error)
+	}
+	text := toolResultText(t, resp.Result)
+	// The bug returned Go map-syntax ("map[id:1 name:bob]"), which fails to parse.
+	var got map[string]interface{}
+	if err := json.Unmarshal([]byte(text), &got); err != nil {
+		t.Fatalf("tool result text is not JSON (the %%v bug): %q", text)
+	}
+	if got["name"] != "bob" {
+		t.Errorf("result = %v, want name=bob", got)
+	}
+}
+
+// A tool-execution failure must be an MCP isError result, not a JSON-RPC
+// protocol error, so the agent can read the failure.
+func TestStdio_ToolsCall_FailureIsIsErrorResult(t *testing.T) {
+	s := newTestServer(Options{})
+	s.opts.Client = &fakeCallClient{Client: client.DefaultClient, err: errors.New("backend down")}
+	s.tools["svc.Echo"] = &Tool{Name: "svc.Echo", Service: "svc", Endpoint: "Echo"}
+
+	resp := driveStdio(t, s, "tools/call", 1, map[string]interface{}{
+		"name":      "svc.Echo",
+		"arguments": map[string]interface{}{},
+	})
+	if resp.Error != nil {
+		t.Fatalf("tool failure returned a protocol error, want isError result: %+v", resp.Error)
+	}
+	if !isToolError(resp.Result) {
+		t.Fatalf("expected isError result, got %+v", resp.Result)
+	}
+	if text := toolResultText(t, resp.Result); text == "" {
+		t.Error("isError result should carry the error text")
+	}
+}

gateway/mcp/toolresult.go

@@ -0,0 +1,32 @@
+package mcp
+
+// MCP tools/call result shaping, shared by the stdio and websocket JSON-RPC
+// transports. Kept in one place so both transports produce spec-shaped results.
+
+// mcpToolResult builds a successful MCP tools/call result. The downstream RPC
+// response body (data) is JSON, so it is returned as JSON text — NOT
+// fmt.Sprintf("%v", ...) of a decoded value, which produces Go map-syntax
+// (map[id:1 name:bob]) instead of JSON and is what an external MCP client
+// (e.g. Claude Desktop over stdio) would otherwise receive.
+func mcpToolResult(traceID string, data []byte) map[string]interface{} {
+	return map[string]interface{}{
+		"content": []interface{}{
+			map[string]interface{}{"type": "text", "text": string(data)},
+		},
+		"trace_id": traceID,
+	}
+}
+
+// mcpToolError builds an MCP tools/call result for a tool-EXECUTION failure.
+// Per the MCP spec a tool that fails returns a normal result with isError:true
+// (the error as text content), NOT a JSON-RPC protocol error — that way the
+// agent can read the failure instead of seeing a transport-level error.
+func mcpToolError(traceID, msg string) map[string]interface{} {
+	return map[string]interface{}{
+		"content": []interface{}{
+			map[string]interface{}{"type": "text", "text": msg},
+		},
+		"isError":  true,
+		"trace_id": traceID,
+	}
+}

gateway/mcp/websocket.go

@@ -275,7 +275,8 @@ func (wc *wsConn) handleToolsCall(req *JSONRPCRequest) {
 			AccountID: accountID, ScopesRequired: tool.Scopes,
 			Allowed: true, Duration: time.Since(start), Error: err.Error(),
 		})
-		wc.sendError(req.ID, InternalError, "RPC call failed", err.Error())
+		// Tool-execution failure → isError result (MCP spec), not a protocol error.
+		wc.sendResponse(req.ID, mcpToolError(traceID, "tool call failed: "+err.Error()))
 		return
 	}
 
@@ -291,23 +292,8 @@ func (wc *wsConn) handleToolsCall(req *JSONRPCRequest) {
 		Allowed: true, Duration: time.Since(start),
 	})
 
-	// Parse response
-	var result interface{}
-	if err := json.Unmarshal(rsp.Data, &result); err != nil {
-		result = map[string]interface{}{
-			"data": string(rsp.Data),
-		}
-	}
-
-	wc.sendResponse(req.ID, map[string]interface{}{
-		"content": []interface{}{
-			map[string]interface{}{
-				"type": "text",
-				"text": fmt.Sprintf("%v", result),
-			},
-		},
-		"trace_id": traceID,
-	})
+	// The downstream response is JSON — return it as JSON text, not %v.
+	wc.sendResponse(req.ID, mcpToolResult(traceID, rsp.Data))
 }
 
 // sendResponse sends a JSON-RPC success response.

gateway/mcp/websocket_test.go

@@ -114,12 +114,13 @@ func TestWebSocket_ToolsCall_NoAuth(t *testing.T) {
 		"arguments": map[string]interface{}{"msg": "hi"},
 	})
 
-	// RPC will fail (no backend), but auth should pass (no auth configured)
-	if resp.Error == nil {
-		t.Fatal("expected RPC error (no backend)")
+	// No auth required → the tool runs; the RPC fails (no backend), which the
+	// MCP spec surfaces as an isError result, not a JSON-RPC protocol error.
+	if resp.Error != nil {
+		t.Fatalf("expected no protocol error, got %+v", resp.Error)
 	}
-	if resp.Error.Code != InternalError {
-		t.Errorf("error code = %d, want %d", resp.Error.Code, InternalError)
+	if !isToolError(resp.Result) {
+		t.Fatalf("expected isError tool result, got %+v", resp.Result)
 	}
 }
 
@@ -168,12 +169,13 @@ func TestWebSocket_ToolsCall_AuthRequired(t *testing.T) {
 			"arguments": map[string]interface{}{},
 			"_token":    "valid-token",
 		})
-		// Auth passes, RPC fails (no backend)
-		if resp.Error == nil {
-			t.Fatal("expected RPC error")
+		// Auth passes → the tool runs; RPC fails (no backend) → isError result,
+		// not a JSON-RPC protocol error (which would mean auth failed).
+		if resp.Error != nil {
+			t.Fatalf("expected no protocol error (auth passed), got %+v", resp.Error)
 		}
-		if resp.Error.Code != InternalError {
-			t.Errorf("error code = %d, want %d (RPC fail, not auth fail)", resp.Error.Code, InternalError)
+		if !isToolError(resp.Result) {
+			t.Fatalf("expected isError tool result, got %+v", resp.Result)
 		}
 	})
 
@@ -185,12 +187,13 @@ func TestWebSocket_ToolsCall_AuthRequired(t *testing.T) {
 			"name":      "svc.Do",
 			"arguments": map[string]interface{}{},
 		})
-		// Auth passes via connection-level header, RPC fails (no backend)
-		if resp.Error == nil {
-			t.Fatal("expected RPC error")
+		// Auth passes via connection-level header → tool runs; RPC fails (no
+		// backend) → isError result, not a JSON-RPC protocol error.
+		if resp.Error != nil {
+			t.Fatalf("expected no protocol error (auth passed), got %+v", resp.Error)
 		}
-		if resp.Error.Code != InternalError {
-			t.Errorf("error code = %d, want %d (RPC fail, not auth fail)", resp.Error.Code, InternalError)
+		if !isToolError(resp.Result) {
+			t.Fatalf("expected isError tool result, got %+v", resp.Result)
 		}
 	})
 }

internal/harness/a2a-streaming/main.go

@@ -0,0 +1,259 @@
+// A2A streaming harness.
+//
+// It exercises the default, no-secret agent streaming path across the
+// services → agents → A2A boundary: an A2A message/stream request invokes an
+// agent StreamAsk turn, the agent executes a tool, and the gateway emits
+// working SSE task updates before the completed final answer.
+package main
+
+import (
+	"bufio"
+	"bytes"
+	"context"
+	"encoding/json"
+	"errors"
+	"flag"
+	"fmt"
+	"io"
+	"net/http"
+	"net/http/httptest"
+	"os"
+	"strings"
+	"time"
+
+	"go-micro.dev/v6/agent"
+	"go-micro.dev/v6/ai"
+	"go-micro.dev/v6/gateway/a2a"
+	"go-micro.dev/v6/internal/harness/harnessutil"
+	"go-micro.dev/v6/registry"
+	"go-micro.dev/v6/store"
+)
+
+type mockModel struct{ opts ai.Options }
+
+func newMock(opts ...ai.Option) ai.Model {
+	m := &mockModel{}
+	_ = m.Init(opts...)
+	return m
+}
+
+func (m *mockModel) Init(opts ...ai.Option) error {
+	for _, o := range opts {
+		o(&m.opts)
+	}
+	return nil
+}
+func (m *mockModel) Options() ai.Options { return m.opts }
+func (m *mockModel) String() string      { return "mock" }
+func (m *mockModel) Stream(context.Context, *ai.Request, ...ai.GenerateOption) (ai.Stream, error) {
+	return nil, ai.ErrStreamingUnsupported
+}
+func (m *mockModel) Generate(ctx context.Context, req *ai.Request, _ ...ai.GenerateOption) (*ai.Response, error) {
+	if req.Prompt == "" {
+		return nil, errors.New("missing prompt")
+	}
+	if len(req.Tools) == 0 || m.opts.ToolHandler == nil {
+		return nil, errors.New("missing tools or tool handler")
+	}
+	res := m.opts.ToolHandler(ctx, ai.ToolCall{ID: "a2a-stream-call", Name: "stream_echo", Input: map[string]any{"value": "a2a-stream"}})
+	if res.Content == "" {
+		return nil, errors.New("empty tool result")
+	}
+	return &ai.Response{
+		Reply:  "streaming completed",
+		Answer: res.Content,
+		ToolCalls: []ai.ToolCall{{
+			ID: "a2a-stream-call", Name: "stream_echo", Input: map[string]any{"value": "a2a-stream"}, Result: res.Content,
+		}},
+	}, nil
+}
+
+type agentStreamAdapter struct{ stream agent.AgentStream }
+
+func (s agentStreamAdapter) Recv() (*ai.Response, error) {
+	for {
+		event, err := s.stream.Recv()
+		if err != nil {
+			return nil, err
+		}
+		if event == nil {
+			continue
+		}
+		switch event.Type {
+		case agent.StreamEventToken:
+			if event.Token != "" {
+				return &ai.Response{Reply: event.Token}, nil
+			}
+		case agent.StreamEventDone:
+			return nil, io.EOF
+		}
+	}
+}
+
+func (s agentStreamAdapter) Close() error { return s.stream.Close() }
+
+func main() {
+	provider := flag.String("provider", "mock", "LLM provider; mock is deterministic and requires no API key")
+	flag.Parse()
+	if *provider == "mock" {
+		ai.Register("mock", newMock)
+	}
+
+	fmt.Printf("\n\033[1mA2A streaming conformance (provider: %s)\033[0m\n", *provider)
+	reg := registry.NewMemoryRegistry()
+	st := store.NewMemoryStore()
+	var sawTool, sawRunInfo bool
+	ag := agent.New(append([]agent.Option{
+		agent.Name("a2a-streaming"),
+		agent.Provider(*provider),
+		agent.Prompt("Use stream_echo exactly once with value a2a-stream, then answer with the tool result."),
+		agent.WithRegistry(reg),
+		agent.WithStore(st),
+		agent.WithMemory(agent.NewInMemory(8)),
+		agent.ModelCallTimeout(45 * time.Second),
+		agent.WithTool("stream_echo", "Echo the A2A stream marker.", map[string]any{
+			"value": map[string]any{"type": "string", "description": "value to echo"},
+		}, func(ctx context.Context, input map[string]any) (string, error) {
+			sawTool = true
+			info, ok := ai.RunInfoFrom(ctx)
+			if !ok || info.RunID == "" || info.Agent != "a2a-streaming" {
+				return "", fmt.Errorf("unexpected run info: %+v", info)
+			}
+			sawRunInfo = true
+			if input["value"] != "a2a-stream" {
+				return "", fmt.Errorf("unexpected value %v", input["value"])
+			}
+			return `{"marker":"a2a-stream-ok"}`, nil
+		}),
+	}, harnessutil.AgentOptions(*provider)...)...)
+
+	handler := a2a.NewAgentStreamHandler(
+		a2a.Card("a2a-streaming", "http://example.invalid/a2a-streaming", "", nil),
+		func(ctx context.Context, text string) (string, error) {
+			resp, err := ag.Ask(ctx, text)
+			if err != nil {
+				return "", err
+			}
+			return resp.Reply, nil
+		},
+		func(ctx context.Context, text string) (ai.Stream, error) {
+			stream, err := agent.StreamAsk(ctx, ag, text)
+			if err != nil {
+				return nil, err
+			}
+			return agentStreamAdapter{stream: stream}, nil
+		},
+	)
+
+	body := []byte(`{"jsonrpc":"2.0","id":1,"method":"message/stream","params":{"message":{"role":"user","parts":[{"kind":"text","text":"Run the A2A streaming conformance check."}],"kind":"message"}}}`)
+	req := httptest.NewRequest(http.MethodPost, "/", bytes.NewReader(body))
+	req.Header.Set("Content-Type", "application/json")
+	rr := httptest.NewRecorder()
+	handler.ServeHTTP(rr, req)
+	res := rr.Result()
+	defer res.Body.Close()
+	if res.StatusCode != http.StatusOK {
+		b, _ := io.ReadAll(res.Body)
+		fmt.Fprintf(os.Stderr, "unexpected status %d: %s\n", res.StatusCode, b)
+		os.Exit(1)
+	}
+	if ct := res.Header.Get("Content-Type"); !strings.HasPrefix(ct, "text/event-stream") {
+		fmt.Fprintf(os.Stderr, "content-type = %q, want text/event-stream\n", ct)
+		os.Exit(1)
+	}
+	summary, err := readSSESummary(res.Body)
+	if err != nil {
+		fmt.Fprintln(os.Stderr, err)
+		os.Exit(1)
+	}
+	if summary.WorkingEvents == 0 || summary.State != "completed" || !strings.Contains(summary.FinalText, "a2a-stream-ok") {
+		fmt.Fprintf(os.Stderr, "unexpected stream summary: %+v\npayload:\n%s", summary, summary.Payload)
+		os.Exit(1)
+	}
+	if !sawTool || !sawRunInfo {
+		fmt.Fprintf(os.Stderr, "tool=%v runInfo=%v\n", sawTool, sawRunInfo)
+		os.Exit(1)
+	}
+	fmt.Println("\n\033[32m✓ A2A message/stream emitted incremental task updates and preserved tool/run metadata\033[0m")
+}
+
+type streamSummary struct {
+	Payload       string
+	State         string
+	FinalText     string
+	WorkingEvents int
+}
+
+func readSSESummary(r io.Reader) (streamSummary, error) {
+	scanner := bufio.NewScanner(r)
+	var event strings.Builder
+	var summary streamSummary
+	seen := false
+	flush := func() error {
+		data := strings.TrimSpace(event.String())
+		event.Reset()
+		if data == "" {
+			return nil
+		}
+		var envelope struct {
+			Result struct {
+				Status struct {
+					State string `json:"state"`
+				} `json:"status"`
+				Artifacts []struct {
+					Parts []struct {
+						Text string `json:"text"`
+					} `json:"parts"`
+				} `json:"artifacts"`
+			} `json:"result"`
+			Error any `json:"error"`
+		}
+		if err := json.Unmarshal([]byte(data), &envelope); err != nil {
+			return fmt.Errorf("SSE data event is not JSON: %s", data)
+		}
+		if envelope.Error != nil {
+			return fmt.Errorf("SSE data event has error: %s", data)
+		}
+		seen = true
+		summary.Payload += data + "\n"
+		if envelope.Result.Status.State == "working" {
+			summary.WorkingEvents++
+		}
+		if envelope.Result.Status.State != "" {
+			summary.State = envelope.Result.Status.State
+		}
+		for _, artifact := range envelope.Result.Artifacts {
+			for _, part := range artifact.Parts {
+				summary.FinalText = part.Text
+			}
+		}
+		return nil
+	}
+	for scanner.Scan() {
+		line := scanner.Text()
+		if strings.TrimSpace(line) == "" {
+			if err := flush(); err != nil {
+				return streamSummary{}, err
+			}
+			continue
+		}
+		data, ok := strings.CutPrefix(line, "data:")
+		if !ok {
+			continue
+		}
+		if event.Len() > 0 {
+			event.WriteByte('\n')
+		}
+		event.WriteString(strings.TrimSpace(data))
+	}
+	if err := scanner.Err(); err != nil {
+		return streamSummary{}, err
+	}
+	if err := flush(); err != nil {
+		return streamSummary{}, err
+	}
+	if !seen {
+		return streamSummary{}, errors.New("no SSE data received")
+	}
+	return summary, nil
+}

internal/harness/provider-conformance/main.go

@@ -35,13 +35,14 @@ import (
 	_ "go-micro.dev/v6/ai/together"
 )
 
-const defaultHarnesses = "agent,universe,agent-flow,plan-delegate,a2a-stream-fallback"
+const defaultHarnesses = "agent,universe,agent-flow,plan-delegate,a2a-streaming,a2a-stream-fallback"
 
 var harnessPhases = map[string]string{
 	"agent":               "model call + tool call",
 	"universe":            "service discovery + tool call",
 	"agent-flow":          "workflow event + tool call",
 	"plan-delegate":       "plan persistence + delegation + tool call",
+	"a2a-streaming":       "A2A streaming + tool call",
 	"a2a-stream-fallback": "streaming fallback + tool call",
 }
 

internal/harness/provider-conformance/main_test.go

@@ -156,8 +156,11 @@ func TestWriteSummaryMarkdown(t *testing.T) {
 }
 
 func TestHarnessPhaseLabelsKnownHarnesses(t *testing.T) {
+	if got := harnessPhase("a2a-streaming"); got != "A2A streaming + tool call" {
+		t.Fatalf("harnessPhase(a2a-streaming) = %q, want A2A streaming phase", got)
+	}
 	if got := harnessPhase("a2a-stream-fallback"); got != "streaming fallback + tool call" {
-		t.Fatalf("harnessPhase() = %q, want streaming fallback phase", got)
+		t.Fatalf("harnessPhase(a2a-stream-fallback) = %q, want streaming fallback phase", got)
 	}
 	if got := harnessPhase("custom"); got != "harness" {
 		t.Fatalf("harnessPhase(custom) = %q, want fallback phase", got)

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

@@ -40,7 +40,7 @@ The built-in providers currently register these capability interfaces:
 | --- | --- | --- | --- | --- | --- |
 | `anthropic` | Yes | No | No | Yes | Yes |
 | `atlascloud` | Yes | Yes | Yes | Yes | No |
-| `gemini` | Yes | No | No | No | No |
+| `gemini` | Yes | No | No | Yes | No |
 | `groq` | Yes | No | No | Yes | Yes |
 | `minimax` | Yes | No | No | Yes | Yes |
 | `mistral` | Yes | No | No | Yes | Yes |

internal/website/docs/guides/debugging-agents.md

@@ -112,6 +112,14 @@ Useful statuses include `done`, `refused`, `timeout`, `rate_limited`, `canceled`
 and `error`. Use `--json` when you want exact timestamps, trace/span ids, and error
 kinds for a bug report.
 
+When a run is paused at `stage=input-required`, continue it from the CLI and then
+inspect the completed checkpoint without writing a Go helper:
+
+```sh
+micro agent resume-input support <run-id> --input "Approve deploy to us-east-1"
+micro inspect agent support --limit 1
+```
+
 Run timelines are stored in the agent's state store under that agent's scoped
 state (`agent/<name>/runs/...`). The persisted timeline is recorded even without
 an OpenTelemetry exporter, so `micro inspect agent` remains useful in local

internal/website/docs/guides/provider-conformance.md

@@ -59,7 +59,7 @@ previous section.
 | --- | --- | --- | --- | --- | --- |
 | `anthropic` | ✅ Verified when configured | — Unsupported | — Unsupported | ✅ Verified when configured | ⚠️ Unverified |
 | `openai` | ✅ Verified when configured | ✅ Registered | — Unsupported | ⚠️ Unverified | ⚠️ Unverified |
-| `gemini` | ✅ Verified when configured | — Unsupported | — Unsupported | ⚠️ Unverified | ⚠️ Unverified |
+| `gemini` | ✅ Verified when configured | — Unsupported | — Unsupported | ✅ Verified when configured | ⚠️ Unverified |
 | `groq` | ✅ Verified when configured | — Unsupported | — Unsupported | ⚠️ Unverified | ⚠️ Unverified |
 | `mistral` | ✅ Verified when configured | — Unsupported | — Unsupported | ⚠️ Unverified | ⚠️ Unverified |
 | `together` | ✅ Verified when configured | — Unsupported | — Unsupported | ⚠️ Unverified | ⚠️ Unverified |

internal/website/docs/guides/x402-payments.md

@@ -100,7 +100,21 @@ 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. 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.)
+`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.
+
+## Agent-level spend guardrail
+
+For unattended agents, set the same cap at the agent tool-execution layer so paid tools are refused before their handler — and therefore before a payer — can run:
+
+```go
+agent := micro.NewAgent("buyer",
+    micro.AgentMaxSteps(8),
+    micro.AgentMaxSpend(20_000), // per Ask, smallest units
+    micro.AgentToolSpend("weather.Weather.Forecast", 10_000),
+)
+```
+
+`AgentMaxSpend` is disabled by default (`0`). `AgentToolSpend` records the price discovered from your shoppable MCP/x402 catalog for the tools this agent may call. When a call would exceed the per-run allowance, the result is a normal structured guardrail refusal with `Refused: "spend_budget"` and an explanatory error in the run timeline/inspect output, distinct from provider/model failures.
 
 ### Live facilitator conformance
 
@@ -121,7 +135,7 @@ Leave those variables unset in normal CI; the live test skips unless the facilit
 
 - **Opt-in.** No pay-to address (and no config), no payments — nothing changes.
 - **No crypto in the framework.** The facilitator does verification and settlement on-chain; Go Micro speaks HTTP.
-- **A paying agent needs a budget.** On the agent side, an unattended agent that spends money needs a spend cap next to `MaxSteps` and `ApproveTool` — see [Plan & Delegate](plan-delegate.html) for the guardrail model. This is active work.
+- **A paying agent needs a budget.** Use `AgentMaxSpend` plus `AgentToolSpend` next to `MaxSteps` and `ApproveTool` so a run has an explicit allowance before any paid tool can execute.
 
 ## See also
 

internal/website/docs/guides/your-first-agent.md

@@ -184,6 +184,14 @@ and timing behind the answer:
 micro inspect agent assistant
 ```
 
+If inspect shows `stage=input-required`, provide the missing value and inspect the
+completed run from the same local store:
+
+```sh
+micro agent resume-input assistant <run-id> --input "Approve the next step"
+micro inspect agent assistant --limit 1
+```
+
 If the model refuses to call tools, tighten the prompt so it explicitly
 uses the `task` service before answering.
 

micro.go

@@ -12,6 +12,7 @@ import (
 	"go-micro.dev/v6/server"
 	"go-micro.dev/v6/service"
 	"go-micro.dev/v6/store"
+	"go-micro.dev/v6/wrapper/x402"
 	"go.opentelemetry.io/otel/trace"
 )
 
@@ -120,6 +121,23 @@ func AgentLoopLimit(n int) AgentOption { return agent.LoopLimit(n) }
 // each action the agent takes.
 func AgentApproveTool(fn ApproveFunc) AgentOption { return agent.ApproveTool(fn) }
 
+// AgentMaxSpend bounds paid x402 tool spend per Ask, in the asset's smallest
+// unit (0 = disabled). Calls that would exceed it are refused before payment.
+func AgentMaxSpend(amount int64) AgentOption { return agent.MaxSpend(amount) }
+
+// AgentToolSpend records the x402 price for a tool, in the asset's smallest
+// unit, so AgentMaxSpend can reserve budget before execution.
+func AgentToolSpend(tool string, amount int64) AgentOption {
+	return agent.ToolSpend(tool, amount)
+}
+
+// AgentPayer configures the wallet/signing hook used to settle x402-paid tools.
+func AgentPayer(p x402.Payer) AgentOption { return agent.Payer(p) }
+
+// AgentBudget bounds autonomous x402 payments per Ask, in the asset's smallest
+// unit (0 = unlimited).
+func AgentBudget(amount int64) AgentOption { return agent.Budget(amount) }
+
 // AgentModelCallTimeout sets the timeout for each provider Generate call.
 func AgentModelCallTimeout(d time.Duration) AgentOption { return agent.ModelCallTimeout(d) }
 

wrapper/x402/client.go

@@ -8,6 +8,7 @@ import (
 	"io"
 	"net/http"
 	"strconv"
+	"strings"
 	"sync"
 )
 
@@ -81,7 +82,16 @@ func (c *Client) Do(req *http.Request) (*http.Response, error) {
 		return resp, fmt.Errorf("x402: 402 response carried no requirements")
 	}
 	reqd := ch.Accepts[0]
-	amount, _ := strconv.ParseInt(reqd.MaxAmountRequired, 10, 64)
+	// The amount governs the whole spend cap, so it must be a real positive
+	// integer. A swallowed parse error (non-decimal, overflow, empty) would
+	// yield 0 and pass the budget check trivially, and a negative amount would
+	// inflate the remaining allowance — either way the cap is defeated. Refuse
+	// before signing anything.
+	amount, err := strconv.ParseInt(strings.TrimSpace(reqd.MaxAmountRequired), 10, 64)
+	if err != nil || amount <= 0 {
+		return resp, fmt.Errorf("x402: refusing to pay %s: invalid maxAmountRequired %q",
+			reqd.Resource, reqd.MaxAmountRequired)
+	}
 
 	// Spend cap: reserve before paying so concurrent calls cannot all pass
 	// the check and overspend the caller's allowance. Roll the reservation

wrapper/x402/client_test.go

@@ -172,6 +172,33 @@ func TestClientBudgetReservationRollsBackOnPayError(t *testing.T) {
 	}
 }
 
+// A 402 whose maxAmountRequired is not a positive integer must be refused
+// before any payment — otherwise a swallowed parse error (0) or a negative
+// amount defeats the spend cap. The payer is never called and nothing is spent.
+func TestClientRefusesInvalidAmount(t *testing.T) {
+	for _, amount := range []string{"abc", "-100", "99999999999999999999999999", "0x10", "1.5"} {
+		srv := paidServer(amount)
+
+		payer := &mockPayer{}
+		c := &Client{Payer: payer, Budget: 1_000_000}
+		req, _ := http.NewRequest(http.MethodGet, srv.URL, nil)
+		resp, err := c.Do(req)
+		if resp != nil {
+			resp.Body.Close()
+		}
+		if err == nil {
+			t.Errorf("amount %q: expected refusal, got nil error", amount)
+		}
+		if payer.calls != 0 {
+			t.Errorf("amount %q: payer called %d times, want 0", amount, payer.calls)
+		}
+		if c.Spent() != 0 {
+			t.Errorf("amount %q: spent %d, want 0", amount, c.Spent())
+		}
+		srv.Close()
+	}
+}
+
 type payerFunc func(context.Context, Requirements) (string, error)
 
 func (f payerFunc) Pay(ctx context.Context, req Requirements) (string, error) {

wrapper/x402/x402.go

@@ -126,6 +126,11 @@ type Config struct {
 	// FacilitatorURL is the verify/settle endpoint used when Facilitator
 	// is nil (e.g. Coinbase CDP or Alchemy).
 	FacilitatorURL string `json:"facilitator,omitempty"`
+	// RequireSettlement fails closed when a paid request cannot be settled:
+	// if the facilitator only verifies (does not implement Settler), Require
+	// refuses to serve rather than releasing the resource while no funds move.
+	// Leave false only for verify-only flows where authorization is enough.
+	RequireSettlement bool `json:"requireSettlement,omitempty"`
 }
 
 func (c Config) network() string {
@@ -222,7 +227,14 @@ func (c Config) Require(w http.ResponseWriter, r *http.Request, amount, resource
 	}
 	// Capture the funds when the facilitator can settle. Verify alone only
 	// authorizes the "exact" transfer; settlement broadcasts it.
-	if s, ok := fac.(Settler); ok {
+	s, canSettle := fac.(Settler)
+	if c.RequireSettlement && !canSettle {
+		// Fail closed: a paid config must not serve the resource on a
+		// verify-only facilitator, or it gives the tool away for free.
+		writeChallenge(w, req, "payment settlement unavailable")
+		return false
+	}
+	if canSettle {
 		sres, err := s.Settle(r.Context(), payment, req)
 		if err != nil {
 			writeChallenge(w, req, "payment settlement failed: "+err.Error())

wrapper/x402/x402_settle_test.go

@@ -159,4 +159,58 @@ func TestCDPAuthorizeAttachesBearer(t *testing.T) {
 	}
 }
 
+// TestRequireSettlementFailsClosed checks that a paid config with
+// RequireSettlement refuses to serve when the facilitator only verifies (does
+// not settle) — otherwise the resource is released while no funds move.
+func TestRequireSettlementFailsClosed(t *testing.T) {
+	// mockFacilitator implements Verify but not Settler.
+	cfg := Config{PayTo: "0xpay", Facilitator: mockFacilitator{valid: true}, RequireSettlement: true}
+	r := httptest.NewRequest(http.MethodGet, "/tool", nil)
+	r.Header.Set(PaymentHeader, "eyJ4IjoxfQ==")
+	rec := httptest.NewRecorder()
+
+	if cfg.Require(rec, r, "10000", "chat") {
+		t.Fatal("Require should fail closed when settlement is required but unavailable")
+	}
+	if rec.Code != http.StatusPaymentRequired {
+		t.Errorf("status = %d, want 402", rec.Code)
+	}
+
+	// Without RequireSettlement the verify-only facilitator still serves.
+	cfg.RequireSettlement = false
+	rec = httptest.NewRecorder()
+	r = httptest.NewRequest(http.MethodGet, "/tool", nil)
+	r.Header.Set(PaymentHeader, "eyJ4IjoxfQ==")
+	if !cfg.Require(rec, r, "10000", "chat") {
+		t.Fatalf("verify-only should serve when settlement is not required; body=%s", rec.Body.String())
+	}
+}
+
+// TestRequireSettlementServesWithSettler checks that a paid config with
+// RequireSettlement serves when the facilitator can settle.
+func TestRequireSettlementServesWithSettler(t *testing.T) {
+	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		switch r.URL.Path {
+		case "/verify":
+			_ = json.NewEncoder(w).Encode(map[string]any{"isValid": true})
+		case "/settle":
+			_ = json.NewEncoder(w).Encode(map[string]any{"success": true, "transaction": "0xabc"})
+		}
+	}))
+	defer srv.Close()
+
+	// HTTPFacilitator implements Settler.
+	cfg := Config{PayTo: "0xpay", FacilitatorURL: srv.URL, RequireSettlement: true}
+	r := httptest.NewRequest(http.MethodGet, "/tool", nil)
+	r.Header.Set(PaymentHeader, "eyJ4IjoxfQ==")
+	rec := httptest.NewRecorder()
+
+	if !cfg.Require(rec, r, "10000", "chat") {
+		t.Fatalf("Require should serve with a settling facilitator; body=%s", rec.Body.String())
+	}
+	if got := rec.Header().Get(PaymentResponseHeader); got != "0xabc" {
+		t.Errorf("settlement header = %q, want 0xabc", got)
+	}
+}
+
 var _ Settler = (*HTTPFacilitator)(nil)