Update architect priorities after CLI inspection

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

.github/workflows/harness.yml

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

CHANGELOG.md

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

agent/agent.go

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

agent/builtin.go

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

agent/builtin_test.go

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

agent/checkpoint.go

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

agent/checkpoint_test.go

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

agent/memory.go

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

agent/memory_test.go

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

agent/otel.go

@@ -253,9 +253,40 @@ func (a *agentImpl) recordSpanEvent(span trace.Span, e RunEvent) {
 		e.TraceID = sc.TraceID().String()
 		e.SpanID = sc.SpanID().String()
 	}
+	span.AddEvent("agent."+e.Kind, trace.WithTimestamp(e.Time), trace.WithAttributes(runEventAttributes(e)...))
 	a.recordRunEvent(e)
 }
 
+func runEventAttributes(e RunEvent) []attribute.KeyValue {
+	attrs := []attribute.KeyValue{
+		attribute.String(AttrRunID, e.RunID),
+		attribute.String(AttrAgentName, e.Agent),
+	}
+	if e.ParentID != "" {
+		attrs = append(attrs, attribute.String(AttrParentRunID, e.ParentID))
+	}
+	if e.Name != "" {
+		attrs = append(attrs, attribute.String("agent.event.name", e.Name))
+	}
+	if e.Provider != "" {
+		attrs = append(attrs, attribute.String(AttrProvider, e.Provider))
+	}
+	if e.Model != "" {
+		attrs = append(attrs, attribute.String(AttrModel, e.Model))
+	}
+	if e.LatencyMS > 0 {
+		attrs = append(attrs, attribute.Int64(AttrLatencyMS, e.LatencyMS))
+	}
+	attrs = appendUsage(attrs, e.Tokens)
+	if e.Refused != "" {
+		attrs = append(attrs, attribute.Bool(AttrGuardrailBlock, true), attribute.String(AttrRefusal, e.Refused))
+	}
+	if e.Error != "" {
+		attrs = append(attrs, attribute.String("agent.error", e.Error))
+	}
+	return attrs
+}
+
 func (a *agentImpl) recordRunEvent(e RunEvent) {
 	if e.RunID == "" {
 		return

agent/otel_test.go

@@ -71,6 +71,16 @@ func TestAgentOpenTelemetrySpans(t *testing.T) {
 	if runID == "" {
 		t.Fatal("run span missing run id attribute")
 	}
+	var runEvents []trace.Event
+	for _, s := range spans {
+		if s.Name() == spanNameRun {
+			runEvents = s.Events()
+			break
+		}
+	}
+	if !spanEventHasRunInfo(runEvents, "agent.run", runID, "runner") || !spanEventHasRunInfo(runEvents, "agent.done", runID, "runner") {
+		t.Fatalf("run span missing run-info events: %#v", runEvents)
+	}
 	for _, s := range spans {
 		if s.Name() != spanNameModelCall && s.Name() != spanNameToolCall {
 			continue
@@ -115,6 +125,19 @@ func TestAgentOpenTelemetrySpans(t *testing.T) {
 	}
 }
 
+func spanEventHasRunInfo(events []trace.Event, name, runID, agentName string) bool {
+	for _, event := range events {
+		if event.Name != name {
+			continue
+		}
+		attrs := spanAttributes(event.Attributes)
+		if attrs[AttrRunID] == runID && attrs[AttrAgentName] == agentName {
+			return true
+		}
+	}
+	return false
+}
+
 func spanAttributes(attrs []attribute.KeyValue) map[string]string {
 	out := make(map[string]string, len(attrs))
 	for _, attr := range attrs {

ai/anthropic/anthropic.go

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

ai/atlascloud/atlascloud.go

@@ -20,6 +20,7 @@
 package atlascloud
 
 import (
+	"bufio"
 	"bytes"
 	"context"
 	"encoding/json"
@@ -91,13 +92,21 @@ func (p *Provider) Generate(ctx context.Context, req *ai.Request, opts ...ai.Gen
 
 	messages := []map[string]any{
 		{"role": "system", "content": req.SystemPrompt},
-		{"role": "user", "content": req.Prompt},
+	}
+	for _, m := range req.Messages {
+		messages = append(messages, map[string]any{"role": m.Role, "content": m.Content})
+	}
+	if req.Prompt != "" {
+		messages = append(messages, map[string]any{"role": "user", "content": req.Prompt})
 	}
 
 	apiReq := map[string]any{
 		"model":    p.opts.Model,
 		"messages": messages,
 	}
+	if p.opts.MaxTokens > 0 {
+		apiReq["max_tokens"] = p.opts.MaxTokens
+	}
 
 	if len(tools) > 0 {
 		apiReq["tools"] = tools
@@ -142,8 +151,111 @@ func (p *Provider) Generate(ctx context.Context, req *ai.Request, opts ...ai.Gen
 	return resp, nil
 }
 
+// Stream generates a streaming response from Atlas Cloud's OpenAI-compatible
+// chat completions endpoint, emitting content deltas as they arrive.
 func (p *Provider) Stream(ctx context.Context, req *ai.Request, opts ...ai.GenerateOption) (ai.Stream, error) {
-	return nil, fmt.Errorf("%w: atlascloud provider", ai.ErrStreamingUnsupported)
+	messages := []map[string]any{
+		{"role": "system", "content": req.SystemPrompt},
+	}
+	for _, m := range req.Messages {
+		messages = append(messages, map[string]any{"role": m.Role, "content": m.Content})
+	}
+	if req.Prompt != "" {
+		messages = append(messages, map[string]any{"role": "user", "content": req.Prompt})
+	}
+	apiReq := map[string]any{
+		"model":          p.opts.Model,
+		"messages":       messages,
+		"stream":         true,
+		"stream_options": map[string]any{"include_usage": true},
+	}
+	if p.opts.MaxTokens > 0 {
+		apiReq["max_tokens"] = p.opts.MaxTokens
+	}
+	reqBody, err := json.Marshal(apiReq)
+	if err != nil {
+		return nil, fmt.Errorf("failed to marshal stream request: %w", err)
+	}
+	apiURL := strings.TrimRight(p.opts.BaseURL, "/") + "/v1/chat/completions"
+	httpReq, err := http.NewRequestWithContext(ctx, http.MethodPost, apiURL, bytes.NewReader(reqBody))
+	if err != nil {
+		return nil, fmt.Errorf("failed to create stream request: %w", err)
+	}
+	httpReq.Header.Set("Content-Type", "application/json")
+	httpReq.Header.Set("Accept", "text/event-stream")
+	httpReq.Header.Set("Authorization", "Bearer "+p.opts.APIKey)
+
+	httpResp, err := http.DefaultClient.Do(httpReq)
+	if err != nil {
+		return nil, fmt.Errorf("stream API request failed: %w", err)
+	}
+	if httpResp.StatusCode != http.StatusOK {
+		defer httpResp.Body.Close()
+		respBody, _ := io.ReadAll(httpResp.Body)
+		return nil, fmt.Errorf("stream API error (%s): %s", httpResp.Status, string(respBody))
+	}
+	return &atlasStream{body: httpResp.Body, scanner: bufio.NewScanner(httpResp.Body)}, nil
+}
+
+type atlasStream struct {
+	body    io.ReadCloser
+	scanner *bufio.Scanner
+	closed  bool
+}
+
+func (s *atlasStream) Recv() (*ai.Response, error) {
+	for s.scanner.Scan() {
+		line := strings.TrimSpace(s.scanner.Text())
+		if line == "" || strings.HasPrefix(line, ":") {
+			continue
+		}
+		if !strings.HasPrefix(line, "data:") {
+			continue
+		}
+		data := strings.TrimSpace(strings.TrimPrefix(line, "data:"))
+		if data == "[DONE]" {
+			return nil, io.EOF
+		}
+		var chunk struct {
+			Choices []struct {
+				Delta struct {
+					Content string `json:"content"`
+				} `json:"delta"`
+			} `json:"choices"`
+			Usage *struct {
+				PromptTokens     int `json:"prompt_tokens"`
+				CompletionTokens int `json:"completion_tokens"`
+				TotalTokens      int `json:"total_tokens"`
+			} `json:"usage"`
+		}
+		if err := json.Unmarshal([]byte(data), &chunk); err != nil {
+			return nil, fmt.Errorf("failed to parse stream chunk: %w", err)
+		}
+		if len(chunk.Choices) > 0 && chunk.Choices[0].Delta.Content != "" {
+			return &ai.Response{Reply: chunk.Choices[0].Delta.Content}, nil
+		}
+		// Final chunk (after include_usage) carries token usage and no content.
+		if chunk.Usage != nil {
+			return &ai.Response{Usage: ai.Usage{
+				InputTokens:  chunk.Usage.PromptTokens,
+				OutputTokens: chunk.Usage.CompletionTokens,
+				TotalTokens:  chunk.Usage.TotalTokens,
+			}}, nil
+		}
+		continue
+	}
+	if err := s.scanner.Err(); err != nil {
+		return nil, err
+	}
+	return nil, io.EOF
+}
+
+func (s *atlasStream) Close() error {
+	if s.closed {
+		return nil
+	}
+	s.closed = true
+	return s.body.Close()
 }
 
 func (p *Provider) callAPI(ctx context.Context, req map[string]any) (*ai.Response, map[string]any, error) {

ai/openai/openai.go

@@ -85,7 +85,12 @@ func (p *Provider) Generate(ctx context.Context, req *ai.Request, opts ...ai.Gen
 	// Build messages
 	messages := []map[string]any{
 		{"role": "system", "content": req.SystemPrompt},
-		{"role": "user", "content": req.Prompt},
+	}
+	for _, m := range req.Messages {
+		messages = append(messages, map[string]any{"role": m.Role, "content": m.Content})
+	}
+	if req.Prompt != "" {
+		messages = append(messages, map[string]any{"role": "user", "content": req.Prompt})
 	}
 
 	// Build initial request
@@ -93,6 +98,9 @@ func (p *Provider) Generate(ctx context.Context, req *ai.Request, opts ...ai.Gen
 		"model":    p.opts.Model,
 		"messages": messages,
 	}
+	if p.opts.MaxTokens > 0 {
+		apiReq["max_tokens"] = p.opts.MaxTokens
+	}
 
 	if len(openaiTools) > 0 {
 		apiReq["tools"] = openaiTools
@@ -146,12 +154,21 @@ func (p *Provider) Generate(ctx context.Context, req *ai.Request, opts ...ai.Gen
 func (p *Provider) Stream(ctx context.Context, req *ai.Request, opts ...ai.GenerateOption) (ai.Stream, error) {
 	messages := []map[string]any{
 		{"role": "system", "content": req.SystemPrompt},
-		{"role": "user", "content": req.Prompt},
+	}
+	for _, m := range req.Messages {
+		messages = append(messages, map[string]any{"role": m.Role, "content": m.Content})
+	}
+	if req.Prompt != "" {
+		messages = append(messages, map[string]any{"role": "user", "content": req.Prompt})
 	}
 	apiReq := map[string]any{
-		"model":    p.opts.Model,
-		"messages": messages,
-		"stream":   true,
+		"model":          p.opts.Model,
+		"messages":       messages,
+		"stream":         true,
+		"stream_options": map[string]any{"include_usage": true},
+	}
+	if p.opts.MaxTokens > 0 {
+		apiReq["max_tokens"] = p.opts.MaxTokens
 	}
 	reqBody, err := json.Marshal(apiReq)
 	if err != nil {
@@ -203,14 +220,27 @@ func (s *openAIStream) Recv() (*ai.Response, error) {
 					Content string `json:"content"`
 				} `json:"delta"`
 			} `json:"choices"`
+			Usage *struct {
+				PromptTokens     int `json:"prompt_tokens"`
+				CompletionTokens int `json:"completion_tokens"`
+				TotalTokens      int `json:"total_tokens"`
+			} `json:"usage"`
 		}
 		if err := json.Unmarshal([]byte(data), &chunk); err != nil {
 			return nil, fmt.Errorf("failed to parse stream chunk: %w", err)
 		}
-		if len(chunk.Choices) == 0 || chunk.Choices[0].Delta.Content == "" {
-			continue
+		if len(chunk.Choices) > 0 && chunk.Choices[0].Delta.Content != "" {
+			return &ai.Response{Reply: chunk.Choices[0].Delta.Content}, nil
 		}
-		return &ai.Response{Reply: chunk.Choices[0].Delta.Content}, nil
+		// Final chunk (after include_usage) carries token usage and no content.
+		if chunk.Usage != nil {
+			return &ai.Response{Usage: ai.Usage{
+				InputTokens:  chunk.Usage.PromptTokens,
+				OutputTokens: chunk.Usage.CompletionTokens,
+				TotalTokens:  chunk.Usage.TotalTokens,
+			}}, nil
+		}
+		continue
 	}
 	if err := s.scanner.Err(); err != nil {
 		return nil, err

ai/openai/openai_test.go

@@ -8,6 +8,7 @@ import (
 	"net/http"
 	"net/http/httptest"
 	"testing"
+	"time"
 
 	"go-micro.dev/v6/ai"
 )
@@ -126,6 +127,58 @@ func TestProvider_Stream(t *testing.T) {
 	}
 }
 
+func TestProvider_StreamPropagatesMalformedChunk(t *testing.T) {
+	ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		w.Header().Set("Content-Type", "text/event-stream")
+		_, _ = w.Write([]byte("data: {bad json}\n\n"))
+	}))
+	defer ts.Close()
+
+	p := NewProvider(ai.WithAPIKey("test-key"), ai.WithBaseURL(ts.URL))
+	stream, err := p.Stream(context.Background(), &ai.Request{Prompt: "Hello"})
+	if err != nil {
+		t.Fatalf("Stream returned error: %v", err)
+	}
+	defer stream.Close()
+
+	if _, err := stream.Recv(); err == nil {
+		t.Fatal("Recv returned nil error for malformed chunk")
+	}
+}
+
+func TestProvider_StreamCloseReleasesResponse(t *testing.T) {
+	released := make(chan struct{})
+	ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		w.Header().Set("Content-Type", "text/event-stream")
+		_, _ = w.Write([]byte("data: {\"choices\":[{\"delta\":{\"content\":\"hel\"}}]}\n\n"))
+		if f, ok := w.(http.Flusher); ok {
+			f.Flush()
+		}
+		<-r.Context().Done()
+		close(released)
+	}))
+	defer ts.Close()
+
+	p := NewProvider(ai.WithAPIKey("test-key"), ai.WithBaseURL(ts.URL))
+	stream, err := p.Stream(context.Background(), &ai.Request{Prompt: "Hello"})
+	if err != nil {
+		t.Fatalf("Stream returned error: %v", err)
+	}
+	first, err := stream.Recv()
+	if err != nil || first.Reply != "hel" {
+		t.Fatalf("first chunk = %#v, %v; want hel", first, err)
+	}
+	if err := stream.Close(); err != nil {
+		t.Fatalf("Close returned error: %v", err)
+	}
+
+	select {
+	case <-released:
+	case <-time.After(time.Second):
+		t.Fatal("server did not observe closed stream request")
+	}
+}
+
 func TestProvider_ImageRegistration(t *testing.T) {
 	ig := ai.NewImage("openai", ai.WithAPIKey("test"))
 	if ig == nil {

ai/options.go

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

ai/retry.go

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

ai/retry_test.go

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

cmd/micro/cli/deploy/deploy.go

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

cmd/micro/cli/deploy/deploy_test.go

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

cmd/micro/cli/new/contract_test.go

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

cmd/micro/inspect/inspect.go

@@ -0,0 +1,169 @@
+// Package inspect registers the 'micro inspect' CLI command.
+package inspect
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"io"
+	"os"
+
+	"github.com/urfave/cli/v2"
+	goagent "go-micro.dev/v6/agent"
+	"go-micro.dev/v6/cmd"
+	aiflow "go-micro.dev/v6/flow"
+	"go-micro.dev/v6/store"
+)
+
+func init() {
+	cmd.Register(&cli.Command{
+		Name:  "inspect",
+		Usage: "Inspect recent agent and workflow activity",
+		Description: `Inspect is the CLI checkpoint in the local scaffold → run → chat → inspect loop.
+It reads durable local run history, so it works after the agent or flow has stopped.`,
+		Subcommands: []*cli.Command{
+			{
+				Name:      "agent",
+				Usage:     "Show recent recorded runs for an agent",
+				ArgsUsage: "[agent]",
+				Flags:     inspectAgentFlags(),
+				Action:    inspectAgent,
+			},
+			{
+				Name:      "flow",
+				Usage:     "Show durable run history for a flow",
+				ArgsUsage: "[flow]",
+				Flags:     inspectFlowFlags(),
+				Action:    inspectFlow,
+			},
+		},
+	})
+}
+
+func inspectAgentFlags() []cli.Flag {
+	return []cli.Flag{
+		&cli.BoolFlag{Name: "json", Usage: "Print run summaries as JSON for automation"},
+		&cli.StringFlag{Name: "status", Usage: "Only show runs with this status (running, done, error, refused)"},
+		&cli.StringFlag{Name: "trace", Usage: "Only show runs whose trace id matches this full id or prefix"},
+		&cli.IntFlag{Name: "limit", Usage: "Show the most recently updated N runs"},
+	}
+}
+
+func inspectFlowFlags() []cli.Flag {
+	return []cli.Flag{
+		&cli.BoolFlag{Name: "json", Usage: "Print durable run history as JSON for automation"},
+		&cli.BoolFlag{Name: "pending", Usage: "Only show runs that have not completed"},
+		&cli.StringFlag{Name: "status", Usage: "Only show runs with this status (running, done, failed)"},
+		&cli.IntFlag{Name: "limit", Usage: "Show the most recently updated N runs"},
+		&cli.StringFlag{Name: "stage", Usage: "Only show runs currently checkpointed at this stage"},
+	}
+}
+
+func inspectAgent(c *cli.Context) error {
+	name := c.Args().First()
+	if name == "" {
+		return fmt.Errorf("agent name required: micro inspect agent <name>")
+	}
+	opts := goagent.RunListOptions{Status: c.String("status"), TraceID: c.String("trace"), Limit: c.Int("limit")}
+	runs, err := goagent.ListRunSummariesWithOptions(store.DefaultStore, name, opts)
+	if err != nil {
+		return err
+	}
+	return writeAgentInspection(os.Stdout, name, runs, c.Bool("json"))
+}
+
+func writeAgentInspection(w io.Writer, name string, runs []goagent.RunSummary, asJSON bool) error {
+	if asJSON {
+		enc := json.NewEncoder(w)
+		enc.SetIndent("", "  ")
+		return enc.Encode(runs)
+	}
+	if len(runs) == 0 {
+		fmt.Fprintf(w, "  No agent runs recorded for %q. After chatting, try: micro inspect agent %s\n", name, name)
+		return nil
+	}
+	fmt.Fprintf(w, "  Agent %q runs\n", name)
+	for _, run := range runs {
+		fmt.Fprintf(w, "  %s  status=%s  events=%d  last=%s", run.RunID, run.Status, run.Events, run.LastKind)
+		if run.LastError != "" {
+			fmt.Fprintf(w, "  error=%q", run.LastError)
+		}
+		if run.TraceID != "" {
+			fmt.Fprintf(w, "  trace=%s", shortID(run.TraceID))
+		}
+		fmt.Fprintln(w)
+	}
+	return nil
+}
+
+func inspectFlow(c *cli.Context) error {
+	name := c.Args().First()
+	if name == "" {
+		return fmt.Errorf("flow name required: micro inspect flow <name>")
+	}
+	runs, err := aiflow.StoreCheckpoint(nil, name).List(context.Background())
+	if err != nil {
+		return err
+	}
+	runs = filterFlowInspection(runs, c.Bool("pending"), c.String("status"), c.String("stage"), c.Int("limit"))
+	return writeFlowInspection(os.Stdout, name, runs, c.Bool("json"), c.Bool("pending"))
+}
+
+func filterFlowInspection(runs []aiflow.Run, pending bool, status, stage string, limit int) []aiflow.Run {
+	filtered := make([]aiflow.Run, 0, len(runs))
+	for _, run := range runs {
+		if pending && run.Status == "done" {
+			continue
+		}
+		if status != "" && run.Status != status {
+			continue
+		}
+		if stage != "" && run.State.Stage != stage {
+			continue
+		}
+		filtered = append(filtered, run)
+	}
+	if limit > 0 && len(filtered) > limit {
+		return filtered[len(filtered)-limit:]
+	}
+	return filtered
+}
+
+func writeFlowInspection(w io.Writer, name string, runs []aiflow.Run, asJSON, pending bool) error {
+	if asJSON {
+		enc := json.NewEncoder(w)
+		enc.SetIndent("", "  ")
+		return enc.Encode(runs)
+	}
+	if len(runs) == 0 {
+		if pending {
+			fmt.Fprintf(w, "  No pending flow runs recorded for %q.\n", name)
+			return nil
+		}
+		fmt.Fprintf(w, "  No flow runs recorded for %q. After executing a durable flow, try: micro inspect flow %s\n", name, name)
+		return nil
+	}
+	fmt.Fprintf(w, "  Flow %q runs\n", name)
+	for _, run := range runs {
+		stage := run.State.Stage
+		if stage == "" {
+			stage = "-"
+		}
+		fmt.Fprintf(w, "  %s  status=%s  stage=%s  steps=%d", shortID(run.ID), run.Status, stage, len(run.Steps))
+		for _, step := range run.Steps {
+			if step.Error != "" {
+				fmt.Fprintf(w, "  error=%q", step.Error)
+				break
+			}
+		}
+		fmt.Fprintln(w)
+	}
+	return nil
+}
+
+func shortID(id string) string {
+	if len(id) <= 12 {
+		return id
+	}
+	return id[:12]
+}

cmd/micro/inspect/inspect_test.go

@@ -0,0 +1,64 @@
+package inspect
+
+import (
+	"bytes"
+	"encoding/json"
+	"strings"
+	"testing"
+
+	goagent "go-micro.dev/v6/agent"
+	aiflow "go-micro.dev/v6/flow"
+)
+
+func TestWriteAgentInspectionIncludesActionableBreadcrumbs(t *testing.T) {
+	runs := []goagent.RunSummary{{RunID: "run-1", Status: "error", Events: 4, LastKind: "tool", LastError: "boom", TraceID: "1234567890abcdef"}}
+	var out bytes.Buffer
+	if err := writeAgentInspection(&out, "support", runs, false); err != nil {
+		t.Fatal(err)
+	}
+	got := out.String()
+	for _, want := range []string{"Agent \"support\" runs", "run-1", "status=error", "events=4", "last=tool", `error="boom"`, "trace=1234567890ab"} {
+		if !strings.Contains(got, want) {
+			t.Fatalf("output missing %q:\n%s", want, got)
+		}
+	}
+}
+
+func TestWriteAgentInspectionEmptyStateNamesInspectCommand(t *testing.T) {
+	var out bytes.Buffer
+	if err := writeAgentInspection(&out, "support", nil, false); err != nil {
+		t.Fatal(err)
+	}
+	if got := out.String(); !strings.Contains(got, "micro inspect agent support") {
+		t.Fatalf("empty state missing next step: %q", got)
+	}
+}
+
+func TestWriteFlowInspectionIncludesFailedStepBreadcrumb(t *testing.T) {
+	runs := []aiflow.Run{{ID: "1234567890abcdef", Status: "failed", State: aiflow.State{Stage: "charge"}, Steps: []aiflow.StepRecord{{Name: "charge", Status: "failed", Error: "card declined"}}}}
+	var out bytes.Buffer
+	if err := writeFlowInspection(&out, "checkout", runs, false, false); err != nil {
+		t.Fatal(err)
+	}
+	got := out.String()
+	for _, want := range []string{"Flow \"checkout\" runs", "1234567890ab", "status=failed", "stage=charge", "steps=1", `error="card declined"`} {
+		if !strings.Contains(got, want) {
+			t.Fatalf("output missing %q:\n%s", want, got)
+		}
+	}
+}
+
+func TestWriteFlowInspectionJSON(t *testing.T) {
+	runs := []aiflow.Run{{ID: "run-1", Flow: "checkout", Status: "done"}}
+	var out bytes.Buffer
+	if err := writeFlowInspection(&out, "checkout", runs, true, false); err != nil {
+		t.Fatal(err)
+	}
+	var got []aiflow.Run
+	if err := json.Unmarshal(out.Bytes(), &got); err != nil {
+		t.Fatalf("invalid JSON: %v\n%s", err, out.String())
+	}
+	if len(got) != 1 || got[0].ID != "run-1" || got[0].Status != "done" {
+		t.Fatalf("decoded runs = %+v", got)
+	}
+}

cmd/micro/main.go

@@ -12,6 +12,7 @@ import (
 	_ "go-micro.dev/v6/cmd/micro/cli/build"
 	_ "go-micro.dev/v6/cmd/micro/cli/deploy"
 	_ "go-micro.dev/v6/cmd/micro/flow"
+	_ "go-micro.dev/v6/cmd/micro/inspect"
 	_ "go-micro.dev/v6/cmd/micro/mcp"
 	_ "go-micro.dev/v6/cmd/micro/resource"
 	_ "go-micro.dev/v6/cmd/micro/run"

cmd/micro/zero_to_hero_cli_test.go

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

examples/agent-human-input/README.md

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

examples/support/main.go

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

examples/support/main_test.go

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

gateway/a2a/a2a.go

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

gateway/a2a/a2a_test.go

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

internal/docs/PRIORITIES.md

@@ -19,14 +19,11 @@ redirect the loop; direction always wins.
 items the loop can auto-merge): brand/positioning copy, breaking public-API
 changes, architectural rewrites. Those go to the human.
 
-## Now (ranked)
+## Developer experience (ranked)
 
-1. **CI-verify the 0-to-1 getting-started path** ([#3234](https://github.com/micro/go-micro/issues/3234)) — add a deterministic no-key check for scaffold → run/build → call so the README/blog promise that building a service stays effortless cannot regress while the agent stack deepens.
-2. **CI-verify the 0-to-hero agent workflow** ([#3241](https://github.com/micro/go-micro/issues/3241)) — after the basic service contract is guarded, make the full services → agents → workflows story executable in CI with a maintained no-secret reference scenario for run → chat → inspect boundaries.
-
-## Later (ranked)
-
-3. **Add A2A resubscribe and input-required handoff support** ([#3235](https://github.com/micro/go-micro/issues/3235)) — after push notifications and multi-turn continuation shipped, finish the remaining long-running A2A interoperability gap: reconnecting to live task streams and carrying human-input-required handoffs through the gateway.
+1. **Add durable agent run checkpoint and resume** ([#3306](https://github.com/micro/go-micro/issues/3306)) — the CLI inspection foothold has shipped, so the highest-value Next-roadmap seam is making agent loops resumable like flows: long-running work must survive restarts without unsafe replay or hidden state loss before deeper streaming and observability can be trusted.
+2. **Broaden provider-backed AI streaming coverage** ([#3315](https://github.com/micro/go-micro/issues/3315)) — after checkpoint/resume, extend tested `ai.Stream` coverage across remaining providers so `micro chat`, A2A streaming, and long-running agent interactions behave consistently instead of depending on adapter-specific gaps.
+3. **Emit agent RunInfo as OpenTelemetry spans** ([#3316](https://github.com/micro/go-micro/issues/3316)) — once runs are durable and streaming paths are consistent, turn agent run timelines into correlated spans so operators can inspect model calls, tool calls, failures, and run IDs through the same observability surface as services and flows.
 
 _Seeded by Claude Code from the roadmap + open issues; thereafter maintained by the
 architecture-review pass._

internal/harness/provider-conformance/README.md

@@ -0,0 +1,84 @@
+# Provider conformance
+
+This harness keeps the services → agents → workflows lifecycle honest across the
+supported AI providers. It runs the same end-to-end scenarios against each
+configured provider and treats missing provider keys as an explicit skip, so the
+suite is safe for local development, forks, and scheduled CI.
+
+## What it exercises
+
+`go run ./internal/harness/provider-conformance` fans out over the harnesses in
+`internal/harness`:
+
+- `universe` — service discovery plus agent tool calls over the real runtime.
+- `agent-flow` — a workflow event that drives an agent to call services.
+- `plan-delegate` — plan persistence plus agent-to-agent delegation and service
+  calls.
+
+The command also emits the registered provider capability matrix so the run shows
+which providers advertise model, image, video, and streaming support.
+
+## Local usage
+
+Run the deterministic path with no secrets:
+
+```sh
+go run ./internal/harness/provider-conformance -providers mock
+```
+
+Run every live provider that has a key in the environment:
+
+```sh
+go run ./internal/harness/provider-conformance \
+  -summary-json provider-conformance-summary.json \
+  -summary-markdown provider-conformance-summary.md \
+  -capabilities-markdown provider-capabilities.md
+```
+
+Provider keys are read from `MICRO_AI_API_KEY` or the provider-specific variable:
+
+| Provider | Secret / environment variable |
+| --- | --- |
+| Anthropic | `ANTHROPIC_API_KEY` |
+| OpenAI | `OPENAI_API_KEY` |
+| Gemini | `GEMINI_API_KEY` |
+| Groq | `GROQ_API_KEY` |
+| Mistral | `MISTRAL_API_KEY` |
+| Together | `TOGETHER_API_KEY` |
+| AtlasCloud | `ATLASCLOUD_API_KEY` |
+
+Use `-require-configured` when you want a selected provider without a key to fail
+instead of skip:
+
+```sh
+go run ./internal/harness/provider-conformance \
+  -providers anthropic,openai \
+  -require-configured
+```
+
+## Scheduled CI behavior
+
+The `Harness (E2E)` workflow runs on pushes and pull requests with deterministic
+mock LLMs. On the daily schedule and manual dispatch it also runs the live
+provider conformance job. That job:
+
+1. reads the provider keys from repository secrets,
+2. skips providers whose secrets are absent,
+3. fails when any configured provider fails a harness, and
+4. uploads JSON and Markdown coverage artifacts for the run.
+
+The job also appends the Markdown summary and capability matrix to the GitHub
+Actions step summary, making configured, skipped, and failed provider coverage
+visible without downloading artifacts.
+
+## Adding a provider
+
+To bring a new provider into scheduled conformance:
+
+1. register its `ai` provider implementation and capability metadata,
+2. add the provider name and key variable to `providerEnv` in `main.go`,
+3. import the provider package in `main.go`,
+4. pass the matching repository secret through `.github/workflows/harness.yml`,
+   and
+5. run `go run ./internal/harness/provider-conformance -providers <name> \
+   -require-configured` with a live key before opening the change.

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

@@ -0,0 +1,17 @@
+# 0→hero CI harness
+
+This directory owns the no-secret reference scenario for the Go Micro
+services → agents → workflows lifecycle. It is intentionally small and
+scripted so CI can run it on every push without external services or model keys.
+
+`run.sh` verifies three boundaries together:
+
+1. **Run** — `micro run` remains available as the local development entry point.
+2. **Chat** — `micro chat` remains available as the interactive agent entry point.
+3. **Inspect** — `micro inspect agent <name>` and `micro inspect flow <name>`
+   remain available as the local run-history inspection step, with `micro flow
+   runs` preserving durable workflow history inspection.
+
+After the CLI boundary smoke checks, the script runs the deterministic harnesses
+that boot real services, agents, workflows, store-backed run history, and A2A
+with only the LLM mocked.

internal/harness/zero-to-hero-ci/run.sh

@@ -0,0 +1,13 @@
+#!/usr/bin/env bash
+set -euo pipefail
+
+ROOT="$(cd "$(dirname "${BASH_SOURCE[0]}")/../../.." && pwd)"
+cd "$ROOT"
+
+# Keep the developer inner-loop boundaries executable and discoverable in CI
+# without secrets or long-running daemons.
+go test ./cmd/micro -run TestZeroToHeroCLIBoundaries -count=1
+
+# Deterministic no-secret reference scenarios. These use the real Go Micro
+# runtime and mock only the LLM provider.
+go test ./internal/harness/universe ./internal/harness/plan-delegate -run 'Test.*Harness|TestPlanDelegateEndToEnd|TestPlanDelegateFlowHandoff' -count=1

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

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