Update README with overview and community sections

Added an overview section and restructured community information in the README.

.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. **Propagate agent run cancellation and deadlines through model and tool calls** ([#4750](https://github.com/micro/go-micro/issues/4750)) — #4748 closed the x402 spend-budget guardrail and there are no open codex PRs in flight. Do not re-queue more docs-link checks, AtlasCloud-specific text repair, or plan/delegate edge hardening for now; those areas have had several recent increments. The next highest-value user-facing gap is making long-running and unattended agents easier to stop safely and inspect: cancellation/deadline propagation through model calls, tool execution, delegation seams, and run metadata without breaking public APIs.
+### Capability — the headline (roadmap: Now / Next)
 
-_Seeded by Claude Code from the roadmap + open issues; thereafter maintained by the
-architecture-review pass._
+1. **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.
+2. **AP2 mandate foundation for agent payments** ([#4841](https://github.com/micro/go-micro/issues/4841)) — add opt-in checkout/payment mandate signing and verification so A2A-carried payment authority can settle over x402 without changing defaults.
+3. **Kubernetes CRD reconciler foundation** ([#4842](https://github.com/micro/go-micro/issues/4842)) — turn the shipped alpha `Agent`, `Service`, and `Flow` CRDs into a minimally runnable native deployment path with workload reconciliation and status conditions.
+
+### In flight — do not re-queue
+
+_None right now._
+
+### 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, provider-failure inspection, x402 buyer safety, gRPC-reflection MCP,
+MCP result conformance, and the alpha Kubernetes CRD surface. Further churn in those
+areas should be marked `needs-human` unless it unlocks a clear user-visible capability._

.github/workflows/loop-builder.yml

@@ -14,8 +14,11 @@ name: "Loop: Builder"
 
 on:
   workflow_dispatch: {}
-  schedule:
-    - cron: "29 * * * *"
+  # PAUSED 2026-07-12: automatic schedule disabled while the team does focused
+  # 1:1 fixes. Still runnable on demand via workflow_dispatch. Re-enable by
+  # uncommenting the schedule below.
+  # schedule:
+  #   - cron: "29 * * * *"
 
 permissions:
   issues: write

.github/workflows/loop-coherence.yml

@@ -14,8 +14,11 @@ name: "Loop: Coherence"
 
 on:
   workflow_dispatch: {}
-  schedule:
-    - cron: "0 7 * * *"
+  # PAUSED 2026-07-12: automatic schedule disabled while the team does focused
+  # 1:1 fixes. Still runnable on demand via workflow_dispatch. Re-enable by
+  # uncommenting the schedule below.
+  # schedule:
+  #   - cron: "0 7 * * *"
 
 permissions:
   issues: write

.github/workflows/loop-planner.yml

@@ -14,8 +14,11 @@ name: "Loop: Planner"
 
 on:
   workflow_dispatch: {}
-  schedule:
-    - cron: "59 * * * *"
+  # PAUSED 2026-07-12: automatic schedule disabled while the team does focused
+  # 1:1 fixes. Still runnable on demand via workflow_dispatch. Re-enable by
+  # uncommenting the schedule below.
+  # schedule:
+  #   - cron: "59 * * * *"
 
 permissions:
   issues: write

.github/workflows/loop-release.yml

@@ -12,8 +12,11 @@ name: "Loop: Release"
 
 on:
   workflow_dispatch: {}
-  schedule:
-    - cron: "0 23 * * *"
+  # PAUSED 2026-07-12: automatic nightly release disabled while the team does
+  # focused 1:1 fixes. Cut a release on demand via workflow_dispatch. Re-enable
+  # by uncommenting the schedule below.
+  # schedule:
+  #   - cron: "0 23 * * *"
 
 permissions:
   contents: read

.github/workflows/loop-security.yml

@@ -14,8 +14,11 @@ name: "Loop: Security"
 
 on:
   workflow_dispatch: {}
-  schedule:
-    - cron: "0 6 * * 1"
+  # PAUSED 2026-07-12: automatic schedule disabled while the team does focused
+  # 1:1 fixes. Still runnable on demand via workflow_dispatch. Re-enable by
+  # uncommenting the schedule below.
+  # schedule:
+  #   - cron: "0 6 * * 1"
 
 permissions:
   issues: write

.github/workflows/loop-triage.yml

@@ -7,9 +7,13 @@ name: "Loop: Triage"
 # failures become fixes with no human in the middle. Gated on CODEX_TRIGGER_TOKEN.
 
 on:
-  workflow_run:
-    workflows: ["Harness (E2E)", "Lint", "Run Tests", "govulncheck"]
-    types: [completed]
+  workflow_dispatch: {}
+  # PAUSED 2026-07-12: automatic CI-failure dispatch disabled while the team
+  # does focused 1:1 fixes, so failures don't auto-spawn agent tasks. Re-enable
+  # by uncommenting the workflow_run trigger below.
+  # workflow_run:
+  #   workflows: ["Harness (E2E)", "Lint", "Run Tests", "govulncheck"]
+  #   types: [completed]
 
 permissions:
   issues: write

CHANGELOG.md

@@ -18,15 +18,56 @@ 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/`)
+- **A2A inbound AP2 mandate verification (opt-in)** — set `Options.AP2PublicKey` (or `a2a.WithPushURLPolicy`'s sibling `a2a.WithAP2PublicKey` for embedded handlers) and the gateway verifies AP2 payment/checkout mandates carried on incoming messages — signature and task/context binding — recording the outcome in each task's `ap2Verifications`, with the x402 settlement rail carried through for the paid path. Off by default; mandates are otherwise carried unverified. (`gateway/a2a/`)
+- **Flow human-in-the-loop pause/resume** — a flow step can suspend a run for external input with `flow.Await(key, prompt)` (or `flow.AwaitStep`): the run checkpoints with status `waiting` and `Execute` returns cleanly. `Flow.Waiting` lists suspended runs with what they await, and `Flow.ResumeWith(ctx, runID, input)` injects the input and continues from the next step. Recovery (`ResumePending`) skips waiting runs since they need input, not a restart. (`flow/`)
+- **Kubernetes reconcile core (alpha)** — `kubernetes.Reconcile(desired, observed)` decides the single action needed to converge an `Agent`/`Service`/`Flow` resource toward its Deployment (create / update / noop) and returns `Ready`/`Error` status conditions. Dependency-free (no controller-runtime / client-go) and fully unit-testable; a future operator binary supplies observed state and applies the action. (`deploy/kubernetes/`)
+- **In-process "local network" fast-path (opt-in)** — `client.Local()` lets a unary `Call` to a service running in the same process skip the network transport and dispatch straight to that server's handlers (for raw `codec/bytes.Frame` bodies — the shape agent/MCP/flow tool calls use), running the same router, wrappers, and codecs. In a benchmark this cut an in-process call from ~545µs to ~28µs (≈20×) with ~3.6× fewer allocations. Off by default; falls back to the network path for anything it doesn't cover. (`client/`, `server/`, `internal/network/`)
+- **`micro.Local()` service option** — turn on the in-process fast-path for a whole service in one place: every co-located unary call its client makes (agent tool calls, flow dispatch, gateway → service) takes the fast-path, with no per-call wiring. Off by default; a no-op for distributed deployments. (`service/`, root `options.go`)
+
+### Changed
+- **Remote agent chat streaming** — `micro chat` now streams replies from remote agents instead of waiting for the full response. (`cmd/micro/`, `agent/`)
+- **A2A external-client conformance** — the A2A gateway now serves the Agent Card at the spec 0.3.0 `/.well-known/agent-card.json` (keeping `/.well-known/agent.json` as a legacy alias), and `message/stream` emits spec-shaped `status-update`/`artifact-update` events ending in a `final:true` status-update instead of repeated full `Task` snapshots — and never sends `result` and `error` together. Standard A2A clients (ADK, LangGraph, a2a-SDK) can now discover and stream from go-micro agents. (`gateway/a2a/`)
+
+### 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/`)
+- **A2A push-notification SSRF guard** — the A2A gateway no longer delivers task push notifications to caller-supplied URLs that resolve to loopback, private, link-local (incl. cloud metadata), or unspecified addresses. Callbacks are validated when set and re-checked at dial time on the resolved IP (DNS-rebinding safe); non-http(s) schemes are rejected. `Options.AllowPushURL` (and `a2a.WithPushURLPolicy` for embedded handlers) lets operators authorize trusted in-cluster receivers. (`gateway/a2a/`)
+
+---
+
+## [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

README.md

@@ -2,8 +2,7 @@
 
 Go Micro is an **agent harness** and service framework for Go.
 
-**Community:** questions, ideas, or just want to build alongside us? [Join the Discord](https://discord.gg/G8Gk5j3uXr).
-
+## Overview
 A harness is the runtime around an agent: the tools it can call, the memory it keeps, the guardrails that bound it, the workflows that trigger it, the services it depends on, and the protocols other agents use to reach it. 
 
 Go Micro gives you the harness as Go code. Build an agent and it gets a model, memory, tools, planning, delegation, guardrails, and service discovery; it is reachable over [MCP](https://modelcontextprotocol.io/) and [A2A](https://a2a-protocol.org). Write services and every endpoint becomes an AI-callable tool. Orchestrate the deterministic parts with durable flows. Agents, services, and flows share one runtime because an agent is a distributed system, and building one is building a service.
@@ -18,6 +17,10 @@ Go Micro gives you the harness as Go code. Build an agent and it gets a model, m
 
 **Want to support Go Micro and see your logo here?** [Become a sponsor](https://discord.gg/G8Gk5j3uXr) — reach out on Discord.
 
+## Community
+
+Questions, ideas, or just want to build alongside us? [Join the Discord](https://discord.gg/G8Gk5j3uXr).
+
 ## Commercial Support
 
 Running Go Micro in production, or building on it and want help? Paid **support, consulting, training, and retainers** are available directly from the maintainer — and they're what keep the project maintained. See [**Support**](SUPPORT.md) for the tiers, or [open a request](https://github.com/micro/go-micro/issues/new?template=commercial_support.md).

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

@@ -68,35 +68,79 @@ func TestA2AStreamUsesAgentChatPathWithTools(t *testing.T) {
 		t.Fatalf("stream body missing tool marker: %s", rr.Body.String())
 	}
 
-	var final 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"`
-	}
+	// The spec-shaped stream carries the answer as append artifact-update
+	// deltas and closes with a completed status-update (final:true).
+	var (
+		text       strings.Builder
+		finalState string
+		sawFinal   bool
+	)
 	for _, line := range strings.Split(strings.TrimSpace(rr.Body.String()), "\n") {
 		line = strings.TrimSpace(strings.TrimPrefix(strings.TrimSpace(line), "data: "))
 		if line == "" {
 			continue
 		}
-		if err := json.Unmarshal([]byte(line), &final); err != nil {
+		var ev struct {
+			Result json.RawMessage `json:"result"`
+			Error  any             `json:"error"`
+		}
+		if err := json.Unmarshal([]byte(line), &ev); err != nil {
 			t.Fatalf("decode event %q: %v", line, err)
 		}
+		if ev.Error != nil {
+			t.Fatalf("event carried an error field: %+v", ev.Error)
+		}
+		var kind struct {
+			Kind string `json:"kind"`
+		}
+		_ = json.Unmarshal(ev.Result, &kind)
+		switch kind.Kind {
+		case "artifact-update":
+			var au struct {
+				Artifact struct {
+					Parts []struct {
+						Text string `json:"text"`
+					} `json:"parts"`
+				} `json:"artifact"`
+			}
+			_ = json.Unmarshal(ev.Result, &au)
+			for _, p := range au.Artifact.Parts {
+				text.WriteString(p.Text)
+			}
+		case "status-update":
+			var su struct {
+				Status struct {
+					State string `json:"state"`
+				} `json:"status"`
+				Final bool `json:"final"`
+			}
+			_ = json.Unmarshal(ev.Result, &su)
+			if su.Final {
+				sawFinal = true
+				finalState = su.Status.State
+			}
+		default: // opening "task" snapshot
+			var task struct {
+				Artifacts []struct {
+					Parts []struct {
+						Text string `json:"text"`
+					} `json:"parts"`
+				} `json:"artifacts"`
+			}
+			_ = json.Unmarshal(ev.Result, &task)
+			for _, a := range task.Artifacts {
+				for _, p := range a.Parts {
+					if p.Text != "" {
+						text.WriteString(p.Text)
+					}
+				}
+			}
+		}
 	}
-	if final.Error != nil {
-		t.Fatalf("final event error: %+v", final.Error)
+	if !sawFinal || finalState != "completed" {
+		t.Fatalf("want a completed final:true status-update; sawFinal=%v state=%q", sawFinal, finalState)
 	}
-	if final.Result.Status.State != "completed" {
-		t.Fatalf("final state = %q, want completed", final.Result.Status.State)
-	}
-	if len(final.Result.Artifacts) != 1 || len(final.Result.Artifacts[0].Parts) != 1 || !strings.Contains(final.Result.Artifacts[0].Parts[0].Text, "a2a-stream-ok") {
-		t.Fatalf("final artifacts = %+v, want tool marker", final.Result.Artifacts)
+	if !strings.Contains(text.String(), "a2a-stream-ok") {
+		t.Fatalf("reassembled stream text missing tool marker: %q", text.String())
 	}
 }

agent/agent.go

@@ -15,7 +15,9 @@ package agent
 import (
 	"context"
 	"encoding/json"
+	"errors"
 	"fmt"
+	"io"
 	"net/http"
 	"strings"
 	"sync"
@@ -263,6 +265,36 @@ func (a *agentImpl) Stream(ctx context.Context, message string) (ai.Stream, erro
 	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) {
@@ -368,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,6 +131,7 @@ 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)
@@ -173,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.
@@ -389,6 +451,11 @@ func (a *agentImpl) spendWrap(next ai.ToolHandler) ai.ToolHandler {
 				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

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
@@ -181,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.
@@ -102,6 +106,10 @@ type Options struct {
 	// 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".
@@ -247,6 +255,18 @@ func ToolSpend(tool string, amount int64) Option {
 	}
 }
 
+// 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.
@@ -275,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 {

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

@@ -135,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")

client/local.go

@@ -0,0 +1,60 @@
+package client
+
+import (
+	"context"
+
+	raw "go-micro.dev/v6/codec/bytes"
+	"go-micro.dev/v6/internal/network"
+	"go-micro.dev/v6/metadata"
+	"go-micro.dev/v6/transport"
+	"go-micro.dev/v6/transport/headers"
+)
+
+// localCall is the in-process fast-path for Call. When Local is enabled
+// and the callee runs in this same process, a unary request whose body and
+// response are raw frames (codec/bytes.Frame) is dispatched straight to the
+// server's handlers via internal/network — no dial, no codec-over-socket,
+// no transport pump. It returns handled=false to fall back to the network path
+// for anything it does not cover (disabled, streaming, non-frame bodies, or a
+// service not registered in-process), so behavior is unchanged unless the
+// fast-path fully applies.
+func (r *rpcClient) localCall(ctx context.Context, req Request, resp interface{}) (handled bool, err error) {
+	if !r.opts.Local || req.Stream() {
+		return false, nil
+	}
+	reqFrame, ok := req.Body().(*raw.Frame)
+	if !ok {
+		return false, nil
+	}
+	respFrame, ok := resp.(*raw.Frame)
+	if !ok {
+		return false, nil
+	}
+	dispatch, ok := network.Lookup(req.Service())
+	if !ok {
+		return false, nil
+	}
+
+	header := make(map[string]string)
+	if md, ok := metadata.FromContext(ctx); ok {
+		for k, v := range md {
+			if k == headers.Message { // pub/sub topic header, never forwarded
+				continue
+			}
+			header[k] = v
+		}
+	}
+	header[headers.Request] = req.Service()
+	header[headers.Endpoint] = req.Endpoint()
+	header["Content-Type"] = req.ContentType()
+	header["Accept"] = req.ContentType()
+
+	reply, err := dispatch(ctx, &transport.Message{Header: header, Body: reqFrame.Data})
+	if err != nil {
+		return true, err
+	}
+	if reply != nil {
+		respFrame.Data = reply.Body
+	}
+	return true, nil
+}

client/local_test.go

@@ -0,0 +1,139 @@
+package client_test
+
+import (
+	"context"
+	"encoding/json"
+	"testing"
+	"time"
+
+	"go-micro.dev/v6/client"
+	raw "go-micro.dev/v6/codec/bytes"
+	"go-micro.dev/v6/registry"
+	"go-micro.dev/v6/selector"
+	"go-micro.dev/v6/server"
+)
+
+type EchoReq struct {
+	Msg string `json:"msg"`
+}
+
+type EchoRsp struct {
+	Msg string `json:"msg"`
+}
+
+type EchoHandler struct{}
+
+func (EchoHandler) Echo(_ context.Context, req *EchoReq, rsp *EchoRsp) error {
+	rsp.Msg = "echo:" + req.Msg
+	return nil
+}
+
+// startEchoServer starts a real server on the given registry and returns a stop
+// func. The server is reachable over the network transport and (via Start)
+// registered for the in-process fast-path.
+func startEchoServer(t testing.TB, reg registry.Registry) func() {
+	t.Helper()
+	srv := server.NewServer(
+		server.Name("echo.local"),
+		server.Address("127.0.0.1:0"),
+		server.Registry(reg),
+	)
+	if err := srv.Handle(srv.NewHandler(&EchoHandler{})); err != nil {
+		t.Fatalf("handle: %v", err)
+	}
+	if err := srv.Start(); err != nil {
+		t.Fatalf("start: %v", err)
+	}
+	// Wait for registration so the client's selector can find a node.
+	deadline := time.Now().Add(5 * time.Second)
+	for time.Now().Before(deadline) {
+		if svcs, err := reg.GetService("echo.local"); err == nil && len(svcs) > 0 && len(svcs[0].Nodes) > 0 {
+			break
+		}
+		time.Sleep(10 * time.Millisecond)
+	}
+	return func() { _ = srv.Stop() }
+}
+
+func newEchoClient(reg registry.Registry, opts ...client.Option) client.Client {
+	base := []client.Option{
+		client.Registry(reg),
+		client.Selector(selector.NewSelector(selector.Registry(reg))),
+		client.ContentType("application/json"),
+	}
+	return client.NewClient(append(base, opts...)...)
+}
+
+// callEcho makes an echo call with a raw-frame body (the shape agent/MCP/flow
+// dispatch uses) and returns the decoded reply.
+func callEcho(t testing.TB, cl client.Client, msg string) EchoRsp {
+	t.Helper()
+	body, _ := json.Marshal(EchoReq{Msg: msg})
+	req := cl.NewRequest("echo.local", "EchoHandler.Echo", &raw.Frame{Data: body}, client.WithContentType("application/json"))
+	var rsp raw.Frame
+	if err := cl.Call(context.Background(), req, &rsp); err != nil {
+		t.Fatalf("call: %v", err)
+	}
+	var out EchoRsp
+	if err := json.Unmarshal(rsp.Data, &out); err != nil {
+		t.Fatalf("decode reply %q: %v", rsp.Data, err)
+	}
+	return out
+}
+
+// TestLocalMatchesNetwork proves the in-process fast-path returns the
+// exact same result as the network path for the same handler and request.
+func TestLocalMatchesNetwork(t *testing.T) {
+	reg := registry.NewMemoryRegistry()
+	stop := startEchoServer(t, reg)
+	defer stop()
+
+	net := newEchoClient(reg)                   // network path
+	local := newEchoClient(reg, client.Local()) // in-process fast-path
+
+	netRsp := callEcho(t, net, "hi")
+	localRsp := callEcho(t, local, "hi")
+
+	if netRsp.Msg != "echo:hi" {
+		t.Fatalf("network reply = %q, want echo:hi", netRsp.Msg)
+	}
+	if localRsp != netRsp {
+		t.Fatalf("fast-path reply %+v != network reply %+v", localRsp, netRsp)
+	}
+}
+
+// TestLocalFallsBackWhenNotLocal confirms a service not registered
+// in-process still works via the network path even with Local on.
+func TestLocalFallsBackWhenNotLocal(t *testing.T) {
+	reg := registry.NewMemoryRegistry()
+	stop := startEchoServer(t, reg)
+	defer stop()
+
+	// Local is on, but the call still resolves — the fast-path only
+	// engages when it fully applies, otherwise the network path runs.
+	local := newEchoClient(reg, client.Local())
+	if got := callEcho(t, local, "x").Msg; got != "echo:x" {
+		t.Fatalf("reply = %q, want echo:x", got)
+	}
+}
+
+func benchmarkEcho(b *testing.B, opts ...client.Option) {
+	reg := registry.NewMemoryRegistry()
+	stop := startEchoServer(b, reg)
+	defer stop()
+	cl := newEchoClient(reg, opts...)
+	body, _ := json.Marshal(EchoReq{Msg: "hi"})
+
+	b.ReportAllocs()
+	b.ResetTimer()
+	for i := 0; i < b.N; i++ {
+		req := cl.NewRequest("echo.local", "EchoHandler.Echo", &raw.Frame{Data: body}, client.WithContentType("application/json"))
+		var rsp raw.Frame
+		if err := cl.Call(context.Background(), req, &rsp); err != nil {
+			b.Fatalf("call: %v", err)
+		}
+	}
+}
+
+func BenchmarkNetworkCall(b *testing.B) { benchmarkEcho(b) }
+func BenchmarkLocalCall(b *testing.B)   { benchmarkEcho(b, client.Local()) }

client/options.go

@@ -68,6 +68,11 @@ type Options struct {
 	PoolSize         int
 	PoolTTL          time.Duration
 	PoolCloseTimeout time.Duration
+
+	// Local, when true, lets a unary Call to a service running in this
+	// same process skip the network transport and dispatch directly to that
+	// server's handlers (raw byte bodies only). Off by default.
+	Local bool
 }
 
 // CallOptions are options used to make calls to a server.
@@ -181,6 +186,17 @@ func ContentType(ct string) Option {
 	}
 }
 
+// Local enables the in-process fast-path: a unary Call to a service
+// running in the same process dispatches straight to that server's handlers
+// (skipping dial, codec-over-socket, and the transport pump) when both request
+// and response bodies are raw frames (codec/bytes.Frame) — the shape agent,
+// MCP, and flow tool calls use. Falls back to the network path otherwise.
+func Local() Option {
+	return func(o *Options) {
+		o.Local = true
+	}
+}
+
 // PoolSize sets the connection pool size.
 func PoolSize(d int) Option {
 	return func(o *Options) {

client/rpc_client.go

@@ -83,6 +83,12 @@ func (r *rpcClient) call(
 	resp interface{},
 	opts CallOptions,
 ) error {
+	// In-process fast-path: if the callee runs in this process and both bodies
+	// are raw frames, dispatch directly and skip the network entirely.
+	if handled, err := r.localCall(ctx, req, resp); handled {
+		return err
+	}
+
 	address := node.Address
 	logger := r.Options().Logger
 

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,36 @@
+# 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.
+- A dependency-free `Reconcile(desired, observed)` core that decides the one
+  action needed to converge (create / update / noop) and the `Ready`/`Error`
+  status conditions — no controller-runtime, no client-go, fully unit-testable.
+  A future operator binary supplies the observed state and applies the action;
+  only that adapter needs the Kubernetes client.
+- Unit tests that validate the structural CRD fragments, the Agent-to-Deployment
+  mapping, and the reconcile decision/conditions.
+
+## 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,32 @@
+package kubernetes
+
+import (
+	"embed"
+	"fmt"
+)
+
+// crdFS holds the canonical CRD manifests. They live as real YAML under
+// config/crd/ so they can be applied directly (`kubectl apply -f
+// deploy/kubernetes/config/crd/`) and are embedded here so the Go API serves
+// the exact same bytes — one source of truth, no drift.
+//
+//go:embed config/crd/agent.yaml config/crd/service.yaml config/crd/flow.yaml
+var crdFS embed.FS
+
+// CRDManifests contains the alpha CRDs for Go Micro lifecycle resources, loaded
+// from the embedded config/crd/ YAML.
+var CRDManifests = map[Kind]string{
+	KindAgent:   mustCRD("agent"),
+	KindService: mustCRD("service"),
+	KindFlow:    mustCRD("flow"),
+}
+
+// mustCRD reads an embedded CRD manifest. The files are embedded at compile
+// time, so a read error means a build/packaging bug, not a runtime condition.
+func mustCRD(name string) string {
+	b, err := crdFS.ReadFile("config/crd/" + name + ".yaml")
+	if err != nil {
+		panic(fmt.Sprintf("kubernetes: embedded CRD %q missing: %v", name, err))
+	}
+	return string(b)
+}

deploy/kubernetes/reconcile.go

@@ -0,0 +1,105 @@
+package kubernetes
+
+import (
+	"fmt"
+	"reflect"
+)
+
+// Reconcile is the pure decision core an operator's reconcile loop runs: given
+// a desired resource and the currently observed cluster state, it computes the
+// one action needed to converge (create / update / nothing) plus the status
+// conditions to publish. It does not talk to a cluster — no controller-runtime,
+// no client-go — so the whole convergence decision is unit-testable. An adapter
+// binary supplies Observed from the live cluster and applies the returned
+// Action; that adapter is the only piece that needs the Kubernetes client.
+
+// ActionType is the change a reconcile wants applied.
+type ActionType string
+
+const (
+	// ActionCreate means the workload does not exist yet and should be created.
+	ActionCreate ActionType = "create"
+	// ActionUpdate means the workload exists but drifts from desired.
+	ActionUpdate ActionType = "update"
+	// ActionNoop means the workload already matches desired.
+	ActionNoop ActionType = "noop"
+)
+
+// Action is the change Reconcile decided on, carrying the desired Deployment.
+type Action struct {
+	Type       ActionType
+	Deployment Deployment
+}
+
+// Observed is the current cluster state Reconcile compares against. The adapter
+// fills it from the live cluster; a nil Deployment means "not created yet".
+type Observed struct {
+	// Deployment is the workload as it currently exists, or nil if absent.
+	Deployment *Deployment
+	// ReadyReplicas is how many pods are ready, from the live Deployment status.
+	ReadyReplicas int32
+}
+
+// Condition is a status condition to publish on the resource — the ready/error
+// signal for the inner-loop and deploy story. It mirrors the Kubernetes
+// condition shape without importing the API types.
+type Condition struct {
+	Type    string `json:"type"`   // "Ready" | "Error"
+	Status  string `json:"status"` // "True" | "False" | "Unknown"
+	Reason  string `json:"reason"`
+	Message string `json:"message,omitempty"`
+}
+
+// Reconcile computes the action to bring observed toward desired, plus the
+// status conditions. A spec that fails to map returns an Error condition and
+// the error (no action).
+func Reconcile(desired Resource, observed Observed) (Action, []Condition, error) {
+	want, err := MapDeployment(desired)
+	if err != nil {
+		return Action{}, []Condition{{
+			Type: "Error", Status: "True", Reason: "InvalidSpec", Message: err.Error(),
+		}}, err
+	}
+
+	var action Action
+	switch {
+	case observed.Deployment == nil:
+		action = Action{Type: ActionCreate, Deployment: want}
+	case deploymentDiffers(*observed.Deployment, want):
+		action = Action{Type: ActionUpdate, Deployment: want}
+	default:
+		action = Action{Type: ActionNoop, Deployment: want}
+	}
+
+	return action, conditions(want, observed), nil
+}
+
+// conditions derives the Ready condition from observed state against desired.
+func conditions(want Deployment, observed Observed) []Condition {
+	switch {
+	case observed.Deployment == nil:
+		return []Condition{{
+			Type: "Ready", Status: "False", Reason: "Creating",
+			Message: "workload not yet created",
+		}}
+	case observed.ReadyReplicas < want.Replicas:
+		return []Condition{{
+			Type: "Ready", Status: "False", Reason: "Progressing",
+			Message: fmt.Sprintf("%d/%d replicas ready", observed.ReadyReplicas, want.Replicas),
+		}}
+	default:
+		return []Condition{{
+			Type: "Ready", Status: "True", Reason: "Available",
+			Message: fmt.Sprintf("%d/%d replicas ready", observed.ReadyReplicas, want.Replicas),
+		}}
+	}
+}
+
+// deploymentDiffers reports whether the observed deployment drifts from desired
+// on the fields this operator manages (replicas, container, labels). Fields the
+// cluster owns (status, cluster-assigned metadata) are intentionally ignored.
+func deploymentDiffers(current, want Deployment) bool {
+	return current.Replicas != want.Replicas ||
+		!reflect.DeepEqual(current.Pod.Container, want.Pod.Container) ||
+		!reflect.DeepEqual(current.Labels, want.Labels)
+}

deploy/kubernetes/reconcile_test.go

@@ -0,0 +1,88 @@
+package kubernetes
+
+import "testing"
+
+func agentResource() Resource {
+	return Resource{
+		Kind:      KindAgent,
+		Name:      "support",
+		Namespace: "agents",
+		Spec:      WorkloadSpec{Image: "example/support:v1", Replicas: 2, Registry: "kubernetes"},
+	}
+}
+
+func TestReconcileCreatesWhenAbsent(t *testing.T) {
+	action, conds, err := Reconcile(agentResource(), Observed{Deployment: nil})
+	if err != nil {
+		t.Fatalf("Reconcile: %v", err)
+	}
+	if action.Type != ActionCreate {
+		t.Fatalf("action = %q, want create", action.Type)
+	}
+	if action.Deployment.Name != "support" || action.Deployment.Replicas != 2 {
+		t.Fatalf("desired deployment = %+v", action.Deployment)
+	}
+	if ready := findCondition(conds, "Ready"); ready == nil || ready.Status != "False" || ready.Reason != "Creating" {
+		t.Fatalf("ready condition = %+v, want False/Creating", ready)
+	}
+}
+
+func TestReconcileNoopWhenMatchedAndReady(t *testing.T) {
+	want, _ := MapDeployment(agentResource())
+	action, conds, err := Reconcile(agentResource(), Observed{Deployment: &want, ReadyReplicas: 2})
+	if err != nil {
+		t.Fatalf("Reconcile: %v", err)
+	}
+	if action.Type != ActionNoop {
+		t.Fatalf("action = %q, want noop", action.Type)
+	}
+	if ready := findCondition(conds, "Ready"); ready == nil || ready.Status != "True" || ready.Reason != "Available" {
+		t.Fatalf("ready condition = %+v, want True/Available", ready)
+	}
+}
+
+func TestReconcileUpdatesOnDrift(t *testing.T) {
+	current, _ := MapDeployment(agentResource())
+	current.Pod.Container.Image = "example/support:v0" // stale image → drift
+	action, _, err := Reconcile(agentResource(), Observed{Deployment: &current, ReadyReplicas: 2})
+	if err != nil {
+		t.Fatalf("Reconcile: %v", err)
+	}
+	if action.Type != ActionUpdate {
+		t.Fatalf("action = %q, want update", action.Type)
+	}
+	if action.Deployment.Pod.Container.Image != "example/support:v1" {
+		t.Fatalf("update should carry the desired image, got %q", action.Deployment.Pod.Container.Image)
+	}
+}
+
+func TestReconcileProgressingWhenUnderReplicated(t *testing.T) {
+	want, _ := MapDeployment(agentResource())
+	_, conds, err := Reconcile(agentResource(), Observed{Deployment: &want, ReadyReplicas: 1})
+	if err != nil {
+		t.Fatalf("Reconcile: %v", err)
+	}
+	if ready := findCondition(conds, "Ready"); ready == nil || ready.Status != "False" || ready.Reason != "Progressing" {
+		t.Fatalf("ready condition = %+v, want False/Progressing", ready)
+	}
+}
+
+func TestReconcileErrorOnInvalidSpec(t *testing.T) {
+	// Missing image → MapDeployment fails → Error condition, no action.
+	_, conds, err := Reconcile(Resource{Kind: KindService, Name: "api"}, Observed{})
+	if err == nil {
+		t.Fatal("Reconcile should error on an invalid spec")
+	}
+	if e := findCondition(conds, "Error"); e == nil || e.Status != "True" || e.Reason != "InvalidSpec" {
+		t.Fatalf("error condition = %+v, want True/InvalidSpec", e)
+	}
+}
+
+func findCondition(conds []Condition, typ string) *Condition {
+	for i := range conds {
+		if conds[i].Type == typ {
+			return &conds[i]
+		}
+	}
+	return nil
+}

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

flow/otel.go

@@ -92,7 +92,10 @@ func (f *Flow) runStepSpan(ctx context.Context, step Step, in State) (State, int
 			span.SetAttributes(attribute.String(AttrFlowVerificationStatus, "failed"))
 		}
 	}
-	if err != nil {
+	if a, ok := isAwaitInput(err); ok {
+		// A suspend is normal control flow, not a step error.
+		span.SetStatus(codes.Ok, "waiting: "+a.Key)
+	} else if err != nil {
 		span.RecordError(err)
 		span.SetAttributes(attribute.String(AttrFlowErrorKind, string(ai.ClassifyError(err))))
 		span.SetStatus(codes.Error, err.Error())

flow/steps.go

@@ -4,6 +4,7 @@ import (
 	"bytes"
 	"context"
 	"encoding/json"
+	"errors"
 	"fmt"
 	"sort"
 	"text/template"
@@ -110,7 +111,8 @@ type Run struct {
 	Flow     string       `json:"flow"`
 	State    State        `json:"state"`
 	Steps    []StepRecord `json:"steps"`
-	Status   string       `json:"status"` // running | done | failed
+	Status   string       `json:"status"` // running | waiting | done | failed
+	Await    *AwaitState  `json:"await,omitempty"`
 	Started  time.Time    `json:"started"`
 	Updated  time.Time    `json:"updated"`
 }
@@ -336,6 +338,54 @@ func LLM(prompt string) StepFunc {
 	}
 }
 
+// AwaitInput is the control signal a step returns (via Await) to suspend a run
+// pending external input. runFrom recognizes it, checkpoints the run as
+// "waiting", and returns cleanly — a suspend is not a failure. ResumeWith
+// injects the input and continues.
+type AwaitInput struct {
+	Key    string // labels what is awaited (e.g. "approval")
+	Prompt string // human-facing description of the input needed
+}
+
+func (e *AwaitInput) Error() string {
+	if e.Prompt != "" {
+		return fmt.Sprintf("flow: awaiting input %q: %s", e.Key, e.Prompt)
+	}
+	return fmt.Sprintf("flow: awaiting input %q", e.Key)
+}
+
+// AwaitState records, on a suspended run, what it is waiting for.
+type AwaitState struct {
+	Step   string `json:"step"`
+	Key    string `json:"key"`
+	Prompt string `json:"prompt,omitempty"`
+}
+
+func isAwaitInput(err error) (*AwaitInput, bool) {
+	var a *AwaitInput
+	if errors.As(err, &a) {
+		return a, true
+	}
+	return nil, false
+}
+
+// Await is a StepFunc that suspends the run pending external input. The run is
+// checkpointed with status "waiting" and returned cleanly; a later call to
+// Flow.ResumeWith(ctx, runID, input) completes this step with the injected
+// input and continues to the next step. key labels what is awaited (surfaced on
+// the run and via Flow.Waiting); prompt describes the input needed.
+func Await(key, prompt string) StepFunc {
+	return func(_ context.Context, in State) (State, error) {
+		return in, &AwaitInput{Key: key, Prompt: prompt}
+	}
+}
+
+// AwaitStep is a convenience for a named await step:
+// Step{Name: name, Run: Await(key, prompt)}.
+func AwaitStep(name, key, prompt string) Step {
+	return Step{Name: name, Run: Await(key, prompt)}
+}
+
 // startRun begins a fresh run of the flow's steps with the given input.
 func (f *Flow) startRun(ctx context.Context, data string) (Run, error) {
 	if err := validateSteps(f.opts.Steps); err != nil {
@@ -421,13 +471,79 @@ func (f *Flow) Pending(ctx context.Context) ([]Run, error) {
 	}
 	var out []Run
 	for _, r := range all {
-		if r.Flow == f.name && r.Status != "done" {
+		// Waiting runs need injected input (ResumeWith), not a restart, so a
+		// recovery loop (ResumePending) should not pick them up.
+		if r.Flow == f.name && r.Status != "done" && r.Status != "waiting" {
 			out = append(out, r)
 		}
 	}
 	return out, nil
 }
 
+// Waiting returns this flow's runs suspended awaiting external input, each with
+// its Await metadata, so a caller can prompt for and inject the needed input
+// with ResumeWith.
+func (f *Flow) Waiting(ctx context.Context) ([]Run, error) {
+	if f.checkpoint == nil {
+		return nil, nil
+	}
+	all, err := f.checkpoint.List(ctx)
+	if err != nil {
+		return nil, err
+	}
+	var out []Run
+	for _, r := range all {
+		if r.Flow == f.name && r.Status == "waiting" {
+			out = append(out, r)
+		}
+	}
+	return out, nil
+}
+
+// ResumeWith completes a suspended (waiting) run: it injects input for the
+// awaited step — the input becomes that step's output state — and continues
+// from the next step. It errors if the run is not waiting for input.
+func (f *Flow) ResumeWith(ctx context.Context, runID, input string) error {
+	ctx, cancel := f.withTimeout(ctx)
+	defer cancel()
+
+	if err := validateSteps(f.opts.Steps); err != nil {
+		return err
+	}
+	if f.checkpoint == nil {
+		return fmt.Errorf("flow %s has no checkpoint configured", f.name)
+	}
+	run, ok, err := f.checkpoint.Load(ctx, runID)
+	if err != nil {
+		return err
+	}
+	if !ok {
+		return fmt.Errorf("run %s not found", runID)
+	}
+	if run.Status != "waiting" {
+		return fmt.Errorf("run %s is not waiting for input (status %q)", runID, run.Status)
+	}
+	steps := f.opts.Steps
+	i := stepIndex(steps, run.State.Stage)
+	if i < 0 {
+		return fmt.Errorf("run %s is waiting at unknown step %q", runID, run.State.Stage)
+	}
+	// The awaited step is satisfied by the injected input; record it done and
+	// advance so runFrom re-enters at the next step.
+	run.Steps[i].Status = "done"
+	run.Steps[i].Result = truncate(input, 200)
+	run.State.Data = []byte(input)
+	if i+1 < len(steps) {
+		run.State.Stage = steps[i+1].Name
+	} else {
+		run.State.Stage = ""
+	}
+	run.Await = nil
+	run.Status = "running"
+	_, err = f.runFrom(ctx, run)
+	return err
+}
+
 // runFrom executes steps from the run's current Stage to the end,
 // checkpointing before and after each step.
 func (f *Flow) runFrom(ctx context.Context, run Run) (Run, error) {
@@ -464,6 +580,19 @@ func (f *Flow) runFrom(ctx context.Context, run Run) (Run, error) {
 		out, attempts, verification, err := f.runStepSpan(ctx, step, run.State)
 		run.Steps[i].Attempts = attempts
 		applyVerificationRecord(&run.Steps[i], verification)
+		if await, ok := isAwaitInput(err); ok {
+			// Suspend the run pending external input — checkpoint and return
+			// cleanly (not a failure). ResumeWith injects the input later.
+			run.Steps[i].Status = "waiting"
+			run.Status = "waiting"
+			run.Await = &AwaitState{Step: step.Name, Key: await.Key, Prompt: await.Prompt}
+			if saveErr := f.save(ctx, run); saveErr != nil {
+				spanErr = saveErr
+				return run, saveErr
+			}
+			f.log.Logf(logger.InfoLevel, "Flow %s run %s waiting for input %q at step %q", f.name, run.ID, await.Key, step.Name)
+			return run, nil
+		}
 		if err != nil {
 			spanErr = err
 			run.Steps[i].Status = "failed"
@@ -537,6 +666,11 @@ func (f *Flow) runStep(ctx context.Context, step Step, in State) (State, int, Ve
 			attemptCtx = ai.WithRunInfo(ctx, info)
 		}
 		out, err := step.Run(attemptCtx, in)
+		// An await signal is control flow, not a failure: suspend immediately
+		// without retrying or grading.
+		if _, ok := isAwaitInput(err); ok {
+			return in, attempt, lastVerification, err
+		}
 		if err == nil && step.Verify != nil {
 			lastVerification, err = step.Verify(attemptCtx, out)
 			if err == nil && !lastVerification.Passed {

flow/steps_test.go

@@ -87,6 +87,93 @@ func TestFlowCheckpointResume(t *testing.T) {
 	}
 }
 
+func TestFlowAwaitAndResumeWith(t *testing.T) {
+	mem := store.NewMemoryStore()
+	var firstCalls int
+	var secondInput string
+
+	steps := []Step{
+		{Name: "first", Run: func(_ context.Context, in State) (State, error) {
+			firstCalls++
+			in.Data = []byte("first-done")
+			return in, nil
+		}},
+		AwaitStep("approval", "approve", "Approve to continue?"),
+		{Name: "second", Run: func(_ context.Context, in State) (State, error) {
+			secondInput = in.String()
+			in.Data = []byte("second-done")
+			return in, nil
+		}},
+	}
+
+	f := New("hitl", WithCheckpoint(StoreCheckpoint(mem, "hitl")), Steps(steps...))
+
+	// Execute suspends at the await step — a clean return, not an error.
+	if err := f.Execute(context.Background(), "start"); err != nil {
+		t.Fatalf("Execute should suspend cleanly, got %v", err)
+	}
+	if firstCalls != 1 {
+		t.Fatalf("first step calls = %d, want 1", firstCalls)
+	}
+
+	// A waiting run is not pending (restart), it needs input.
+	if pend, _ := f.Pending(context.Background()); len(pend) != 0 {
+		t.Errorf("a waiting run must not be pending, got %d", len(pend))
+	}
+	waiting, err := f.Waiting(context.Background())
+	if err != nil {
+		t.Fatal(err)
+	}
+	if len(waiting) != 1 {
+		t.Fatalf("waiting runs = %d, want 1", len(waiting))
+	}
+	w := waiting[0]
+	if w.Status != "waiting" || w.Await == nil || w.Await.Key != "approve" ||
+		w.Await.Prompt != "Approve to continue?" || w.Await.Step != "approval" {
+		t.Fatalf("await metadata = %+v (status %q)", w.Await, w.Status)
+	}
+	if w.State.Stage != "approval" {
+		t.Fatalf("waiting stage = %q, want approval", w.State.Stage)
+	}
+
+	// Injecting input completes the awaited step and runs the rest.
+	if err := f.ResumeWith(context.Background(), w.ID, "approved"); err != nil {
+		t.Fatalf("ResumeWith: %v", err)
+	}
+	if firstCalls != 1 {
+		t.Errorf("completed step re-ran on resume; first calls = %d", firstCalls)
+	}
+	if secondInput != "approved" {
+		t.Errorf("second step input = %q, want the injected 'approved'", secondInput)
+	}
+	if wr, _ := f.Waiting(context.Background()); len(wr) != 0 {
+		t.Errorf("no waiting runs after resume, got %d", len(wr))
+	}
+	runs, _ := StoreCheckpoint(mem, "hitl").List(context.Background())
+	if len(runs) != 1 || runs[0].Status != "done" {
+		t.Fatalf("run should be done after resume, got %+v", runs)
+	}
+	if runs[0].Await != nil {
+		t.Errorf("await metadata should be cleared after resume, got %+v", runs[0].Await)
+	}
+}
+
+func TestFlowResumeWithRejectsNonWaiting(t *testing.T) {
+	mem := store.NewMemoryStore()
+	f := New("hitl2", WithCheckpoint(StoreCheckpoint(mem, "hitl2")),
+		Steps(Step{Name: "only", Run: func(_ context.Context, in State) (State, error) { return in, nil }}))
+	if err := f.Execute(context.Background(), "x"); err != nil {
+		t.Fatalf("Execute: %v", err)
+	}
+	runs, _ := StoreCheckpoint(mem, "hitl2").List(context.Background())
+	if len(runs) != 1 {
+		t.Fatalf("runs = %d", len(runs))
+	}
+	if err := f.ResumeWith(context.Background(), runs[0].ID, "input"); err == nil {
+		t.Error("ResumeWith on a completed (non-waiting) run should error")
+	}
+}
+
 func TestFlowStepContextIncludesRunInfo(t *testing.T) {
 	var got ai.RunInfo
 	step := Step{Name: "inspect", Run: func(ctx context.Context, in State) (State, error) {

gateway/a2a/a2a.go

@@ -27,12 +27,14 @@ package a2a
 
 import (
 	"context"
+	"crypto/ed25519"
 	"encoding/json"
 	"errors"
 	"fmt"
 	"io"
 	"log"
 	"net/http"
+	"net/url"
 	"strings"
 	"sync"
 	"time"
@@ -64,6 +66,20 @@ type Options struct {
 	Client client.Client
 	// Logger for startup/debug output (defaults to log.Default()).
 	Logger *log.Logger
+	// AllowPushURL authorizes an outbound push-notification callback URL
+	// (tasks/pushNotificationConfig/set). Return a non-nil error to reject it.
+	// When nil, a default SSRF-safe policy applies: only http/https URLs whose
+	// host does not resolve to a loopback, private, link-local, or unspecified
+	// address are allowed, and the connection is pinned to that check at dial
+	// time (DNS-rebinding safe). Set this to permit a trusted in-cluster
+	// receiver, or to narrow delivery to an allowlist.
+	AllowPushURL func(*url.URL) error
+	// AP2PublicKey, when set, verifies AP2 payment/checkout mandates carried on
+	// incoming A2A messages against this Ed25519 key and records the outcome in
+	// each task's ap2Verifications (signature + task/context binding). When
+	// unset, mandates are carried through unverified. This is opt-in so the
+	// default flow stays free of a payment trust decision.
+	AP2PublicKey ed25519.PublicKey
 }
 
 // Gateway serves the A2A protocol over HTTP for the registry's agents.
@@ -87,7 +103,20 @@ func New(opts Options) *Gateway {
 		opts.BaseURL = "http://localhost" + opts.Address
 	}
 	opts.BaseURL = strings.TrimRight(opts.BaseURL, "/")
-	return &Gateway{opts: opts, disp: newDispatcher()}
+	g := &Gateway{opts: opts, disp: newDispatcher()}
+	if opts.AllowPushURL != nil {
+		// Operator owns the trust decision: use their policy and skip the
+		// built-in private-IP dial guard so trusted in-cluster hosts resolve.
+		g.disp.allowPushURL = opts.AllowPushURL
+		g.disp.guardPushDial = false
+	}
+	if len(opts.AP2PublicKey) > 0 {
+		pub := opts.AP2PublicKey
+		g.disp.ap2Verify = func(s AP2SignedMandate, task Task) AP2Verification {
+			return VerifyAP2ForTask(s, pub, task, nil)
+		}
+	}
+	return g
 }
 
 // Invoke runs an agent for one message and returns its reply. It is the
@@ -98,16 +127,55 @@ type Invoke func(ctx context.Context, text string) (string, error)
 // StreamInvoke runs an agent for one message and returns streaming output chunks.
 type StreamInvoke func(ctx context.Context, text string) (ai.Stream, error)
 
+// AgentHandlerOption configures an embedded A2A agent handler.
+type AgentHandlerOption func(*dispatcher)
+
+// WithPushURLPolicy sets the push-notification callback URL policy for an
+// embedded agent handler (the analog of Options.AllowPushURL on the gateway).
+// Return a non-nil error to reject a URL. Without it, the default SSRF-safe
+// policy applies. Supplying a policy also disables the built-in private-IP dial
+// guard, so a trusted in-cluster receiver resolves.
+func WithPushURLPolicy(allow func(*url.URL) error) AgentHandlerOption {
+	return func(d *dispatcher) {
+		if allow == nil {
+			return
+		}
+		d.allowPushURL = allow
+		d.guardPushDial = false
+	}
+}
+
+// WithAP2PublicKey verifies AP2 mandates carried on incoming messages against
+// pub (the embedded-handler analog of Options.AP2PublicKey), recording the
+// outcome in each task's ap2Verifications. Without it, mandates are carried
+// unverified.
+func WithAP2PublicKey(pub ed25519.PublicKey) AgentHandlerOption {
+	return func(d *dispatcher) {
+		if len(pub) == 0 {
+			return
+		}
+		d.ap2Verify = func(s AP2SignedMandate, task Task) AP2Verification {
+			return VerifyAP2ForTask(s, pub, task, nil)
+		}
+	}
+}
+
 // NewAgentHandler returns an http.Handler that serves the A2A protocol
 // for a single agent: its Agent Card at / and /.well-known/agent.json,
 // and the JSON-RPC endpoint at /. invoke runs the agent. This is what an
 // agent embeds to speak A2A directly, without a separate gateway.
-func NewAgentHandler(card AgentCard, invoke Invoke) http.Handler {
+func NewAgentHandler(card AgentCard, invoke Invoke, opts ...AgentHandlerOption) http.Handler {
 	d := newDispatcher()
+	for _, o := range opts {
+		o(d)
+	}
 	mux := http.NewServeMux()
 	card.URL = strings.TrimRight(card.URL, "/")
 	serveCard := func(w http.ResponseWriter, _ *http.Request) { writeJSON(w, http.StatusOK, card) }
 	mux.HandleFunc("GET /{$}", serveCard)
+	// A2A 0.3.0 discovery is /.well-known/agent-card.json; agent.json is the
+	// pre-0.3 alias, kept so existing clients don't break.
+	mux.HandleFunc("GET /.well-known/agent-card.json", serveCard)
 	mux.HandleFunc("GET /.well-known/agent.json", serveCard)
 	mux.HandleFunc("POST /{$}", func(w http.ResponseWriter, r *http.Request) { d.serve(w, r, invoke) })
 	return mux
@@ -115,12 +183,16 @@ func NewAgentHandler(card AgentCard, invoke Invoke) http.Handler {
 
 // NewAgentStreamHandler is like NewAgentHandler, but serves A2A message/stream
 // by forwarding model chunks as server-sent task updates when stream is non-nil.
-func NewAgentStreamHandler(card AgentCard, invoke Invoke, stream StreamInvoke) http.Handler {
+func NewAgentStreamHandler(card AgentCard, invoke Invoke, stream StreamInvoke, opts ...AgentHandlerOption) http.Handler {
 	d := newDispatcher()
+	for _, o := range opts {
+		o(d)
+	}
 	mux := http.NewServeMux()
 	card.URL = strings.TrimRight(card.URL, "/")
 	serveCard := func(w http.ResponseWriter, _ *http.Request) { writeJSON(w, http.StatusOK, card) }
 	mux.HandleFunc("GET /{$}", serveCard)
+	mux.HandleFunc("GET /.well-known/agent-card.json", serveCard)
 	mux.HandleFunc("GET /.well-known/agent.json", serveCard)
 	mux.HandleFunc("POST /{$}", func(w http.ResponseWriter, r *http.Request) { d.serveWithStream(w, r, invoke, stream) })
 	return mux
@@ -139,15 +211,19 @@ func (g *Gateway) Handler() http.Handler {
 	// Discovery: a directory of all agent cards.
 	mux.HandleFunc("GET /agents", g.handleList)
 	// Per-agent card (served at the agent's url and at its well-known path).
+	// A2A 0.3.0 uses agent-card.json; agent.json is the pre-0.3 alias.
 	mux.HandleFunc("GET /agents/{name}", g.handleCard)
+	mux.HandleFunc("GET /agents/{name}/.well-known/agent-card.json", g.handleCard)
 	mux.HandleFunc("GET /agents/{name}/.well-known/agent.json", g.handleCard)
 	mux.HandleFunc("GET /agents/{name}/skills/{skill}", g.handleSkillCard)
+	mux.HandleFunc("GET /agents/{name}/skills/{skill}/.well-known/agent-card.json", g.handleSkillCard)
 	mux.HandleFunc("GET /agents/{name}/skills/{skill}/.well-known/agent.json", g.handleSkillCard)
 	// Per-agent JSON-RPC endpoint.
 	mux.HandleFunc("POST /agents/{name}", g.handleRPC)
 	mux.HandleFunc("POST /agents/{name}/skills/{skill}", g.handleSkillRPC)
 	// Top-level well-known: serve the single agent's card if there's
 	// exactly one, otherwise point to the directory.
+	mux.HandleFunc("GET /.well-known/agent-card.json", g.handleWellKnown)
 	mux.HandleFunc("GET /.well-known/agent.json", g.handleWellKnown)
 	return mux
 }
@@ -222,6 +298,39 @@ type Artifact struct {
 	Parts      []Part `json:"parts"`
 }
 
+// TaskStatusUpdateEvent is an A2A streaming event reporting a change in a
+// task's status. External SSE clients parse stream events by `kind` and stop
+// on the event whose `final` is true — a full Task snapshot (which older
+// versions emitted) carries neither, so strict clients never terminate.
+type TaskStatusUpdateEvent struct {
+	TaskID    string     `json:"taskId"`
+	ContextID string     `json:"contextId"`
+	Kind      string     `json:"kind"` // "status-update"
+	Status    TaskStatus `json:"status"`
+	Final     bool       `json:"final"`
+}
+
+// TaskArtifactUpdateEvent is an A2A streaming event carrying an artifact (or,
+// with Append, one incremental chunk of one).
+type TaskArtifactUpdateEvent struct {
+	TaskID    string   `json:"taskId"`
+	ContextID string   `json:"contextId"`
+	Kind      string   `json:"kind"` // "artifact-update"
+	Artifact  Artifact `json:"artifact"`
+	Append    bool     `json:"append,omitempty"`
+	LastChunk bool     `json:"lastChunk,omitempty"`
+}
+
+func statusUpdateEvent(t *Task, final bool) TaskStatusUpdateEvent {
+	return TaskStatusUpdateEvent{
+		TaskID:    t.ID,
+		ContextID: t.ContextID,
+		Kind:      "status-update",
+		Status:    t.Status,
+		Final:     final,
+	}
+}
+
 // Task is the unit of work returned by message/send and tasks/get.
 type Task struct {
 	ID               string             `json:"id"`
@@ -469,10 +578,26 @@ type dispatcher struct {
 	pushConfigs map[string]PushNotificationConfig
 	watchers    map[string]map[chan *Task]struct{}
 	order       []string // task ids in insertion order, for bounded eviction
+
+	// allowPushURL authorizes an outbound push-notification callback URL; nil
+	// means the default SSRF-safe policy. guardPushDial applies the private-IP
+	// dial guard (on unless an operator supplied a custom policy).
+	allowPushURL  func(*url.URL) error
+	guardPushDial bool
+
+	// ap2Verify, when non-nil, verifies each AP2 mandate carried on a task and
+	// records the result in the task's AP2Verifications. Nil = carry unverified.
+	ap2Verify func(AP2SignedMandate, Task) AP2Verification
 }
 
 func newDispatcher() *dispatcher {
-	return &dispatcher{tasks: map[string]*Task{}, pushConfigs: map[string]PushNotificationConfig{}, watchers: map[string]map[chan *Task]struct{}{}}
+	return &dispatcher{
+		tasks:         map[string]*Task{},
+		pushConfigs:   map[string]PushNotificationConfig{},
+		watchers:      map[string]map[chan *Task]struct{}{},
+		allowPushURL:  defaultPushURLPolicy,
+		guardPushDial: true,
+	}
 }
 
 func (d *dispatcher) serve(w http.ResponseWriter, r *http.Request, invoke Invoke) {
@@ -534,14 +659,11 @@ func (d *dispatcher) stream(ctx context.Context, w http.ResponseWriter, req rpcR
 		writeRPC(w, req.ID, nil, e)
 		return
 	}
-	w.Header().Set("Content-Type", "text/event-stream")
-	w.Header().Set("Cache-Control", "no-cache")
-	w.Header().Set("Connection", "keep-alive")
-	w.WriteHeader(http.StatusOK)
-	_ = json.NewEncoder(sseWriter{w: w}).Encode(rpcResponse{JSONRPC: "2.0", ID: req.ID, Result: task})
-	if f, ok := w.(http.Flusher); ok {
-		f.Flush()
-	}
+	enc, flush := sseResponse(w)
+	// The Task snapshot first (carries ids and the final artifact), then a
+	// terminal status-update so external SSE clients see `final:true` and stop.
+	writeSSE(enc, flush, req.ID, task)
+	writeSSE(enc, flush, req.ID, statusUpdateEvent(task, true))
 }
 
 func (d *dispatcher) streamChunks(ctx context.Context, w http.ResponseWriter, req rpcRequest, invoke StreamInvoke, fallback Invoke) {
@@ -565,46 +687,53 @@ func (d *dispatcher) streamChunks(ctx context.Context, w http.ResponseWriter, re
 		return
 	}
 	defer stream.Close()
-	w.Header().Set("Content-Type", "text/event-stream")
-	w.Header().Set("Cache-Control", "no-cache")
-	w.Header().Set("Connection", "keep-alive")
-	w.WriteHeader(http.StatusOK)
-	enc := json.NewEncoder(sseWriter{w: w})
-	flush := func() {
-		if f, ok := w.(http.Flusher); ok {
-			f.Flush()
-		}
-	}
+	enc, flush := sseResponse(w)
 	taskID := uuid.New().String()
 	contextID := p.Message.ContextID
 	if contextID == "" {
 		contextID = uuid.New().String()
 	}
+	// One artifact id for the whole stream so append:true chunks target it.
+	artifactID := uuid.New().String()
+
+	// Open with the Task snapshot (working) so the client learns the ids.
+	initial := taskFromReplyWithIDs(p.Message, "", stateWorking, taskID, contextID)
+	d.store(initial)
+	writeSSE(enc, flush, req.ID, initial)
+
 	var reply strings.Builder
 	for {
 		chunk, err := stream.Recv()
 		if err == io.EOF {
 			task := taskFromReplyWithIDs(p.Message, reply.String(), stateCompleted, taskID, contextID)
 			d.store(task)
-			_ = enc.Encode(rpcResponse{JSONRPC: "2.0", ID: req.ID, Result: task})
-			flush()
+			// Spec-shaped terminal: a status-update with final:true — not a
+			// full Task snapshot, which carries no terminal marker.
+			writeSSE(enc, flush, req.ID, statusUpdateEvent(task, true))
 			return
 		}
 		if err != nil {
 			task := taskFromReplyWithIDs(p.Message, "error: "+err.Error(), stateFailed, taskID, contextID)
 			d.store(task)
-			_ = enc.Encode(rpcResponse{JSONRPC: "2.0", ID: req.ID, Result: task, Error: &rpcError{Code: errInternal, Message: err.Error()}})
-			flush()
+			// A failed status-update (final) — never `result` and `error`
+			// together in one response, which strict clients reject.
+			writeSSE(enc, flush, req.ID, statusUpdateEvent(task, true))
 			return
 		}
 		if chunk == nil || chunk.Reply == "" {
 			continue
 		}
 		reply.WriteString(chunk.Reply)
-		task := taskFromReplyWithIDs(p.Message, reply.String(), stateWorking, taskID, contextID)
-		d.store(task)
-		_ = enc.Encode(rpcResponse{JSONRPC: "2.0", ID: req.ID, Result: task})
-		flush()
+		// Emit the delta as an append artifact-update; keep the stored task
+		// current for tasks/get and resubscribe watchers.
+		d.store(taskFromReplyWithIDs(p.Message, reply.String(), stateWorking, taskID, contextID))
+		writeSSE(enc, flush, req.ID, TaskArtifactUpdateEvent{
+			TaskID:    taskID,
+			ContextID: contextID,
+			Kind:      "artifact-update",
+			Artifact:  Artifact{ArtifactID: artifactID, Parts: []Part{{Kind: "text", Text: chunk.Reply}}},
+			Append:    true,
+		})
 	}
 }
 
@@ -653,20 +782,16 @@ func (d *dispatcher) resubscribe(ctx context.Context, w http.ResponseWriter, req
 	}
 	defer unsubscribe()
 
-	w.Header().Set("Content-Type", "text/event-stream")
-	w.Header().Set("Cache-Control", "no-cache")
-	w.Header().Set("Connection", "keep-alive")
-	w.WriteHeader(http.StatusOK)
-	enc := json.NewEncoder(sseWriter{w: w})
-	flush := func() {
-		if f, ok := w.(http.Flusher); ok {
-			f.Flush()
-		}
-	}
+	enc, flush := sseResponse(w)
 	writeEvent := func(t *Task) bool {
-		_ = enc.Encode(rpcResponse{JSONRPC: "2.0", ID: req.ID, Result: t})
-		flush()
-		return isTerminal(t.Status.State)
+		writeSSE(enc, flush, req.ID, t)
+		if isTerminal(t.Status.State) {
+			// Close the stream with a spec-shaped terminal marker so external
+			// clients see `final:true`.
+			writeSSE(enc, flush, req.ID, statusUpdateEvent(t, true))
+			return true
+		}
+		return false
 	}
 	if writeEvent(task) {
 		return
@@ -710,6 +835,11 @@ func (d *dispatcher) setPushConfig(w http.ResponseWriter, req rpcRequest) {
 		writeRPC(w, req.ID, nil, &rpcError{Code: errInvalidParams, Message: "invalid params"})
 		return
 	}
+	// Reject SSRF-unsafe callback targets before storing them.
+	if err := d.checkPushURL(p.PushNotificationConfig.URL); err != nil {
+		writeRPC(w, req.ID, nil, &rpcError{Code: errInvalidParams, Message: "push notification url not allowed"})
+		return
+	}
 	d.mu.Lock()
 	task := d.tasks[p.ID]
 	if task != nil {
@@ -765,6 +895,15 @@ func (g *Gateway) callAgent(ctx context.Context, name, message string) (string,
 // ---------------------------------------------------------------------------
 
 func (d *dispatcher) store(t *Task) {
+	// Verify any AP2 mandates carried on the task (opt-in) and surface the
+	// outcome so a downstream paid path can trust — or reject — the mandate.
+	if d.ap2Verify != nil && len(t.AP2Mandates) > 0 && len(t.AP2Verifications) == 0 {
+		v := make([]AP2Verification, 0, len(t.AP2Mandates))
+		for _, m := range t.AP2Mandates {
+			v = append(v, d.ap2Verify(m, *t))
+		}
+		t.AP2Verifications = v
+	}
 	d.mu.Lock()
 	_, exists := d.tasks[t.ID]
 	d.tasks[t.ID] = t
@@ -878,6 +1017,11 @@ func (d *dispatcher) deliverPush(taskID string, task *Task) {
 	if !ok || cfg.URL == "" || task == nil {
 		return
 	}
+	// Defense in depth: re-validate the callback URL at delivery time in case
+	// the policy tightened or the config was set before it applied.
+	if err := d.checkPushURL(cfg.URL); err != nil {
+		return
+	}
 	body, err := json.Marshal(task)
 	if err != nil {
 		return
@@ -892,7 +1036,7 @@ func (d *dispatcher) deliverPush(taskID string, task *Task) {
 	if cfg.Token != "" {
 		req.Header.Set("Authorization", "Bearer "+cfg.Token)
 	}
-	resp, err := http.DefaultClient.Do(req)
+	resp, err := d.pushClient().Do(req)
 	if err == nil && resp.Body != nil {
 		_ = resp.Body.Close()
 	}
@@ -1029,6 +1173,27 @@ func requestContext(parent context.Context) context.Context {
 	return ctx
 }
 
+// sseResponse writes the SSE response headers and returns an encoder and a
+// flush func for emitting `data:`-framed JSON-RPC events.
+func sseResponse(w http.ResponseWriter) (*json.Encoder, func()) {
+	w.Header().Set("Content-Type", "text/event-stream")
+	w.Header().Set("Cache-Control", "no-cache")
+	w.Header().Set("Connection", "keep-alive")
+	w.WriteHeader(http.StatusOK)
+	enc := json.NewEncoder(sseWriter{w: w})
+	return enc, func() {
+		if f, ok := w.(http.Flusher); ok {
+			f.Flush()
+		}
+	}
+}
+
+// writeSSE emits one JSON-RPC event (result only — never with an error) and flushes.
+func writeSSE(enc *json.Encoder, flush func(), id json.RawMessage, result any) {
+	_ = enc.Encode(rpcResponse{JSONRPC: "2.0", ID: id, Result: result})
+	flush()
+}
+
 type sseWriter struct {
 	w http.ResponseWriter
 }

gateway/a2a/a2a_test.go

@@ -9,6 +9,7 @@ import (
 	"io"
 	"net/http"
 	"net/http/httptest"
+	"net/url"
 	"strings"
 	"testing"
 	"time"
@@ -99,6 +100,38 @@ func TestAgentCardFromRegistry(t *testing.T) {
 	}
 }
 
+// A2A 0.3.0 discovery is /.well-known/agent-card.json. The card must be
+// reachable there (canonical) as well as at the legacy agent.json alias, both
+// per-agent and at the single-agent top level.
+func TestAgentCardCanonicalWellKnownPath(t *testing.T) {
+	ts, cleanup := newGatewayWithAgent(t)
+	defer cleanup()
+
+	for _, path := range []string{
+		"/agents/echo/.well-known/agent-card.json",
+		"/agents/echo/.well-known/agent.json",
+		"/agents/echo/skills/task/.well-known/agent-card.json",
+	} {
+		resp, err := http.Get(ts.URL + path)
+		if err != nil {
+			t.Fatalf("get %s: %v", path, err)
+		}
+		if resp.StatusCode != http.StatusOK {
+			resp.Body.Close()
+			t.Fatalf("%s status = %d, want 200", path, resp.StatusCode)
+		}
+		var card AgentCard
+		if err := json.NewDecoder(resp.Body).Decode(&card); err != nil {
+			resp.Body.Close()
+			t.Fatalf("%s decode card: %v", path, err)
+		}
+		resp.Body.Close()
+		if card.Name != "echo" {
+			t.Errorf("%s card name = %q, want echo", path, card.Name)
+		}
+	}
+}
+
 func TestSkillEndpointServesFocusedCardAndRoutesRPC(t *testing.T) {
 	ts, cleanup := newGatewayWithAgent(t)
 	defer cleanup()
@@ -191,6 +224,10 @@ func TestMessageSendContinuesExistingTask(t *testing.T) {
 
 func TestPushNotificationConfigDeliversTaskUpdates(t *testing.T) {
 	d := newDispatcher()
+	// The test receiver is a loopback httptest server; authorize it the way a
+	// deployment would authorize a trusted in-cluster push receiver.
+	d.allowPushURL = func(*url.URL) error { return nil }
+	d.guardPushDial = false
 	updates := make(chan Task, 2)
 	push := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
 		if got := r.Header.Get("Authorization"); got != "Bearer secret" {
@@ -335,6 +372,67 @@ func (s *sliceStream) Recv() (*ai.Response, error) {
 
 func (s *sliceStream) Close() error { return nil }
 
+// streamEvent is one decoded SSE JSON-RPC event from a message/stream response.
+// A2A streams carry heterogeneous results (Task, status-update, artifact-update)
+// discriminated by `kind`, so we keep the raw result and decode on demand.
+type streamEvent struct {
+	Result json.RawMessage `json:"result"`
+	Error  *rpcError       `json:"error"`
+}
+
+func (e streamEvent) kind() string {
+	var k struct {
+		Kind string `json:"kind"`
+	}
+	_ = json.Unmarshal(e.Result, &k)
+	return k.Kind
+}
+
+func (e streamEvent) task(t *testing.T) Task {
+	t.Helper()
+	var task Task
+	if err := json.Unmarshal(e.Result, &task); err != nil {
+		t.Fatalf("decode task event: %v", err)
+	}
+	return task
+}
+
+func (e streamEvent) status(t *testing.T) TaskStatusUpdateEvent {
+	t.Helper()
+	var s TaskStatusUpdateEvent
+	if err := json.Unmarshal(e.Result, &s); err != nil {
+		t.Fatalf("decode status-update event: %v", err)
+	}
+	return s
+}
+
+func (e streamEvent) artifactUpdate(t *testing.T) TaskArtifactUpdateEvent {
+	t.Helper()
+	var a TaskArtifactUpdateEvent
+	if err := json.Unmarshal(e.Result, &a); err != nil {
+		t.Fatalf("decode artifact-update event: %v", err)
+	}
+	return a
+}
+
+// collectSSE parses the `data:`-framed JSON-RPC events from an SSE body.
+func collectSSE(t *testing.T, body string) []streamEvent {
+	t.Helper()
+	var events []streamEvent
+	for _, line := range strings.Split(strings.TrimSpace(body), "\n") {
+		line = strings.TrimSpace(strings.TrimPrefix(strings.TrimSpace(line), "data:"))
+		if line == "" {
+			continue
+		}
+		var e streamEvent
+		if err := json.Unmarshal([]byte(line), &e); err != nil {
+			t.Fatalf("decode event %q: %v", line, err)
+		}
+		events = append(events, e)
+	}
+	return events
+}
+
 func TestMessageStreamChunksStoreFinalTask(t *testing.T) {
 	d := newDispatcher()
 	body := `{"jsonrpc":"2.0","id":1,"method":"message/stream","params":{"message":{"role":"user","parts":[{"kind":"text","text":"ping"}],"kind":"message"}}}`
@@ -351,47 +449,61 @@ func TestMessageStreamChunksStoreFinalTask(t *testing.T) {
 	if ct := rr.Result().Header.Get("Content-Type"); !strings.HasPrefix(ct, "text/event-stream") {
 		t.Fatalf("content-type = %q, want text/event-stream", ct)
 	}
-	var events []struct {
-		Result Task      `json:"result"`
-		Error  *rpcError `json:"error"`
+	events := collectSSE(t, rr.Body.String())
+	// Opening Task snapshot + one append artifact-update per chunk + terminal
+	// status-update.
+	if len(events) != 4 {
+		t.Fatalf("events = %d, want 4; body %s", len(events), rr.Body.String())
 	}
-	for _, line := range strings.Split(strings.TrimSpace(rr.Body.String()), "\n") {
-		line = strings.TrimSpace(line)
-		if line == "" {
-			continue
-		}
-		line = strings.TrimPrefix(line, "data: ")
-		var event struct {
-			Result Task      `json:"result"`
-			Error  *rpcError `json:"error"`
-		}
-		if err := json.Unmarshal([]byte(line), &event); err != nil {
-			t.Fatalf("decode event %q: %v", line, err)
-		}
-		events = append(events, event)
-	}
-	if len(events) != 3 {
-		t.Fatalf("events = %d, want 3; body %s", len(events), rr.Body.String())
-	}
-	for i, event := range events {
-		if event.Error != nil {
-			t.Fatalf("event %d error: %+v", i, event.Error)
-		}
-		if event.Result.ID != events[0].Result.ID || event.Result.ContextID != events[0].Result.ContextID {
-			t.Fatalf("event %d changed task identity: %+v vs %+v", i, event.Result, events[0].Result)
+	for i, e := range events {
+		if e.Error != nil {
+			t.Fatalf("event %d carried an error field: %+v", i, e.Error)
 		}
 	}
-	if events[0].Result.Status.State != stateWorking || textOf(events[0].Result.Artifacts[0].Parts) != "po" {
-		t.Fatalf("first event = %+v, want working po", events[0].Result)
+	if events[0].kind() != "task" {
+		t.Fatalf("first event kind = %q, want task", events[0].kind())
 	}
-	final := events[len(events)-1].Result
-	if final.Status.State != stateCompleted || textOf(final.Artifacts[0].Parts) != "pong" {
-		t.Fatalf("final event = %+v, want completed pong", final)
+	opening := events[0].task(t)
+	if opening.Status.State != stateWorking {
+		t.Fatalf("opening task state = %q, want working", opening.Status.State)
+	}
+	taskID := opening.ID
+
+	// The middle events are append artifact-updates carrying the chunk deltas.
+	var text strings.Builder
+	for _, e := range events[1:3] {
+		if e.kind() != "artifact-update" {
+			t.Fatalf("event kind = %q, want artifact-update", e.kind())
+		}
+		au := e.artifactUpdate(t)
+		if !au.Append {
+			t.Fatalf("artifact-update should be append: %+v", au)
+		}
+		if au.TaskID != taskID {
+			t.Fatalf("artifact-update taskId = %q, want %q", au.TaskID, taskID)
+		}
+		text.WriteString(textOf(au.Artifact.Parts))
+	}
+	if text.String() != "pong" {
+		t.Fatalf("accumulated artifact text = %q, want pong", text.String())
 	}
 
-	got := rpcTaskFromDispatcher(t, d, final.ID)
-	if got.ID != final.ID || got.Status.State != stateCompleted || textOf(got.Artifacts[0].Parts) != "pong" {
-		t.Fatalf("stored task = %+v, want final", got)
+	// The stream closes with a terminal status-update (final:true).
+	last := events[len(events)-1]
+	if last.kind() != "status-update" {
+		t.Fatalf("last event kind = %q, want status-update", last.kind())
+	}
+	su := last.status(t)
+	if !su.Final || su.Status.State != stateCompleted {
+		t.Fatalf("terminal event = %+v, want final completed", su)
+	}
+	if su.TaskID != taskID {
+		t.Fatalf("terminal taskId = %q, want %q", su.TaskID, taskID)
+	}
+
+	got := rpcTaskFromDispatcher(t, d, taskID)
+	if got.ID != taskID || got.Status.State != stateCompleted || textOf(got.Artifacts[0].Parts) != "pong" {
+		t.Fatalf("stored task = %+v, want final completed pong", got)
 	}
 }
 
@@ -432,37 +544,31 @@ func TestMessageStreamChunksPropagatesCancellationAndClosesStream(t *testing.T)
 		t.Fatal("stream was not closed")
 	}
 
-	var events []struct {
-		Result Task      `json:"result"`
-		Error  *rpcError `json:"error"`
+	events := collectSSE(t, rr.Body.String())
+	// Opening Task snapshot, then a terminal failed status-update.
+	if len(events) != 2 {
+		t.Fatalf("events = %d, want 2; body %s", len(events), rr.Body.String())
 	}
-	for _, line := range strings.Split(strings.TrimSpace(rr.Body.String()), "\n") {
-		line = strings.TrimSpace(line)
-		if line == "" {
-			continue
+	// A streaming failure must be a failed status-update, never `result` and
+	// `error` set together in one response.
+	for i, e := range events {
+		if e.Error != nil {
+			t.Fatalf("event %d carried an error field (result+error not allowed): %+v", i, e.Error)
 		}
-		line = strings.TrimPrefix(line, "data: ")
-		var event struct {
-			Result Task      `json:"result"`
-			Error  *rpcError `json:"error"`
-		}
-		if err := json.Unmarshal([]byte(line), &event); err != nil {
-			t.Fatalf("decode event %q: %v", line, err)
-		}
-		events = append(events, event)
 	}
-	if len(events) != 1 {
-		t.Fatalf("events = %d, want 1; body %s", len(events), rr.Body.String())
+	if events[0].kind() != "task" || events[0].task(t).Status.State != stateWorking {
+		t.Fatalf("first event = %s, want working task", string(events[0].Result))
 	}
-	event := events[0]
-	if event.Error == nil || event.Error.Code != errInternal || event.Error.Message != context.Canceled.Error() {
-		t.Fatalf("error = %+v, want context cancellation", event.Error)
+	last := events[1]
+	if last.kind() != "status-update" {
+		t.Fatalf("last event kind = %q, want status-update", last.kind())
 	}
-	if event.Result.Status.State != stateFailed || textOf(event.Result.Artifacts[0].Parts) != "error: context canceled" {
-		t.Fatalf("failed task = %+v, want context cancellation artifact", event.Result)
+	su := last.status(t)
+	if !su.Final || su.Status.State != stateFailed {
+		t.Fatalf("terminal event = %+v, want final failed", su)
 	}
 
-	got := rpcTaskFromDispatcher(t, d, event.Result.ID)
+	got := rpcTaskFromDispatcher(t, d, su.TaskID)
 	if got.Status.State != stateFailed || textOf(got.Artifacts[0].Parts) != "error: context canceled" {
 		t.Fatalf("stored task = %+v, want failed cancellation", got)
 	}
@@ -493,33 +599,24 @@ func TestMessageStreamChunksFallsBackWhenUnsupported(t *testing.T) {
 	if ct := rr.Result().Header.Get("Content-Type"); !strings.HasPrefix(ct, "text/event-stream") {
 		t.Fatalf("content-type = %q, want text/event-stream", ct)
 	}
-	var events []struct {
-		Result Task      `json:"result"`
-		Error  *rpcError `json:"error"`
+	events := collectSSE(t, rr.Body.String())
+	// The non-streaming fallback emits a completed Task snapshot then a terminal
+	// status-update.
+	if len(events) != 2 {
+		t.Fatalf("events = %d, want 2; body %s", len(events), rr.Body.String())
 	}
-	for _, line := range strings.Split(strings.TrimSpace(rr.Body.String()), "\n") {
-		line = strings.TrimSpace(line)
-		if line == "" {
-			continue
+	for i, e := range events {
+		if e.Error != nil {
+			t.Fatalf("fallback event %d error: %+v", i, e.Error)
 		}
-		line = strings.TrimPrefix(line, "data: ")
-		var event struct {
-			Result Task      `json:"result"`
-			Error  *rpcError `json:"error"`
-		}
-		if err := json.Unmarshal([]byte(line), &event); err != nil {
-			t.Fatalf("decode event %q: %v", line, err)
-		}
-		events = append(events, event)
 	}
-	if len(events) != 1 {
-		t.Fatalf("events = %d, want 1; body %s", len(events), rr.Body.String())
+	task := events[0].task(t)
+	if task.Status.State != stateCompleted || textOf(task.Artifacts[0].Parts) != "pong" {
+		t.Fatalf("fallback task = %+v, want completed pong", task)
 	}
-	if events[0].Error != nil {
-		t.Fatalf("fallback event error: %+v", events[0].Error)
-	}
-	if events[0].Result.Status.State != stateCompleted || textOf(events[0].Result.Artifacts[0].Parts) != "pong" {
-		t.Fatalf("fallback task = %+v, want completed pong", events[0].Result)
+	su := events[1].status(t)
+	if !su.Final || su.Status.State != stateCompleted {
+		t.Fatalf("terminal event = %+v, want final completed", su)
 	}
 }
 
@@ -535,30 +632,34 @@ func TestMessageStreamFallbackDoesNotCompleteWithEmptyText(t *testing.T) {
 		return nil, fmt.Errorf("%w: test provider", ai.ErrStreamingUnsupported)
 	})
 
-	var event struct {
-		Result Task      `json:"result"`
-		Error  *rpcError `json:"error"`
-	}
-	for _, line := range strings.Split(strings.TrimSpace(rr.Body.String()), "\n") {
-		line = strings.TrimSpace(strings.TrimPrefix(strings.TrimSpace(line), "data: "))
-		if line == "" {
-			continue
+	events := collectSSE(t, rr.Body.String())
+	var task Task
+	var foundTask bool
+	for _, e := range events {
+		if e.Error != nil {
+			t.Fatalf("fallback event error: %+v", e.Error)
 		}
-		if err := json.Unmarshal([]byte(line), &event); err != nil {
-			t.Fatalf("decode event %q: %v", line, err)
+		if e.kind() == "task" {
+			task = e.task(t)
+			foundTask = true
 		}
 	}
-	if event.Error != nil {
-		t.Fatalf("fallback event error: %+v", event.Error)
+	if !foundTask {
+		t.Fatalf("no task event in stream; body %s", rr.Body.String())
 	}
-	if event.Result.Status.State != stateFailed {
-		t.Fatalf("fallback state = %q, want failed", event.Result.Status.State)
+	if task.Status.State != stateFailed {
+		t.Fatalf("fallback state = %q, want failed", task.Status.State)
 	}
-	if got := textOf(event.Result.Artifacts[0].Parts); got == "" {
-		t.Fatalf("fallback artifact text is empty: %+v", event.Result.Artifacts)
+	if got := textOf(task.Artifacts[0].Parts); got == "" {
+		t.Fatalf("fallback artifact text is empty: %+v", task.Artifacts)
 	}
-	if got := textOf(event.Result.History[len(event.Result.History)-1].Parts); got == "" {
-		t.Fatalf("fallback history text is empty: %+v", event.Result.History)
+	if got := textOf(task.History[len(task.History)-1].Parts); got == "" {
+		t.Fatalf("fallback history text is empty: %+v", task.History)
+	}
+	// The stream still ends with a terminal marker.
+	last := events[len(events)-1]
+	if last.kind() != "status-update" || !last.status(t).Final {
+		t.Fatalf("stream must end with a final status-update; got %s", string(last.Result))
 	}
 }
 

gateway/a2a/ap2_test.go

@@ -1,7 +1,10 @@
 package a2a
 
 import (
+	"context"
 	"crypto/ed25519"
+	"encoding/json"
+	"fmt"
 	"strings"
 	"testing"
 	"time"
@@ -50,6 +53,83 @@ func TestAP2PaymentMandateX402RailReference(t *testing.T) {
 	}
 }
 
+// TestAP2GatewayVerifiesInboundPaymentMandate drives a real A2A message/send
+// carrying a signed x402 payment mandate through the gateway and asserts the
+// mandate is verified (and the x402 rail carried) into the task a paid path
+// consults — and that a tampered mandate is surfaced as unverified.
+func TestAP2GatewayVerifiesInboundPaymentMandate(t *testing.T) {
+	pub, priv := testAP2Key(t)
+	d := newDispatcher()
+	d.ap2Verify = func(s AP2SignedMandate, task Task) AP2Verification {
+		return VerifyAP2ForTask(s, pub, task, nil)
+	}
+	invoke := func(context.Context, string) (string, error) { return "fetched", nil }
+
+	send := func(t *testing.T, mandate AP2SignedMandate) Task {
+		t.Helper()
+		msg := AP2AttachMandate(
+			Message{Role: "user", Kind: "message", MessageID: "m1", Parts: []Part{{Kind: "text", Text: "pay and fetch"}}},
+			mandate,
+		)
+		params, err := json.Marshal(sendParams{Message: msg})
+		if err != nil {
+			t.Fatal(err)
+		}
+		body := fmt.Sprintf(`{"jsonrpc":"2.0","id":1,"method":"message/send","params":%s}`, params)
+		return rpcTaskFromBody(t, d, body, invoke)
+	}
+
+	rail := X402AP2Rail("payreq_777")
+	good, err := SignAP2Mandate(AP2Mandate{ID: "pay-1", Kind: AP2PaymentMandate, Rail: &rail, IssuedAt: time.Unix(1, 0).UTC()}, "k", priv)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	task := send(t, good)
+	if len(task.AP2Verifications) != 1 || !task.AP2Verifications[0].Verified {
+		t.Fatalf("inbound payment mandate not verified: %+v", task.AP2Verifications)
+	}
+	if task.AP2Verifications[0].Kind != string(AP2PaymentMandate) {
+		t.Errorf("verification kind = %q, want payment", task.AP2Verifications[0].Kind)
+	}
+	if len(task.AP2Mandates) != 1 || task.AP2Mandates[0].Mandate.Rail == nil ||
+		task.AP2Mandates[0].Mandate.Rail.Type != "x402" || task.AP2Mandates[0].Mandate.Rail.Reference != "payreq_777" {
+		t.Fatalf("x402 settlement rail not carried onto task: %+v", task.AP2Mandates)
+	}
+
+	tampered := good
+	tampered.Mandate.Amount = "999.00"
+	bad := send(t, tampered)
+	if len(bad.AP2Verifications) != 1 || bad.AP2Verifications[0].Verified {
+		t.Fatalf("tampered mandate should be unverified: %+v", bad.AP2Verifications)
+	}
+	if !strings.Contains(bad.AP2Verifications[0].Error, "signature") {
+		t.Errorf("tampered verification error = %q, want signature failure", bad.AP2Verifications[0].Error)
+	}
+}
+
+// TestAP2CarriedUnverifiedWithoutKey confirms the default (no configured key)
+// is unchanged: mandates are carried but not verified.
+func TestAP2CarriedUnverifiedWithoutKey(t *testing.T) {
+	_, priv := testAP2Key(t)
+	d := newDispatcher() // no ap2Verify configured
+	rail := X402AP2Rail("payreq_1")
+	signed, err := SignAP2Mandate(AP2Mandate{ID: "pay-1", Kind: AP2PaymentMandate, Rail: &rail, IssuedAt: time.Unix(1, 0).UTC()}, "k", priv)
+	if err != nil {
+		t.Fatal(err)
+	}
+	msg := AP2AttachMandate(Message{Role: "user", Kind: "message", MessageID: "m1", Parts: []Part{{Kind: "text", Text: "x"}}}, signed)
+	params, _ := json.Marshal(sendParams{Message: msg})
+	body := fmt.Sprintf(`{"jsonrpc":"2.0","id":1,"method":"message/send","params":%s}`, params)
+	task := rpcTaskFromBody(t, d, body, func(context.Context, string) (string, error) { return "ok", nil })
+	if len(task.AP2Mandates) != 1 {
+		t.Fatalf("mandate should still be carried: %+v", task.AP2Mandates)
+	}
+	if len(task.AP2Verifications) != 0 {
+		t.Errorf("no verifications without a configured key, got %+v", task.AP2Verifications)
+	}
+}
+
 func TestAP2TamperCasesFailDistinctly(t *testing.T) {
 	pub, priv := testAP2Key(t)
 	rail := X402AP2Rail("payreq_123")

gateway/a2a/client_test.go

@@ -6,6 +6,7 @@ import (
 	"errors"
 	"net/http"
 	"net/http/httptest"
+	"net/url"
 	"strings"
 	"testing"
 	"time"
@@ -56,9 +57,11 @@ func TestClientSendAndCard(t *testing.T) {
 
 func TestClientContinuesTaskAndConfiguresPush(t *testing.T) {
 	card := Card("solo", "http://localhost:4000", "", []string{"task"})
+	// The push receiver below is a loopback test server; authorize it as a
+	// deployment would authorize its trusted push receiver.
 	h := NewAgentHandler(card, func(_ context.Context, text string) (string, error) {
 		return "echo:" + text, nil
-	})
+	}, WithPushURLPolicy(func(*url.URL) error { return nil }))
 	ts := httptest.NewServer(h)
 	defer ts.Close()
 

gateway/a2a/pushsecurity.go

@@ -0,0 +1,131 @@
+package a2a
+
+import (
+	"fmt"
+	"net"
+	"net/http"
+	"net/url"
+	"syscall"
+	"time"
+)
+
+// Push-notification callbacks are the one place the A2A gateway makes an
+// outbound HTTP request to an address chosen by a (possibly untrusted) caller:
+// tasks/pushNotificationConfig/set records a URL and deliverPush POSTs task
+// state to it. Without a guard that is a server-side request forgery vector —
+// a caller can aim the gateway at loopback, link-local (cloud metadata), or
+// private hosts it would otherwise never reach.
+//
+// The default policy allows only http/https callbacks whose host does not
+// resolve to a loopback, private, link-local, or unspecified address, and the
+// guarded HTTP client re-checks the *resolved* IP at dial time so a hostname
+// that passes validation cannot be rebound to an internal address before the
+// connection is made. Operators who need to reach a trusted in-cluster
+// receiver set Options.AllowPushURL to take over the policy.
+
+// pushLookupIP resolves a host to IPs; overridable in tests.
+var pushLookupIP = net.LookupIP
+
+// defaultPushURLPolicy is the SSRF-safe policy applied when no AllowPushURL is
+// configured. It rejects non-http(s) schemes and hosts that resolve to a
+// loopback, private, link-local, multicast, or unspecified address.
+func defaultPushURLPolicy(u *url.URL) error {
+	switch u.Scheme {
+	case "http", "https":
+	default:
+		return fmt.Errorf("push callback scheme %q not allowed (want http or https)", u.Scheme)
+	}
+	host := u.Hostname()
+	if host == "" {
+		return fmt.Errorf("push callback url has no host")
+	}
+	ips, err := resolvePushHost(host)
+	if err != nil {
+		return fmt.Errorf("push callback host %q: %w", host, err)
+	}
+	if len(ips) == 0 {
+		return fmt.Errorf("push callback host %q did not resolve", host)
+	}
+	for _, ip := range ips {
+		if blockedPushIP(ip) {
+			return fmt.Errorf("push callback host %q resolves to a blocked address %s", host, ip)
+		}
+	}
+	return nil
+}
+
+func resolvePushHost(host string) ([]net.IP, error) {
+	if ip := net.ParseIP(host); ip != nil {
+		return []net.IP{ip}, nil
+	}
+	return pushLookupIP(host)
+}
+
+// blockedPushIP reports whether ip is one an outbound push callback must not
+// reach: loopback, private (RFC1918 / ULA), link-local (incl. 169.254.169.254
+// cloud metadata), multicast, or the unspecified address.
+func blockedPushIP(ip net.IP) bool {
+	return ip == nil ||
+		ip.IsLoopback() ||
+		ip.IsPrivate() ||
+		ip.IsLinkLocalUnicast() ||
+		ip.IsLinkLocalMulticast() ||
+		ip.IsInterfaceLocalMulticast() ||
+		ip.IsMulticast() ||
+		ip.IsUnspecified()
+}
+
+// pushDialControl runs after DNS resolution, immediately before connect, on the
+// resolved address — so it blocks a host that passed URL validation but was
+// rebound to an internal IP (DNS rebinding).
+func pushDialControl(_, address string, _ syscall.RawConn) error {
+	host, _, err := net.SplitHostPort(address)
+	if err != nil {
+		return err
+	}
+	ip := net.ParseIP(host)
+	if ip == nil {
+		return fmt.Errorf("push callback: cannot parse dial address %q", address)
+	}
+	if blockedPushIP(ip) {
+		return fmt.Errorf("push callback: refusing to connect to blocked address %s", ip)
+	}
+	return nil
+}
+
+// pushGuardClient is the HTTP client used for default-policy push delivery. Its
+// dialer refuses connections to blocked addresses at connect time.
+var pushGuardClient = &http.Client{
+	Timeout: 10 * time.Second,
+	Transport: &http.Transport{
+		Proxy: http.ProxyFromEnvironment,
+		DialContext: (&net.Dialer{
+			Timeout: 5 * time.Second,
+			Control: pushDialControl,
+		}).DialContext,
+	},
+}
+
+// checkPushURL validates a callback URL against the dispatcher's effective
+// policy (Options.AllowPushURL, or the default SSRF-safe policy).
+func (d *dispatcher) checkPushURL(raw string) error {
+	u, err := url.Parse(raw)
+	if err != nil {
+		return fmt.Errorf("invalid push callback url: %w", err)
+	}
+	policy := d.allowPushURL
+	if policy == nil {
+		policy = defaultPushURLPolicy
+	}
+	return policy(u)
+}
+
+// pushClient is the HTTP client deliverPush uses: the guarded client under the
+// default policy, or the default client when an operator has taken over the
+// policy via Options.AllowPushURL (they own the trust decision then).
+func (d *dispatcher) pushClient() *http.Client {
+	if d.guardPushDial {
+		return pushGuardClient
+	}
+	return http.DefaultClient
+}

gateway/a2a/pushsecurity_test.go

@@ -0,0 +1,147 @@
+package a2a
+
+import (
+	"encoding/json"
+	"net"
+	"net/http"
+	"net/http/httptest"
+	"net/url"
+	"testing"
+)
+
+func TestDefaultPushURLPolicy(t *testing.T) {
+	// Resolve test hostnames deterministically without real DNS.
+	orig := pushLookupIP
+	pushLookupIP = func(host string) ([]net.IP, error) {
+		switch host {
+		case "internal.example":
+			return []net.IP{net.ParseIP("10.1.2.3")}, nil
+		case "public.example":
+			return []net.IP{net.ParseIP("93.184.216.34")}, nil
+		case "rebind.example":
+			// A host that resolves to both a public and an internal IP must be
+			// rejected — any blocked address is disqualifying.
+			return []net.IP{net.ParseIP("93.184.216.34"), net.ParseIP("127.0.0.1")}, nil
+		}
+		return nil, &net.DNSError{Err: "no such host", Name: host, IsNotFound: true}
+	}
+	defer func() { pushLookupIP = orig }()
+
+	blocked := []string{
+		"http://127.0.0.1/hook",              // loopback
+		"http://169.254.169.254/latest/meta", // cloud metadata (link-local)
+		"http://10.0.0.5/hook",               // RFC1918
+		"http://[::1]/hook",                  // IPv6 loopback
+		"http://[fd00::1]/hook",              // IPv6 ULA (private)
+		"http://0.0.0.0/hook",                // unspecified
+		"http://internal.example/hook",       // hostname → private
+		"http://rebind.example/hook",         // one internal IP among many
+		"ftp://public.example/hook",          // non-http(s) scheme
+		"file:///etc/passwd",                 // scheme
+		"http:///nohost",                     // no host
+	}
+	for _, raw := range blocked {
+		u, err := url.Parse(raw)
+		if err != nil {
+			t.Fatalf("parse %q: %v", raw, err)
+		}
+		if err := defaultPushURLPolicy(u); err == nil {
+			t.Errorf("defaultPushURLPolicy(%q) = nil, want blocked", raw)
+		}
+	}
+
+	allowed := []string{
+		"http://93.184.216.34/hook",   // public literal IP
+		"https://public.example/hook", // hostname → public
+	}
+	for _, raw := range allowed {
+		u, _ := url.Parse(raw)
+		if err := defaultPushURLPolicy(u); err != nil {
+			t.Errorf("defaultPushURLPolicy(%q) = %v, want allowed", raw, err)
+		}
+	}
+}
+
+func TestPushDialControlBlocksPrivate(t *testing.T) {
+	blocked := []string{"127.0.0.1:80", "169.254.169.254:80", "10.0.0.1:443", "[::1]:80", "0.0.0.0:80"}
+	for _, addr := range blocked {
+		if err := pushDialControl("tcp", addr, nil); err == nil {
+			t.Errorf("pushDialControl(%q) = nil, want blocked", addr)
+		}
+	}
+	if err := pushDialControl("tcp", "8.8.8.8:443", nil); err != nil {
+		t.Errorf("pushDialControl(public) = %v, want allowed", err)
+	}
+}
+
+// TestSetPushConfigRejectsSSRFURL: an untrusted caller cannot register a
+// callback pointing at an internal address — it is refused and nothing stored.
+func TestSetPushConfigRejectsSSRFURL(t *testing.T) {
+	d := newDispatcher()
+	d.store(&Task{ID: "t1", ContextID: "c1", Status: TaskStatus{State: stateCompleted}})
+
+	params, _ := json.Marshal(map[string]any{
+		"id":                     "t1",
+		"pushNotificationConfig": map[string]any{"url": "http://169.254.169.254/latest/meta-data"},
+	})
+	rr := httptest.NewRecorder()
+	d.setPushConfig(rr, rpcRequest{JSONRPC: "2.0", ID: json.RawMessage("1"), Params: params})
+
+	var resp rpcResponse
+	if err := json.Unmarshal(rr.Body.Bytes(), &resp); err != nil {
+		t.Fatalf("decode: %v", err)
+	}
+	if resp.Error == nil || resp.Error.Code != errInvalidParams {
+		t.Fatalf("response = %+v, want invalid-params rejection", resp)
+	}
+	d.mu.Lock()
+	_, stored := d.pushConfigs["t1"]
+	d.mu.Unlock()
+	if stored {
+		t.Error("SSRF callback url must not be stored")
+	}
+}
+
+// TestDeliverPushBlocksInternalByDefault: even if a config for an internal URL
+// slips into the map, deliverPush must not POST to it under the default policy.
+func TestDeliverPushBlocksInternalByDefault(t *testing.T) {
+	var hit bool
+	srv := httptest.NewServer(http.HandlerFunc(func(http.ResponseWriter, *http.Request) { hit = true }))
+	defer srv.Close() // srv.URL is http://127.0.0.1:PORT — loopback, must be blocked
+
+	d := newDispatcher()
+	task := &Task{ID: "t1", Status: TaskStatus{State: stateCompleted}}
+	d.pushConfigs["t1"] = PushNotificationConfig{URL: srv.URL}
+
+	d.deliverPush("t1", task)
+	if hit {
+		t.Error("deliverPush reached a loopback callback under the default policy")
+	}
+}
+
+// TestAllowPushURLOverrideDelivers: an operator policy can authorize a trusted
+// (here loopback) receiver, and delivery then goes through.
+func TestAllowPushURLOverrideDelivers(t *testing.T) {
+	done := make(chan struct{}, 1)
+	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		if r.Header.Get("Content-Type") == "application/json" {
+			done <- struct{}{}
+		}
+	}))
+	defer srv.Close()
+
+	g := New(Options{AllowPushURL: func(*url.URL) error { return nil }})
+	d := g.disp
+	if d.guardPushDial {
+		t.Fatal("custom AllowPushURL should disable the dial guard")
+	}
+	task := &Task{ID: "t1", Status: TaskStatus{State: stateCompleted}}
+	d.pushConfigs["t1"] = PushNotificationConfig{URL: srv.URL}
+
+	d.deliverPush("t1", task)
+	select {
+	case <-done:
+	default:
+		t.Error("operator-authorized callback was not delivered")
+	}
+}

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

@@ -166,7 +166,9 @@ func main() {
 		fmt.Fprintln(os.Stderr, err)
 		os.Exit(1)
 	}
-	if summary.WorkingEvents == 0 || summary.State != "completed" || !strings.Contains(summary.FinalText, "a2a-stream-ok") {
+	// Spec-shaped stream: at least one artifact-update carrying the reassembled
+	// answer, terminating in a completed status-update with final:true.
+	if summary.ArtifactEvents == 0 || summary.State != "completed" || !summary.Final || !strings.Contains(summary.FinalText, "a2a-stream-ok") {
 		fmt.Fprintf(os.Stderr, "unexpected stream summary: %+v\npayload:\n%s", summary, summary.Payload)
 		os.Exit(1)
 	}
@@ -174,14 +176,16 @@ func main() {
 		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")
+	fmt.Println("\n\033[32m✓ A2A message/stream emitted spec-shaped artifact/status updates and preserved tool/run metadata\033[0m")
 }
 
 type streamSummary struct {
-	Payload       string
-	State         string
-	FinalText     string
-	WorkingEvents int
+	Payload        string
+	State          string
+	FinalText      string
+	Final          bool
+	ArtifactEvents int
+	WorkingEvents  int
 }
 
 func readSSESummary(r io.Reader) (streamSummary, error) {
@@ -197,14 +201,23 @@ func readSSESummary(r io.Reader) (streamSummary, error) {
 		}
 		var envelope struct {
 			Result struct {
+				Kind   string `json:"kind"`
+				Final  bool   `json:"final"`
 				Status struct {
 					State string `json:"state"`
 				} `json:"status"`
+				// Task snapshots carry artifacts (plural)...
 				Artifacts []struct {
 					Parts []struct {
 						Text string `json:"text"`
 					} `json:"parts"`
 				} `json:"artifacts"`
+				// ...artifact-update events carry a single artifact.
+				Artifact struct {
+					Parts []struct {
+						Text string `json:"text"`
+					} `json:"parts"`
+				} `json:"artifact"`
 			} `json:"result"`
 			Error any `json:"error"`
 		}
@@ -216,15 +229,34 @@ func readSSESummary(r io.Reader) (streamSummary, error) {
 		}
 		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
+		switch envelope.Result.Kind {
+		case "artifact-update":
+			// Incremental deltas: reassemble the streamed answer.
+			summary.ArtifactEvents++
+			for _, part := range envelope.Result.Artifact.Parts {
+				summary.FinalText += part.Text
+			}
+		case "status-update":
+			if envelope.Result.Status.State != "" {
+				summary.State = envelope.Result.Status.State
+			}
+			if envelope.Result.Final {
+				summary.Final = true
+			}
+		default: // "task" snapshot
+			if envelope.Result.Status.State == "working" {
+				summary.WorkingEvents++
+			}
+			if envelope.Result.Status.State != "" {
+				summary.State = envelope.Result.Status.State
+			}
+			// The non-streaming path carries the full text in the snapshot.
+			for _, artifact := range envelope.Result.Artifacts {
+				for _, part := range artifact.Parts {
+					if part.Text != "" {
+						summary.FinalText = part.Text
+					}
+				}
 			}
 		}
 		return nil

internal/network/network.go

@@ -0,0 +1,51 @@
+// Package network is a process-local registry of server dispatchers — the
+// neutral seam an in-process client fast-path uses to reach a server running in
+// the same process without going over the network transport.
+//
+// It lives in internal/ and speaks only in transport.Message so neither the
+// client nor the server package has to import the other: a running server
+// registers a Handler under its service name; an opted-in client looks one up
+// and dispatches directly, skipping dial, codec-over-socket, and the transport
+// pump. Nothing here runs unless a server registers and a client opts in.
+package network
+
+import (
+	"context"
+	"sync"
+
+	"go-micro.dev/v6/transport"
+)
+
+// Handler dispatches one request against a process-local server's handler
+// table and returns the reply. req and the returned message carry the same
+// codec-encoded body + headers the transport would have carried.
+type Handler func(ctx context.Context, req *transport.Message) (*transport.Message, error)
+
+var (
+	mu  sync.RWMutex
+	reg = map[string]Handler{}
+)
+
+// Register makes service reachable in-process via h. A server calls this when
+// it starts; calling again replaces the handler.
+func Register(service string, h Handler) {
+	mu.Lock()
+	reg[service] = h
+	mu.Unlock()
+}
+
+// Deregister removes service's in-process handler. A server calls this when it
+// stops, so a later in-process call falls back to the network path.
+func Deregister(service string) {
+	mu.Lock()
+	delete(reg, service)
+	mu.Unlock()
+}
+
+// Lookup returns the in-process handler for service, if one is registered.
+func Lookup(service string) (Handler, bool) {
+	mu.RLock()
+	h, ok := reg[service]
+	mu.RUnlock()
+	return h, ok
+}

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

@@ -150,29 +150,71 @@ See [Native gRPC Compatibility](grpc-compatibility.md) for a complete guide.
 
 ## vs Dapr
 
-### Dapr Approach
-- Multi-language via sidecar
-- Rich building blocks (state, pub/sub, bindings)
-- Cloud-native focused
-- Requires running sidecar process
+[Dapr](https://dapr.io/) is a distributed application runtime. Its building
+blocks cover service invocation, state, pub/sub, bindings, secrets,
+configuration, distributed locks, actors, jobs, and workflow, usually accessed
+through a sidecar from many languages. [Dapr Agents](https://docs.dapr.io/developing-ai/dapr-agents/)
+adds an agent framework on top of those runtime capabilities.
 
-### Go Micro Approach
-- Go library, no sidecar
-- Direct service-to-service calls
-- Simpler deployment
-- Lower latency (no extra hop)
+Go Micro overlaps with Dapr on distributed-systems primitives, but the product
+shape is different: Go Micro is a Go framework where services, agents, tools,
+and flows are built from the same runtime. A service endpoint can become an
+AI-callable tool, and an agent is itself a registered service with memory,
+guardrails, planning, delegation, MCP, and A2A around it.
 
-### When to Choose Dapr
-- You have polyglot services (Node, Python, Java, etc)
-- You want portable abstractions across clouds
-- You're fully on Kubernetes
-- You need state management abstractions
+### Decision table
 
-### When to Choose Go Micro
-- You're building Go services
-- You want lower latency
-- You prefer libraries over sidecars
-- You want simpler deployment (no sidecar management)
+| Need | Prefer Go Micro | Prefer Dapr | Use both |
+|---|---|---|---|
+| **Primary language** | Your core runtime is Go and you want library-native APIs | You run a polyglot estate and want one sidecar API across languages | Go services use Go Micro while non-Go services expose Dapr APIs |
+| **Agent model** | Agents should be ordinary services: registered, discoverable, callable by RPC, MCP, and A2A | Agents are primarily Python applications using Dapr Agents | Dapr-hosted agents call Go Micro MCP tools, or Go Micro agents call Dapr-backed services |
+| **Tools** | Existing service endpoints should become tools with minimal extra code | Tools are modeled through Dapr components, bindings, or agent framework code | Use Dapr components behind Go Micro services that expose a stable tool surface |
+| **Workflows** | Deterministic steps should live beside Go services and agents in the same codebase | You want Dapr Workflow's sidecar-backed orchestration model across languages | Let Dapr own cross-language workflows and let Go Micro own Go-native agent/tool execution |
+| **State and pub/sub** | You want Go interfaces and pluggable packages directly in-process | You want component YAML and sidecar portability across backing services | Put portable infrastructure behind Dapr and domain/tool logic in Go Micro |
+| **Deployment** | You want a simple Go binary/runtime first, with Kubernetes support as an explicit deployment target | You are already standardized on Dapr sidecars in Kubernetes | Run Go Micro services in clusters that already have Dapr for shared infrastructure |
+| **Interop** | MCP and A2A are first-class requirements for exposing services and agents | Dapr's app APIs and agent framework are the integration boundary | Bridge through MCP/A2A at the agent edge and Dapr APIs at the infrastructure edge |
+
+### When to choose Dapr
+
+- You need a **polyglot** runtime contract for Node, Python, Java, .NET, Go, and
+  other services.
+- Your platform team already operates sidecars and component configuration across
+  Kubernetes clusters.
+- You want Dapr's standard building blocks for state, pub/sub, bindings, secrets,
+  actors, jobs, and workflow more than you want a Go-native service framework.
+- You are adopting Dapr Agents and want to stay in its Python-first agent stack.
+
+### When to choose Go Micro
+
+- You are building mostly in Go and want the agent harness to be the same runtime
+  as your services.
+- You want service methods and their comments/examples to become AI-callable tools
+  without maintaining a separate tool layer.
+- You want agents to be deployed, discovered, called, load-balanced, and inspected
+  like ordinary services.
+- You need MCP and A2A at the agent/service boundary, not only an internal
+  application API.
+- You prefer library-native composition and direct Go interfaces over sidecar
+  component wiring.
+
+### Where Go Micro still needs to prove itself
+
+Dapr has a mature platform narrative and broad deployment footprint. Go Micro's
+agent-harness story is sharper for Go teams, but production adoption depends on
+keeping the no-secret getting-started path green, documenting durability
+semantics clearly, proving MCP/A2A conformance with external clients, and making
+Kubernetes deployment first-class.
+
+### Practical migration path
+
+1. Start with one Go Micro service that wraps a real domain capability.
+2. Add doc comments and examples so the endpoint is useful as an agent tool.
+3. Expose it through MCP for external agents or through A2A if the capability is
+   itself an agent.
+4. If your platform already uses Dapr, keep Dapr components behind the service
+   boundary and let Go Micro present the agent/tool contract.
+5. Move deterministic multi-step work into flows only after the service/tool
+   boundary is stable.
 
 ## vs Agent Frameworks (Google ADK)
 

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

internal/website/docs/index.md

@@ -64,7 +64,7 @@ Otherwise continue to read the docs for more information about the framework.
 
 ## Advanced
 
-- [Framework Comparison](guides/comparison.html)
+- [Framework Comparison](guides/comparison.html) - Including Go Micro vs Dapr for agents, services, and workflows
 - [Architecture Decisions](architecture/)
 - [Real-World Examples](examples/realworld/)
 - [Migration Guides](guides/migration/)

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"
 )
 
@@ -130,6 +131,13 @@ 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) }
 

options.go

@@ -8,6 +8,7 @@ var Broker = service.Broker
 var Cache = service.Cache
 var Cmd = service.Cmd
 var Client = service.Client
+var Local = service.Local
 var Context = service.Context
 var Handle = service.Handle
 var HandleSignal = service.HandleSignal

server/local.go

@@ -0,0 +1,119 @@
+package server
+
+import (
+	"context"
+	"io"
+
+	"go-micro.dev/v6/internal/network"
+	"go-micro.dev/v6/transport"
+	"go-micro.dev/v6/transport/headers"
+)
+
+// local.go gives a same-process caller a way to reach this server's handlers
+// without the network transport. A running server registers a dispatcher in
+// internal/network keyed by its name; an opted-in client looks it up and
+// calls localDispatch, which serves the request synchronously through the same
+// router (so handler wrappers, codecs, and error mapping are identical) over an
+// in-memory socket — skipping dial, the transport pump, and the codec-over-pipe
+// double serialization. Unary only; streaming and pub/sub keep the normal path.
+
+// localSocket is a transport.Socket that carries exactly one request in and
+// captures exactly one reply — no network, no pipe, no gob. Recv delivers the
+// request message once (the RPC codec reads it on the first ReadHeader), then
+// reports EOF; Send captures the encoded reply.
+type localSocket struct {
+	req   *transport.Message
+	recvd bool
+	reply *transport.Message
+}
+
+func (s *localSocket) Recv(m *transport.Message) error {
+	if s.recvd || s.req == nil {
+		return io.EOF
+	}
+	s.recvd = true
+	m.Header = s.req.Header
+	m.Body = s.req.Body
+	return nil
+}
+
+func (s *localSocket) Send(m *transport.Message) error {
+	cp := &transport.Message{Header: make(map[string]string, len(m.Header))}
+	for k, v := range m.Header {
+		cp.Header[k] = v
+	}
+	if len(m.Body) > 0 {
+		cp.Body = append([]byte(nil), m.Body...)
+	}
+	s.reply = cp
+	return nil
+}
+
+func (s *localSocket) Close() error   { return nil }
+func (s *localSocket) Local() string  { return "local" }
+func (s *localSocket) Remote() string { return "local" }
+
+// localDispatch serves req against this server's router in-process and returns
+// the reply. It mirrors the request/response construction ServeConn does for a
+// networked request, so the served path is identical apart from the transport.
+func (s *rpcServer) localDispatch(ctx context.Context, req *transport.Message) (*transport.Message, error) {
+	contentType := req.Header["Content-Type"]
+	if contentType == "" {
+		contentType = DefaultContentType
+		req.Header["Content-Type"] = contentType
+	}
+
+	cf := setupProtocol(req)
+	if cf == nil {
+		var err error
+		if cf, err = s.newCodec(contentType); err != nil {
+			return nil, err
+		}
+	}
+
+	sock := &localSocket{req: req}
+	rcodec := newRPCCodec(req, sock, cf)
+
+	request := rpcRequest{
+		service:     getHeader(headers.Request, req.Header),
+		method:      getHeader(headers.Method, req.Header),
+		endpoint:    getHeader(headers.Endpoint, req.Header),
+		contentType: contentType,
+		codec:       rcodec,
+		header:      req.Header,
+		body:        req.Body,
+		socket:      sock,
+	}
+	response := rpcResponse{
+		header: make(map[string]string),
+		socket: sock,
+		codec:  rcodec,
+	}
+
+	if err := s.getRouter().ServeRequest(ctx, &request, &response); err != nil {
+		return nil, err
+	}
+	if sock.reply == nil {
+		// A handler that wrote no body still completed successfully.
+		return &transport.Message{Header: map[string]string{}}, nil
+	}
+	return sock.reply, nil
+}
+
+// registerLocal makes this server reachable in-process under its name; called
+// on Start. deregisterLocal removes it on Stop.
+func (s *rpcServer) registerLocal() {
+	name := s.Options().Name
+	if name == "" {
+		return
+	}
+	network.Register(name, s.localDispatch)
+}
+
+func (s *rpcServer) deregisterLocal() {
+	name := s.Options().Name
+	if name == "" {
+		return
+	}
+	network.Deregister(name)
+}

server/rpc_server.go

@@ -571,6 +571,9 @@ func (s *rpcServer) Start() error {
 	// Keep the service registered to registry
 	go s.registrar(listener, addr, config, exit)
 
+	// Make this server reachable in-process for the client fast-path.
+	s.registerLocal()
+
 	s.setStarted(true)
 
 	return nil
@@ -581,6 +584,8 @@ func (s *rpcServer) Stop() error {
 		return nil
 	}
 
+	s.deregisterLocal()
+
 	ch := make(chan error)
 	s.exit <- ch
 

service/options.go

@@ -111,6 +111,18 @@ func Client(c client.Client) Option {
 	}
 }
 
+// Local enables the in-process fast-path on the service's client: a
+// unary call to another service running in the same process — agent tool calls,
+// flow dispatch, gateway → service — skips the network transport and dispatches
+// straight to that server's handlers (see client.Local). Off by
+// default; it falls back to the network path for anything not co-located, so
+// it is a pure win for all-in-one binaries and a no-op for distributed ones.
+func Local() Option {
+	return func(o *Options) {
+		_ = o.Client.Init(client.Local())
+	}
+}
+
 // Context specifies a context for the service.
 // Can be used to signal shutdown of the service and for extra option values.
 func Context(ctx context.Context) Option {

service/options_test.go

@@ -0,0 +1,14 @@
+package service
+
+import "testing"
+
+func TestLocalOption(t *testing.T) {
+	// Off by default.
+	if newOptions().Client.Options().Local {
+		t.Fatal("Local should be off by default")
+	}
+	// Enabled by the option, on the service's own client.
+	if !newOptions(Local()).Client.Options().Local {
+		t.Fatal("Local() did not enable the client fast-path")
+	}
+}

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)