Refresh planner queue after retry controls

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

.github/loop/PRIORITIES.md

@@ -21,7 +21,7 @@ changes, architectural rewrites. Those go to the human.
 
 ## Work queue (ranked)
 
-1. **Add end-to-end streaming chat conformance** ([#4738](https://github.com/micro/go-micro/issues/4738)) — #4736 closed the pgx/v5 security-upkeep item, and there are no open codex PRs in flight. The next highest-value user-visible gap is streaming chat that works coherently across the agent/CLI/A2A path, starting with a no-secret harness and extending to keyed provider checks without another provider-specific malformed-output workaround.
+1. **Normalize provider error status in agent inspection** ([#4777](https://github.com/micro/go-micro/issues/4777)) — #4775 closed the retry/timeout controls gap, so the next highest-value user-facing gap is operability when a provider still fails: classify timeout, cancellation, rate-limit, auth/configuration, and transient provider errors into stable RunInfo/inspect/tracing details so developers know whether to retry, fix credentials, raise deadlines, or wait out an outage without changing provider public APIs.
 
 _Seeded by Claude Code from the roadmap + open issues; thereafter maintained by the
 architecture-review pass._

agent/agent.go

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

agent/builtin.go

@@ -131,6 +131,7 @@ func (a *agentImpl) toolHandler() ai.ToolHandler {
 	h = a.toolRetryWrap(h)
 	h = a.checkpointToolWrap(h)
 	h = a.approveWrap(h)
+	h = a.spendWrap(h)
 	h = a.loopWrap(h)
 	h = a.stepWrap(h)
 	h = a.planWrap(h)
@@ -375,6 +376,27 @@ func (a *agentImpl) approveWrap(next ai.ToolHandler) ai.ToolHandler {
 	}
 }
 
+// spendWrap reserves a per-run x402 spend budget before paid tool execution.
+func (a *agentImpl) spendWrap(next ai.ToolHandler) ai.ToolHandler {
+	return func(ctx context.Context, call ai.ToolCall) ai.ToolResult {
+		amount := a.opts.ToolSpend[call.Name]
+		if amount <= 0 || a.opts.MaxSpend <= 0 {
+			return next(ctx, call)
+		}
+		if a.spend+amount > a.opts.MaxSpend {
+			return refused(call.ID, ai.RefusedSpendBudget, fmt.Sprintf(
+				"x402 spend budget exceeded: paying %d for %s would exceed per-run budget (spent %d of %d)",
+				amount, call.Name, a.spend, a.opts.MaxSpend))
+		}
+		a.spend += amount
+		res := next(ctx, call)
+		if res.Refused != "" || toolErrorMessage(res) != "" {
+			a.spend -= amount
+		}
+		return res
+	}
+}
+
 // handlePlan persists the supplied plan to the agent's memory and
 // echoes it back so the model can see the stored state.
 func (a *agentImpl) handlePlan(call ai.ToolCall) ai.ToolResult {

agent/guardrails_test.go

@@ -91,6 +91,77 @@ func TestApproveToolDoesNotGatePlan(t *testing.T) {
 	}
 }
 
+func TestMaxSpendAllowsPaidToolWithinBudget(t *testing.T) {
+	calls := 0
+	a := newTestAgent(Name("paid-within-budget"),
+		MaxSpend(10),
+		ToolSpend("paid.lookup", 7),
+		WithTool("paid.lookup", "paid lookup", nil, func(context.Context, map[string]any) (string, error) {
+			calls++
+			return `{"ok":true}`, nil
+		}),
+	)
+
+	res := a.toolHandler()(context.Background(), ai.ToolCall{ID: "paid-1", Name: "paid.lookup", Input: map[string]any{}})
+	if calls != 1 {
+		t.Fatalf("paid tool was not executed")
+	}
+	if res.Refused != "" {
+		t.Fatalf("paid tool was refused: %+v", res)
+	}
+	if res.Content != `{"ok":true}` {
+		t.Fatalf("content = %q, want paid result", res.Content)
+	}
+}
+
+func TestMaxSpendRefusesPaidToolBeforePaymentWhenBudgetExceeded(t *testing.T) {
+	calls := 0
+	a := newTestAgent(Name("paid-over-budget"),
+		MaxSpend(5),
+		ToolSpend("paid.lookup", 7),
+		WithTool("paid.lookup", "paid lookup", nil, func(context.Context, map[string]any) (string, error) {
+			calls++
+			return `{"ok":true}`, nil
+		}),
+	)
+
+	res := a.toolHandler()(context.Background(), ai.ToolCall{ID: "paid-1", Name: "paid.lookup", Input: map[string]any{}})
+	if calls != 0 {
+		t.Fatalf("paid tool ran despite budget refusal")
+	}
+	if res.Refused != ai.RefusedSpendBudget {
+		t.Fatalf("Refused = %q, want %q (result %+v)", res.Refused, ai.RefusedSpendBudget, res)
+	}
+	if !strings.Contains(res.Content, "x402 spend budget exceeded") {
+		t.Fatalf("content = %q, want inspectable budget refusal", res.Content)
+	}
+}
+
+func TestMaxSpendRollsBackFailedPaidToolReservation(t *testing.T) {
+	calls := 0
+	a := newTestAgent(Name("paid-rollback"),
+		MaxSpend(10),
+		ToolSpend("paid.lookup", 7),
+		WithTool("paid.lookup", "paid lookup", nil, func(context.Context, map[string]any) (string, error) {
+			calls++
+			if calls == 1 {
+				return "", context.Canceled
+			}
+			return `{"ok":true}`, nil
+		}),
+	)
+
+	h := a.toolHandler()
+	first := h(context.Background(), ai.ToolCall{ID: "paid-1", Name: "paid.lookup", Input: map[string]any{}})
+	if first.Refused != "" || !strings.Contains(first.Content, "context canceled") {
+		t.Fatalf("first result = %+v, want tool error without guardrail refusal", first)
+	}
+	second := h(context.Background(), ai.ToolCall{ID: "paid-2", Name: "paid.lookup", Input: map[string]any{}})
+	if second.Refused != "" || second.Content != `{"ok":true}` {
+		t.Fatalf("second result = %+v, want reservation rollback to allow retry", second)
+	}
+}
+
 func TestNestedTextToolCallArgumentsAreRefused(t *testing.T) {
 	called := false
 	a := newTestAgent(Name("nested-tool-arg"),

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

@@ -59,6 +59,9 @@ type Options struct {
 	// ModelRetryBackoff is the base delay between transient provider failures
 	// (grows exponentially per attempt when retries are enabled).
 	ModelRetryBackoff time.Duration
+	// ModelRetryJitter adds up to this random delay to each provider retry
+	// backoff. Default 0 preserves deterministic timing unless explicitly set.
+	ModelRetryJitter time.Duration
 	// ToolTimeout bounds each tool execution (0 disables). The timeout is
 	// applied before custom tools, delegate, and service RPC calls so context
 	// deadlines propagate consistently through the agent loop.
@@ -98,6 +101,10 @@ type Options struct {
 	LoopLimit int
 	// Approve gates each action before it runs. Nil = allow all.
 	Approve ApproveFunc
+	// MaxSpend bounds paid x402 tool spend per Ask in the asset's smallest
+	// unit (0 = disabled). ToolSpend lists known paid tools and their prices.
+	MaxSpend  int64
+	ToolSpend map[string]int64
 
 	// A2AAddress, if set, makes Run serve this agent over the A2A protocol
 	// on that address directly (no separate gateway), e.g. ":4000".
@@ -224,6 +231,25 @@ func ApproveTool(fn ApproveFunc) Option {
 	return func(o *Options) { o.Approve = fn }
 }
 
+// MaxSpend bounds paid x402 tool spend per Ask, in the asset's smallest unit
+// (0 = disabled). A paid tool that would exceed the cap is refused before the
+// tool handler runs or any payment can be made.
+func MaxSpend(amount int64) Option {
+	return func(o *Options) { o.MaxSpend = amount }
+}
+
+// ToolSpend records the x402 price for a tool, in the asset's smallest unit,
+// so MaxSpend can reserve budget before execution. Non-positive amounts are
+// treated as free.
+func ToolSpend(tool string, amount int64) Option {
+	return func(o *Options) {
+		if o.ToolSpend == nil {
+			o.ToolSpend = map[string]int64{}
+		}
+		o.ToolSpend[tool] = amount
+	}
+}
+
 // LoopLimit sets how many times the agent may repeat the same tool call
 // (same name and arguments) in one Ask before it is refused as a
 // no-progress loop. 0 disables loop detection.
@@ -252,6 +278,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/proto/agent.pb.micro.go

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

agent/proto/agent.proto

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

agent/stream_test.go

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

ai/model.go

@@ -109,6 +109,9 @@ const (
 	RefusedMaxSteps = "max_steps"
 	RefusedLoop     = "loop"
 	RefusedApproval = "approval"
+	// RefusedSpendBudget means an agent refused a paid tool before execution
+	// because the configured per-run x402 spend budget would be exceeded.
+	RefusedSpendBudget = "spend_budget"
 )
 
 // RunInfo describes the agent run a tool call belongs to. The agent

ai/retry.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"math/rand/v2"
 	"net/http"
 	"strconv"
 	"strings"
@@ -130,6 +131,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 +187,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 +223,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 +244,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
 	}

ai/retry_test.go

@@ -423,3 +423,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)
+		}
+	}
+}

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

@@ -106,7 +106,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

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

internal/harness/a2a-streaming/main.go

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

internal/harness/provider-conformance/main.go

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

internal/harness/provider-conformance/main_test.go

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

internal/website/docs/guides/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/x402-payments.md

@@ -100,7 +100,21 @@ c := &x402.Client{
 resp, err := c.Do(req) // a 402 is paid and retried; over-budget calls error instead
 ```
 
-`Payer` is an interface (`Pay(ctx, Requirements) (payment string, error)`) — the consumer counterpart to `Facilitator`. The budget accumulates across calls, so a long-running agent can be handed a fixed allowance for a task. Budget is reserved before payment is created, which means parallel paid calls cannot race past the cap; if payment creation or verification fails, the reservation is released. (The agent-level `AgentMaxSpend` option, wiring this into the agent loop next to `MaxSteps`/`ApproveTool`, is the next step.)
+`Payer` is an interface (`Pay(ctx, Requirements) (payment string, error)`) — the consumer counterpart to `Facilitator`. The budget accumulates across calls, so a long-running agent can be handed a fixed allowance for a task. Budget is reserved before payment is created, which means parallel paid calls cannot race past the cap; if payment creation or verification fails, the reservation is released.
+
+## Agent-level spend guardrail
+
+For unattended agents, set the same cap at the agent tool-execution layer so paid tools are refused before their handler — and therefore before a payer — can run:
+
+```go
+agent := micro.NewAgent("buyer",
+    micro.AgentMaxSteps(8),
+    micro.AgentMaxSpend(20_000), // per Ask, smallest units
+    micro.AgentToolSpend("weather.Weather.Forecast", 10_000),
+)
+```
+
+`AgentMaxSpend` is disabled by default (`0`). `AgentToolSpend` records the price discovered from your shoppable MCP/x402 catalog for the tools this agent may call. When a call would exceed the per-run allowance, the result is a normal structured guardrail refusal with `Refused: "spend_budget"` and an explanatory error in the run timeline/inspect output, distinct from provider/model failures.
 
 ### Live facilitator conformance
 
@@ -121,7 +135,7 @@ Leave those variables unset in normal CI; the live test skips unless the facilit
 
 - **Opt-in.** No pay-to address (and no config), no payments — nothing changes.
 - **No crypto in the framework.** The facilitator does verification and settlement on-chain; Go Micro speaks HTTP.
-- **A paying agent needs a budget.** On the agent side, an unattended agent that spends money needs a spend cap next to `MaxSteps` and `ApproveTool` — see [Plan & Delegate](plan-delegate.html) for the guardrail model. This is active work.
+- **A paying agent needs a budget.** Use `AgentMaxSpend` plus `AgentToolSpend` next to `MaxSteps` and `ApproveTool` so a run has an explicit allowance before any paid tool can execute.
 
 ## See also
 

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

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

micro.go

@@ -120,6 +120,16 @@ func AgentLoopLimit(n int) AgentOption { return agent.LoopLimit(n) }
 // each action the agent takes.
 func AgentApproveTool(fn ApproveFunc) AgentOption { return agent.ApproveTool(fn) }
 
+// AgentMaxSpend bounds paid x402 tool spend per Ask, in the asset's smallest
+// unit (0 = disabled). Calls that would exceed it are refused before payment.
+func AgentMaxSpend(amount int64) AgentOption { return agent.MaxSpend(amount) }
+
+// AgentToolSpend records the x402 price for a tool, in the asset's smallest
+// unit, so AgentMaxSpend can reserve budget before execution.
+func AgentToolSpend(tool string, amount int64) AgentOption {
+	return agent.ToolSpend(tool, amount)
+}
+
 // AgentModelCallTimeout sets the timeout for each provider Generate call.
 func AgentModelCallTimeout(d time.Duration) AgentOption { return agent.ModelCallTimeout(d) }