docs: align public TLS and A2A status

.github/workflows/harness.yml

@@ -3,8 +3,8 @@ name: Harness (E2E)
 # Runs the end-to-end harnesses for agents, services, flows, and provider
 # conformance. The default job uses deterministic mock LLMs and needs no
 # secrets. A second job runs the same harnesses against the live provider set and
-# skips providers whose secrets are absent, so scheduled conformance
-# remains safe in no-key forks while still failing configured providers that drift.
+# fails if any selected provider secret is missing, so scheduled conformance
+# cannot silently degrade.
 
 on:
   push:
@@ -27,8 +27,14 @@ jobs:
           cache: true
       - name: Build
         run: go build ./...
-      - name: 0→1 and 0→hero developer-flow harness
-        run: make harness
+      - name: 0→1 scaffold contract
+        run: go test ./cmd/micro/cli/new -run TestZeroToOneContract -count=1
+      - name: Universe end-to-end (asserts; exits non-zero on failure)
+        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
 
   harness-live:
     name: Provider harnesses (live LLM conformance)
@@ -55,23 +61,10 @@ jobs:
           ATLASCLOUD_API_KEY: ${{ secrets.ATLASCLOUD_API_KEY }}
         run: |
           go run ./internal/harness/provider-conformance \
+            -require-configured \
             -summary-json provider-conformance-summary.json \
             -summary-markdown provider-conformance-summary.md \
             -capabilities-markdown provider-capabilities.md
-      - name: Publish provider conformance summary
-        if: always()
-        run: |
-          if [ -f provider-conformance-summary.md ]; then
-            cat provider-conformance-summary.md >> "$GITHUB_STEP_SUMMARY"
-          fi
-          if [ -f provider-capabilities.md ]; then
-            {
-              echo
-              echo "## Registered provider capabilities"
-              echo
-              cat provider-capabilities.md
-            } >> "$GITHUB_STEP_SUMMARY"
-          fi
       - name: Upload provider conformance summary
         if: always()
         uses: actions/upload-artifact@v4

CHANGELOG.md

@@ -19,7 +19,7 @@ else is additive. See the [v5 → v6 migration guide](internal/website/docs/guid
 - **JWT auth ported in-module.** The external `github.com/micro/plugins/v5/auth/jwt` (pinned to v5) is replaced by `go-micro.dev/v6/auth/jwt/token`, now on the maintained `golang-jwt/jwt/v5`; the deprecated `dgrijalva/jwt-go` dependency is dropped.
 
 ### Added
-- **A2A protocol — both directions** — `gateway/a2a` exposes registered agents over the open Agent2Agent (A2A) protocol so agents on other frameworks can discover and call them: Agent Cards are generated from registry metadata (the same way the MCP gateway derives tools), and incoming tasks are translated to the agent's existing `Agent.Chat` RPC, with no per-agent code (`micro a2a serve`). The outbound `a2a.Client` calls external A2A agents by URL, wired into `flow.A2A(url)` (a workflow step) and `delegate` to an `http(s)` URL (from inside an agent). An agent can also serve A2A **directly** without a gateway via `AgentA2A(addr)` (`a2a.NewAgentHandler`), handling tasks in-process. 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/`)
+- **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/`)
 - **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)

Makefile

@@ -18,7 +18,7 @@ help:
 	@echo "  make test-race     - Run tests with race detector"
 	@echo "  make test-coverage - Run tests with coverage"
 	@echo "  make lint          - Run linter"
-	@echo "  make harness       - Run deterministic getting-started and end-to-end harnesses"
+	@echo "  make harness       - Run deterministic end-to-end harnesses"
 	@echo "  make provider-conformance - Run harnesses against configured live providers"
 	@echo "  make fmt           - Format code"
 	@echo "  make install-tools - Install development tools"
@@ -42,15 +42,12 @@ test-coverage:
 	go tool cover -html=coverage.out -o coverage.html
 	@echo "Coverage report: coverage.html"
 
-# Run the documented getting-started contracts plus the deterministic
-# services → agents → workflows harnesses (mock LLM — no API key).
-# This mirrors the default CI path so local dogfooding catches scaffold,
-# run/chat/inspect, and 0→hero regressions before a PR is opened.
+# Run the end-to-end harnesses (deterministic, mock LLM — no API key).
+# The universe harness exits non-zero on assertion failure.
 harness:
-	go test ./cmd/micro/cli/new -run TestZeroToOne -count=1
-	./internal/harness/zero-to-hero-ci/run.sh
+	go run ./internal/harness/universe
 	go run ./internal/harness/agent-flow
-	go run ./internal/harness/provider-conformance -providers mock
+	go run ./internal/harness/plan-delegate # 0→hero: services + agents + flow + plan/delegate
 
 # Run the same harnesses against every configured live provider. Providers
 # without API keys are skipped; configured providers must pass.

README.md

@@ -64,14 +64,6 @@ curl -X POST http://localhost:8080/api/helloworld/Helloworld.Call \
   -H 'Content-Type: application/json' -d '{"name":"World"}'
 ```
 
-This scaffold → run → call path is covered by the no-secret CI harness. To run
-the same local contract (including the 0→hero services → agents → workflows path,
-chat/inspect CLI boundaries, and deploy dry-run), use:
-
-```bash
-make harness
-```
-
 ### Generate from a prompt — with an LLM key
 
 Set a provider key, describe what you want, and the AI designs services, writes handlers, compiles, and starts them:

agent/agent.go

@@ -92,11 +92,6 @@ 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.
@@ -139,35 +134,19 @@ func (a *agentImpl) String() string {
 }
 
 func (a *agentImpl) setup() {
-	a.setupWithToolHandler(nil)
-}
-
-func (a *agentImpl) setupWithToolHandler(handler ai.ToolHandler) {
 	var modelOpts []ai.Option
 	modelOpts = append(modelOpts, ai.WithAPIKey(a.opts.APIKey))
 	if a.opts.Model != "" {
 		modelOpts = append(modelOpts, ai.WithModel(a.opts.Model))
 	}
 
-	// Reuse the existing tools instance: its name map is populated by
-	// discoverTools, and rebuilding it here would orphan a base handler that
-	// already captured the old instance (breaking StreamAsk tool resolution).
-	if a.tools == nil {
-		a.tools = ai.NewTools(a.opts.Registry, ai.ToolClient(a.opts.Client))
-	}
-	if handler == nil {
-		handler = a.toolHandler()
-	}
-	modelOpts = append(modelOpts, ai.WithToolHandler(handler))
+	a.tools = ai.NewTools(a.opts.Registry, ai.ToolClient(a.opts.Client))
+	modelOpts = append(modelOpts, ai.WithToolHandler(a.toolHandler()))
 	a.model = ai.New(a.opts.Provider, modelOpts...)
 	if a.model != nil {
 		a.model = a.tracedModel(a.model)
 	}
 
-	if a.mem != nil {
-		return
-	}
-
 	// Memory is pluggable. Use the configured one, otherwise the default
 	// store-backed memory — except ephemeral sub-agents, which keep an
 	// isolated, non-persistent context.
@@ -223,6 +202,17 @@ 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) {
@@ -241,18 +231,16 @@ func (a *agentImpl) ask(ctx context.Context, message, parentRunID string) (*Resp
 		a.setup()
 	}
 
-	return a.askLocked(ctx, uuid.New().String(), message, parentRunID, nil, true)
+	return a.askLocked(ctx, uuid.New().String(), message, parentRunID, nil)
 }
 
-func (a *agentImpl) askLocked(ctx context.Context, runID, message, parentRunID string, existing *flow.Run, addUserMessage bool) (*Response, error) {
+func (a *agentImpl) askLocked(ctx context.Context, runID, message, parentRunID string, existing *flow.Run) (*Response, error) {
 	toolList, err := a.discoverTools()
 	if err != nil {
 		return nil, fmt.Errorf("discover tools: %w", err)
 	}
 
-	if addUserMessage {
-		a.mem.Add("user", message)
-	}
+	a.mem.Add("user", message)
 	a.steps = 0
 	a.calls = map[string]int{}
 	a.pause = nil
@@ -265,8 +253,6 @@ 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
 	}
@@ -297,9 +283,6 @@ 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
 	}
@@ -307,10 +290,6 @@ 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
@@ -347,12 +326,6 @@ 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,9 +20,8 @@ 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"
-	toolHumanInput = "request_input"
+	toolPlan     = "plan"
+	toolDelegate = "delegate"
 )
 
 // builtinTools returns the tool definitions exposed to the model in
@@ -42,18 +41,6 @@ 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,
@@ -91,9 +78,6 @@ 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
@@ -113,15 +97,13 @@ func Builtins(opts ...Option) (tools []ai.Tool, handle func(name string, input m
 // prevents runaway recursion).
 func (a *agentImpl) toolHandler() ai.ToolHandler {
 	if a.ephemeral {
-		return a.toolTimeoutWrap(a.tools.Handler())
+		return a.tools.Handler()
 	}
 
 	// Innermost first: base, then guardrails (approve → loop → step →
 	// plan), then developer wrappers outermost. Wrapping reverses order,
-	// so the result runs plan → step → loop → approve → checkpoint → base.
+	// so the result runs plan → step → loop → approve → base.
 	h := a.baseHandler()
-	h = a.toolTimeoutWrap(h)
-	h = a.checkpointToolWrap(h)
 	h = a.approveWrap(h)
 	h = a.loopWrap(h)
 	h = a.stepWrap(h)
@@ -149,21 +131,6 @@ func contextWrap(next ai.ToolHandler) ai.ToolHandler {
 	}
 }
 
-// toolTimeoutWrap gives each tool execution its own deadline while preserving
-// caller cancellation. Handlers still execute synchronously; tools that honor
-// context (custom tools, delegate RPC/A2A, and go-micro RPC clients) return
-// promptly with a bounded error result when the deadline expires.
-func (a *agentImpl) toolTimeoutWrap(next ai.ToolHandler) ai.ToolHandler {
-	return func(ctx context.Context, call ai.ToolCall) ai.ToolResult {
-		if a.opts.ToolTimeout <= 0 {
-			return next(ctx, call)
-		}
-		toolCtx, cancel := context.WithTimeout(ctx, a.opts.ToolTimeout)
-		defer cancel()
-		return next(toolCtx, call)
-	}
-}
-
 // baseHandler executes a tool call: a developer custom tool, the built-in
 // delegate, or an RPC to the service. It is the innermost handler.
 func (a *agentImpl) baseHandler() ai.ToolHandler {
@@ -178,9 +145,6 @@ 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)
 		}
@@ -241,11 +205,6 @@ 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 {
@@ -273,17 +232,6 @@ 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
@@ -335,9 +283,6 @@ func (a *agentImpl) handleDelegate(ctx context.Context, call ai.ToolCall) ai.Too
 		WithRegistry(a.opts.Registry),
 		WithClient(a.opts.Client),
 		WithStore(a.opts.Store),
-		ModelCallTimeout(a.opts.ModelTimeout),
-		ModelRetry(a.opts.ModelMaxAttempts, a.opts.ModelRetryBackoff),
-		ToolCallTimeout(a.opts.ToolTimeout),
 		TraceProvider(a.opts.TraceProvider),
 	)
 	// Record lineage so the sub-agent's tool calls carry this run as parent.

agent/builtin_test.go

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

agent/checkpoint.go

@@ -6,14 +6,12 @@ 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 {
@@ -37,7 +35,6 @@ 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
 }
@@ -52,17 +49,6 @@ 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)
@@ -75,9 +61,6 @@ 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
 	}
@@ -88,9 +71,6 @@ func (a *agentImpl) resume(ctx context.Context, runID string) (*Response, error)
 		}
 		return &resp, nil
 	}
-	if terminalAgentRunStatus(run.Status) {
-		return nil, fmt.Errorf("agent run %s is terminal with status %q", runID, run.Status)
-	}
 	message := string(run.State.Data)
 	parentID := run.ParentID
 	a.mu.Lock()
@@ -98,52 +78,7 @@ func (a *agentImpl) resume(ctx context.Context, runID string) (*Response, error)
 	if a.model == nil {
 		a.setup()
 	}
-	return a.askLocked(ctx, run.ID, message, parentID, &run, false)
-}
-
-// 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, true)
+	return a.askLocked(ctx, run.ID, message, parentID, &run)
 }
 
 func (a *agentImpl) pending(ctx context.Context) ([]flow.Run, error) {
@@ -156,75 +91,9 @@ func (a *agentImpl) pending(ctx context.Context) ([]flow.Run, error) {
 	}
 	out := runs[:0]
 	for _, run := range runs {
-		if run.Flow == a.opts.Name && !terminalAgentRunStatus(run.Status) {
+		if run.Flow == a.opts.Name && run.Status != "done" {
 			out = append(out, run)
 		}
 	}
 	return out, nil
 }
-
-func terminalAgentRunStatus(status string) bool {
-	switch status {
-	case "done", "canceled", "expired":
-		return true
-	default:
-		return false
-	}
-}
-
-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,8 +2,6 @@ package agent
 
 import (
 	"context"
-	"errors"
-	"strings"
 	"testing"
 
 	"go-micro.dev/v6/ai"
@@ -13,7 +11,7 @@ import (
 
 func TestResumeCompletedCheckpointDoesNotReplayModel(t *testing.T) {
 	ctx := context.Background()
-	cp := flow.StoreCheckpoint(store.NewMemoryStore(), "durable-agent")
+	cp := flow.StoreCheckpoint(store.NewStore(), "durable-agent")
 	calls := 0
 	fakeGen = func(ctx context.Context, opts ai.Options, req *ai.Request) (*ai.Response, error) {
 		calls++
@@ -50,127 +48,9 @@ func TestResumeCompletedCheckpointDoesNotReplayModel(t *testing.T) {
 	}
 }
 
-func TestResumeFailedCheckpointDoesNotReplayCompletedTool(t *testing.T) {
-	ctx := context.Background()
-	cp := flow.StoreCheckpoint(store.NewMemoryStore(), "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 TestResumeFailedCheckpointDoesNotDuplicateCompactedMemory(t *testing.T) {
-	ctx := context.Background()
-	st := store.NewMemoryStore()
-	cp := flow.StoreCheckpoint(st, "memory-resume-agent")
-	failRetry := true
-	var sawRecall bool
-	fakeGen = func(ctx context.Context, opts ai.Options, req *ai.Request) (*ai.Response, error) {
-		for _, msg := range req.Messages {
-			if text, ok := msg.Content.(string); ok && strings.Contains(text, "alpha code is 42") {
-				sawRecall = true
-			}
-		}
-		if strings.Contains(req.Prompt, "use alpha code") && failRetry {
-			failRetry = false
-			return nil, errors.New("model connection dropped")
-		}
-		return &ai.Response{Reply: "ok"}, nil
-	}
-	defer func() { fakeGen = nil }()
-
-	a := newTestAgent(Name("memory-resume-agent"), WithStore(st), WithCheckpoint(cp), CompactMemory(4, 1), MemoryRecallLimit(2))
-	for _, msg := range []string{"alpha code is 42", "beta note", "gamma note"} {
-		if _, err := a.Ask(ctx, msg); err != nil {
-			t.Fatalf("Ask(%q): %v", msg, err)
-		}
-	}
-
-	_, err := a.Ask(ctx, "use alpha code now")
-	if err == nil {
-		t.Fatal("Ask succeeded, want simulated provider failure")
-	}
-	if got := countMemoryContent(a.mem.Messages(), "use alpha code now"); got != 1 {
-		t.Fatalf("failed Ask stored prompt %d times, want 1", got)
-	}
-
-	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))
-	}
-	if _, err := Resume(ctx, a, runs[0].ID); err != nil {
-		t.Fatalf("Resume: %v", err)
-	}
-	if got := countMemoryContent(a.mem.Messages(), "use alpha code now"); got != 1 {
-		t.Fatalf("resumed failed Ask stored prompt %d times, want no duplicate", got)
-	}
-	if !sawRecall {
-		t.Fatal("resume did not retrieve archived compacted memory")
-	}
-	if got := len(a.mem.Messages()); got > 4 {
-		t.Fatalf("compacted memory retained %d messages after resume, want <= 4", got)
-	}
-}
-
-func countMemoryContent(messages []ai.Message, needle string) int {
-	var count int
-	for _, msg := range messages {
-		if text, ok := msg.Content.(string); ok && strings.Contains(text, needle) {
-			count++
-		}
-	}
-	return count
-}
-
 func TestPendingReturnsUnfinishedAgentRuns(t *testing.T) {
 	ctx := context.Background()
-	cp := flow.StoreCheckpoint(store.NewMemoryStore(), "pending-agent")
+	cp := flow.StoreCheckpoint(store.NewStore(), "pending-agent")
 	run := flow.Run{ID: "run-1", Flow: "pending-agent", Status: "failed", State: flow.State{Stage: agentAskStep, Data: []byte("retry me")}}
 	if err := cp.Save(ctx, run); err != nil {
 		t.Fatalf("Save: %v", err)
@@ -185,141 +65,9 @@ func TestPendingReturnsUnfinishedAgentRuns(t *testing.T) {
 	}
 }
 
-func TestPendingSkipsTerminalCanceledAndExpiredAgentRuns(t *testing.T) {
-	ctx := context.Background()
-	cp := flow.StoreCheckpoint(store.NewMemoryStore(), "terminal-agent")
-	for _, run := range []flow.Run{
-		{ID: "active", Flow: "terminal-agent", Status: "failed", State: flow.State{Stage: agentAskStep, Data: []byte("retry me")}},
-		{ID: "done", Flow: "terminal-agent", Status: "done", State: flow.State{Stage: agentAskStep, Data: []byte("done")}},
-		{ID: "canceled", Flow: "terminal-agent", Status: "canceled", State: flow.State{Stage: agentAskStep, Data: []byte("canceled")}},
-		{ID: "expired", Flow: "terminal-agent", Status: "expired", State: flow.State{Stage: agentAskStep, Data: []byte("expired")}},
-	} {
-		if err := cp.Save(ctx, run); err != nil {
-			t.Fatalf("Save(%s): %v", run.ID, err)
-		}
-	}
-
-	a := newTestAgent(Name("terminal-agent"), WithCheckpoint(cp))
-	runs, err := Pending(ctx, a)
-	if err != nil {
-		t.Fatalf("Pending: %v", err)
-	}
-	if len(runs) != 1 || runs[0].ID != "active" {
-		t.Fatalf("Pending = %#v, want only active failed run", runs)
-	}
-	for _, id := range []string{"canceled", "expired"} {
-		if _, err := Resume(ctx, a, id); err == nil || !strings.Contains(err.Error(), "terminal") {
-			t.Fatalf("Resume(%s) err = %v, want terminal status error", id, err)
-		}
-	}
-}
-
-func TestHumanInputPauseResumesSameRunWithInput(t *testing.T) {
-	ctx := context.Background()
-	cp := flow.StoreCheckpoint(store.NewMemoryStore(), "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 TestHumanInputResumeHonorsCanceledContextAndLeavesRunPending(t *testing.T) {
-	ctx := context.Background()
-	cp := flow.StoreCheckpoint(store.NewMemoryStore(), "input-cancel-agent")
-	fakeGen = func(ctx context.Context, opts ai.Options, req *ai.Request) (*ai.Response, error) {
-		if opts.ToolHandler != nil {
-			opts.ToolHandler(ctx, ai.ToolCall{ID: "input-1", Name: toolHumanInput, Input: map[string]any{"prompt": "Approve deploy?"}})
-		}
-		return &ai.Response{Reply: "waiting"}, nil
-	}
-	defer func() { fakeGen = nil }()
-
-	a := newTestAgent(Name("input-cancel-agent"), WithCheckpoint(cp))
-	if _, err := a.Ask(ctx, "deploy the service"); 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 {
-		t.Fatalf("Pending returned %d runs, want 1: %#v", len(runs), runs)
-	}
-
-	canceled, cancel := context.WithCancel(ctx)
-	cancel()
-	if _, err := ResumeInput(canceled, a, runs[0].ID, "yes"); !errors.Is(err, context.Canceled) {
-		t.Fatalf("ResumeInput canceled err = %v, want context.Canceled", err)
-	}
-
-	loaded, ok, err := cp.Load(ctx, runs[0].ID)
-	if err != nil || !ok {
-		t.Fatalf("Load paused run ok=%v err=%v", ok, err)
-	}
-	if loaded.Status != "paused" || loaded.State.Stage != agentInputStep {
-		t.Fatalf("run status/stage after canceled resume = %s/%s, want paused/%s", loaded.Status, loaded.State.Stage, agentInputStep)
-	}
-	var pause inputPause
-	if err := loaded.State.Scan(&pause); err != nil {
-		t.Fatalf("Scan pause after canceled resume: %v", err)
-	}
-	if pause.OriginalMessage != "deploy the service" || pause.Prompt != "Approve deploy?" {
-		t.Fatalf("pause after canceled resume = %#v", pause)
-	}
-}
-
 func TestApprovalDenialPausesCheckpointedRunAndResumeContinues(t *testing.T) {
 	ctx := context.Background()
-	cp := flow.StoreCheckpoint(store.NewMemoryStore(), "approval-agent")
+	cp := flow.StoreCheckpoint(store.NewStore(), "approval-agent")
 	calls := 0
 	fakeGen = func(ctx context.Context, opts ai.Options, req *ai.Request) (*ai.Response, error) {
 		calls++

agent/conformance_test.go

@@ -33,29 +33,14 @@ func TestAgentProviderConformanceMatrix(t *testing.T) {
 		{name: "together", key: "TOGETHER_API_KEY", model: "GO_MICRO_CONFORMANCE_TOGETHER_MODEL", live: true},
 	}
 
-	selected := selectedConformanceProviders(os.Getenv("GO_MICRO_AGENT_CONFORMANCE_PROVIDERS"))
 	for _, provider := range providers {
 		provider := provider
-		if len(selected) > 0 && !selected[provider.name] {
-			continue
-		}
 		t.Run(provider.name, func(t *testing.T) {
 			runAgentConformanceScenario(t, provider)
 		})
 	}
 }
 
-func selectedConformanceProviders(csv string) map[string]bool {
-	out := map[string]bool{}
-	for _, part := range strings.Split(csv, ",") {
-		part = strings.TrimSpace(part)
-		if part != "" {
-			out[part] = true
-		}
-	}
-	return out
-}
-
 func runAgentConformanceScenario(t *testing.T, provider conformanceProvider) {
 	t.Helper()
 	if provider.live {

agent/memory.go

@@ -3,7 +3,6 @@ package agent
 import (
 	"encoding/json"
 	"fmt"
-	"sort"
 	"strings"
 	"sync"
 
@@ -124,31 +123,17 @@ func (m *storeMemory) Recall(query string, limit int) []ai.Message {
 		limit = 5
 	}
 	terms := recallTerms(query)
-	type match struct {
-		msg   ai.Message
-		score int
-		index int
-	}
-	matches := make([]match, 0, len(m.archive))
-	for i := len(m.archive) - 1; i >= 0; i-- {
+	var out []ai.Message
+	for i := len(m.archive) - 1; i >= 0 && len(out) < limit; i-- {
 		msg := m.archive[i]
-		if score := recallScore(msg, terms); score > 0 {
-			matches = append(matches, match{msg: msg, score: score, index: i})
+		text := strings.ToLower(fmt.Sprint(msg.Content))
+		for _, term := range terms {
+			if strings.Contains(text, term) {
+				out = append(out, msg)
+				break
+			}
 		}
 	}
-	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
 }
 
@@ -243,17 +228,6 @@ 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,46 +62,6 @@ func TestWithMemoryUsed(t *testing.T) {
 	}
 }
 
-func TestCompactingMemoryRecallRanksSpecificMatches(t *testing.T) {
-	m := NewCompactingMemory(store.NewMemoryStore(), "agent/rank/history", 3, 1).(MemoryRecall)
-	writer := m.(Memory)
-	writer.Add("user", "alpha budget is 42")
-	writer.Add("assistant", "noted")
-	writer.Add("user", "beta budget is 7")
-	writer.Add("assistant", "noted")
-	writer.Add("user", "alpha owner is sam")
-
-	recalled := m.Recall("alpha budget", 2)
-	if len(recalled) == 0 {
-		t.Fatal("expected recalled messages")
-	}
-	if got := recalled[0].Content.(string); !strings.Contains(got, "alpha budget is 42") {
-		t.Fatalf("top recall = %q, want alpha budget match", got)
-	}
-}
-
-func TestCompactingMemoryArchivePersistsAndReloads(t *testing.T) {
-	st := store.NewMemoryStore()
-	m := NewCompactingMemory(st, "agent/reload/history", 3, 1)
-	m.Add("user", "alpha budget is 42")
-	m.Add("assistant", "noted")
-	m.Add("user", "beta budget is 7")
-	m.Add("assistant", "noted")
-
-	reloaded := NewCompactingMemory(st, "agent/reload/history", 3, 1)
-	recall, ok := reloaded.(MemoryRecall)
-	if !ok {
-		t.Fatal("compacting memory should support recall")
-	}
-	recalled := recall.Recall("alpha budget", 1)
-	if len(recalled) != 1 {
-		t.Fatalf("recalled %d messages, want 1", len(recalled))
-	}
-	if got := recalled[0].Content.(string); !strings.Contains(got, "alpha budget is 42") {
-		t.Fatalf("reloaded recall = %q, want alpha budget", got)
-	}
-}
-
 // A custom tool is offered to the model and dispatched to its handler.
 func TestWithToolExposedAndDispatched(t *testing.T) {
 	var got map[string]any

agent/options.go

@@ -56,10 +56,6 @@ type Options struct {
 	// ModelRetryBackoff is the base delay between transient provider failures
 	// (grows exponentially per attempt when retries are enabled).
 	ModelRetryBackoff 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.
-	ToolTimeout time.Duration
 
 	// Memory is the agent's conversation memory. Nil = the default
 	// store-backed memory (durable across restarts).
@@ -95,11 +91,6 @@ type Options struct {
 	// and tool calls. Nil disables instrumentation.
 	TraceProvider trace.TracerProvider
 
-	// TraceInputs controls whether agent observability records include raw
-	// user messages. It is false by default so spans and persisted run
-	// timelines carry correlation and shape without leaking prompts.
-	TraceInputs bool
-
 	// tools are developer-registered custom tools (see WithTool).
 	tools []customTool
 	// wrappers are developer-registered tool-execution wrappers
@@ -116,7 +107,6 @@ func newOptions(opts ...Option) Options {
 		ModelTimeout:      30 * time.Second,
 		ModelMaxAttempts:  1, // retries opt-in via ModelRetry (see field doc)
 		ModelRetryBackoff: 100 * time.Millisecond,
-		ToolTimeout:       30 * time.Second,
 		// On by default and lenient: identical repeated calls are a
 		// no-progress loop, never useful. Set LoopLimit(0) to disable.
 		LoopLimit: 3,
@@ -209,14 +199,6 @@ func ModelCallTimeout(d time.Duration) Option {
 	return func(o *Options) { o.ModelTimeout = d }
 }
 
-// ToolCallTimeout sets the timeout for each tool execution. It bounds custom
-// tools, built-in delegate calls, and service RPC tools with the same context
-// deadline so mid-run cancellation and slow tools produce safe error results
-// instead of unbounded agent runs. Set 0 to disable.
-func ToolCallTimeout(d time.Duration) Option {
-	return func(o *Options) { o.ToolTimeout = d }
-}
-
 // ModelRetry sets the provider retry budget and backoff for transient failures.
 func ModelRetry(maxAttempts int, backoff time.Duration) Option {
 	return func(o *Options) {
@@ -313,11 +295,3 @@ func WithTool(name, description string, properties map[string]any, handler ToolF
 func TraceProvider(tp trace.TracerProvider) Option {
 	return func(o *Options) { o.TraceProvider = tp }
 }
-
-// TraceInputs opts in to recording raw user messages on agent run events.
-// By default inputs are redacted from OpenTelemetry spans and persisted run
-// timelines; use this only when the observability backend is approved to store
-// prompt content.
-func TraceInputs(enabled bool) Option {
-	return func(o *Options) { o.TraceInputs = enabled }
-}

agent/otel.go

@@ -31,35 +31,27 @@ const (
 	AttrInputTokens    = "agent.tokens.input"
 	AttrOutputTokens   = "agent.tokens.output"
 	AttrTotalTokens    = "agent.tokens.total"
-	AttrAttempt        = "agent.model.attempt"
-	AttrMaxAttempts    = "agent.model.max_attempts"
 	AttrToolName       = "agent.tool.name"
 	AttrDelegate       = "agent.delegate"
 	AttrGuardrailBlock = "agent.guardrail.block"
 	AttrRefusal        = "agent.refusal"
-	AttrInputChars     = "agent.input.chars"
-	AttrErrorKind      = "agent.error.kind"
 )
 
 type RunEvent struct {
-	Time        time.Time `json:"time"`
-	RunID       string    `json:"run_id"`
-	ParentID    string    `json:"parent_id,omitempty"`
-	TraceID     string    `json:"trace_id,omitempty"`
-	SpanID      string    `json:"span_id,omitempty"`
-	Agent       string    `json:"agent"`
-	Kind        string    `json:"kind"`
-	Name        string    `json:"name,omitempty"`
-	Provider    string    `json:"provider,omitempty"`
-	Model       string    `json:"model,omitempty"`
-	Attempt     int       `json:"attempt,omitempty"`
-	MaxAttempts int       `json:"max_attempts,omitempty"`
-	LatencyMS   int64     `json:"latency_ms,omitempty"`
-	Tokens      Usage     `json:"tokens,omitempty"`
-	Refused     string    `json:"refused,omitempty"`
-	Error       string    `json:"error,omitempty"`
-	ErrorKind   string    `json:"error_kind,omitempty"`
-	InputChars  int       `json:"input_chars,omitempty"`
+	Time      time.Time `json:"time"`
+	RunID     string    `json:"run_id"`
+	ParentID  string    `json:"parent_id,omitempty"`
+	TraceID   string    `json:"trace_id,omitempty"`
+	SpanID    string    `json:"span_id,omitempty"`
+	Agent     string    `json:"agent"`
+	Kind      string    `json:"kind"`
+	Name      string    `json:"name,omitempty"`
+	Provider  string    `json:"provider,omitempty"`
+	Model     string    `json:"model,omitempty"`
+	LatencyMS int64     `json:"latency_ms,omitempty"`
+	Tokens    Usage     `json:"tokens,omitempty"`
+	Refused   string    `json:"refused,omitempty"`
+	Error     string    `json:"error,omitempty"`
 }
 
 type Usage = ai.Usage
@@ -102,17 +94,13 @@ func (a *agentImpl) tracer() trace.Tracer {
 func (a *agentImpl) startRun(ctx context.Context, message string) (context.Context, func(error)) {
 	info, _ := ai.RunInfoFrom(ctx)
 	start := time.Now()
-	runEvent := RunEvent{Time: start, RunID: info.RunID, ParentID: info.ParentID, Agent: info.Agent, Kind: "run", InputChars: len(message)}
-	if a.opts.TraceInputs {
-		runEvent.Name = message
-	}
 
 	if a.opts.TraceProvider == nil {
-		a.recordRunEvent(runEvent)
+		a.recordRunEvent(RunEvent{Time: start, RunID: info.RunID, ParentID: info.ParentID, Agent: info.Agent, Kind: "run", Name: message})
 		return ctx, func(err error) {
 			latency := time.Since(start).Milliseconds()
 			if err != nil {
-				a.recordRunEvent(RunEvent{Time: time.Now(), RunID: info.RunID, ParentID: info.ParentID, Agent: info.Agent, Kind: "error", LatencyMS: latency, Error: err.Error(), ErrorKind: string(ai.ClassifyError(err))})
+				a.recordRunEvent(RunEvent{Time: time.Now(), RunID: info.RunID, ParentID: info.ParentID, Agent: info.Agent, Kind: "error", LatencyMS: latency, Error: err.Error()})
 				return
 			}
 			a.recordRunEvent(RunEvent{Time: time.Now(), RunID: info.RunID, ParentID: info.ParentID, Agent: info.Agent, Kind: "done", LatencyMS: latency})
@@ -121,15 +109,14 @@ func (a *agentImpl) startRun(ctx context.Context, message string) (context.Conte
 
 	ctx, span := a.tracer().Start(ctx, spanNameRun, trace.WithSpanKind(trace.SpanKindInternal), trace.WithAttributes(
 		attribute.String(AttrRunID, info.RunID), attribute.String(AttrParentRunID, info.ParentID), attribute.String(AttrAgentName, info.Agent)))
-	a.recordSpanEvent(span, runEvent)
+	a.recordSpanEvent(span, RunEvent{Time: start, RunID: info.RunID, ParentID: info.ParentID, Agent: info.Agent, Kind: "run", Name: message})
 	return ctx, func(err error) {
 		latency := time.Since(start).Milliseconds()
 		span.SetAttributes(attribute.Int64(AttrLatencyMS, latency))
 		if err != nil {
-			span.SetAttributes(attribute.String(AttrErrorKind, string(ai.ClassifyError(err))))
 			span.RecordError(err)
 			span.SetStatus(codes.Error, err.Error())
-			a.recordSpanEvent(span, RunEvent{Time: time.Now(), RunID: info.RunID, ParentID: info.ParentID, Agent: info.Agent, Kind: "error", LatencyMS: latency, Error: err.Error(), ErrorKind: string(ai.ClassifyError(err))})
+			a.recordSpanEvent(span, RunEvent{Time: time.Now(), RunID: info.RunID, ParentID: info.ParentID, Agent: info.Agent, Kind: "error", LatencyMS: latency, Error: err.Error()})
 		} else {
 			span.SetStatus(codes.Ok, "")
 			a.recordSpanEvent(span, RunEvent{Time: time.Now(), RunID: info.RunID, ParentID: info.ParentID, Agent: info.Agent, Kind: "done", LatencyMS: latency})
@@ -157,10 +144,9 @@ func (m *tracedModel) Generate(ctx context.Context, req *ai.Request, opts ...ai.
 		if resp != nil {
 			usage = resp.Usage
 		}
-		e := RunEvent{Time: time.Now(), RunID: info.RunID, ParentID: info.ParentID, Agent: info.Agent, Kind: "model", Provider: provider, Model: model, Attempt: info.Attempt, MaxAttempts: info.MaxAttempts, LatencyMS: dur, Tokens: usage}
+		e := RunEvent{Time: time.Now(), RunID: info.RunID, ParentID: info.ParentID, Agent: info.Agent, Kind: "model", Provider: provider, Model: model, LatencyMS: dur, Tokens: usage}
 		if err != nil {
 			e.Error = err.Error()
-			e.ErrorKind = string(ai.ClassifyError(err))
 		}
 		m.a.recordRunEvent(e)
 		return resp, err
@@ -176,12 +162,6 @@ func (m *tracedModel) Generate(ctx context.Context, req *ai.Request, opts ...ai.
 	resp, err := m.Model.Generate(ctx, req, opts...)
 	dur := time.Since(start).Milliseconds()
 	attrs := []attribute.KeyValue{attribute.Int64(AttrLatencyMS, dur)}
-	if info.Attempt > 0 {
-		attrs = append(attrs, attribute.Int(AttrAttempt, info.Attempt))
-	}
-	if info.MaxAttempts > 0 {
-		attrs = append(attrs, attribute.Int(AttrMaxAttempts, info.MaxAttempts))
-	}
 	usage := ai.Usage{}
 	if resp != nil {
 		usage = resp.Usage
@@ -189,17 +169,15 @@ func (m *tracedModel) Generate(ctx context.Context, req *ai.Request, opts ...ai.
 	}
 	span.SetAttributes(attrs...)
 	if err != nil {
-		span.SetAttributes(attribute.String(AttrErrorKind, string(ai.ClassifyError(err))))
 		span.RecordError(err)
 		span.SetStatus(codes.Error, err.Error())
 	} else {
 		span.SetStatus(codes.Ok, "")
 	}
 	span.End()
-	e := RunEvent{Time: time.Now(), RunID: info.RunID, ParentID: info.ParentID, Agent: info.Agent, Kind: "model", Provider: provider, Model: model, Attempt: info.Attempt, MaxAttempts: info.MaxAttempts, LatencyMS: dur, Tokens: usage}
+	e := RunEvent{Time: time.Now(), RunID: info.RunID, ParentID: info.ParentID, Agent: info.Agent, Kind: "model", Provider: provider, Model: model, LatencyMS: dur, Tokens: usage}
 	if err != nil {
 		e.Error = err.Error()
-		e.ErrorKind = string(ai.ClassifyError(err))
 	}
 	m.a.recordSpanEvent(span, e)
 	return resp, err
@@ -226,8 +204,7 @@ func (a *agentImpl) traceTool(next ai.ToolHandler) ai.ToolHandler {
 		if a.opts.TraceProvider == nil {
 			res := next(ctx, call)
 			dur := time.Since(start).Milliseconds()
-			resErr := resultError(res)
-			a.recordRunEvent(RunEvent{Time: time.Now(), RunID: info.RunID, ParentID: info.ParentID, Agent: info.Agent, Kind: "tool", Name: call.Name, LatencyMS: dur, Refused: res.Refused, Error: resErr, ErrorKind: classifyToolError(resErr)})
+			a.recordRunEvent(RunEvent{Time: time.Now(), RunID: info.RunID, ParentID: info.ParentID, Agent: info.Agent, Kind: "tool", Name: call.Name, LatencyMS: dur, Refused: res.Refused, Error: resultError(res)})
 			return res
 		}
 
@@ -244,11 +221,8 @@ func (a *agentImpl) traceTool(next ai.ToolHandler) ai.ToolHandler {
 		if res.Refused != "" {
 			attrs = append(attrs, attribute.Bool(AttrGuardrailBlock, true), attribute.String(AttrRefusal, res.Refused))
 		}
-		resErr := resultError(res)
-		if kind := classifyToolError(resErr); kind != "" {
-			attrs = append(attrs, attribute.String(AttrErrorKind, kind))
-		}
 		span.SetAttributes(attrs...)
+		resErr := resultError(res)
 		if res.Refused != "" {
 			span.SetStatus(codes.Error, res.Refused)
 		} else if resErr != "" {
@@ -257,7 +231,7 @@ func (a *agentImpl) traceTool(next ai.ToolHandler) ai.ToolHandler {
 			span.SetStatus(codes.Ok, "")
 		}
 		span.End()
-		a.recordSpanEvent(span, RunEvent{Time: time.Now(), RunID: info.RunID, ParentID: info.ParentID, Agent: info.Agent, Kind: "tool", Name: call.Name, LatencyMS: dur, Refused: res.Refused, Error: resErr, ErrorKind: classifyToolError(resErr)})
+		a.recordSpanEvent(span, RunEvent{Time: time.Now(), RunID: info.RunID, ParentID: info.ParentID, Agent: info.Agent, Kind: "tool", Name: call.Name, LatencyMS: dur, Refused: res.Refused, Error: resErr})
 		return res
 	}
 }
@@ -274,70 +248,14 @@ func resultError(res ai.ToolResult) string {
 	return ""
 }
 
-func classifyToolError(err string) string {
-	switch {
-	case err == "":
-		return ""
-	case strings.Contains(strings.ToLower(err), "context canceled"):
-		return string(ai.ErrorKindCanceled)
-	case strings.Contains(strings.ToLower(err), "deadline exceeded"):
-		return string(ai.ErrorKindTimeout)
-	default:
-		return string(ai.ErrorKindProvider)
-	}
-}
-
 func (a *agentImpl) recordSpanEvent(span trace.Span, e RunEvent) {
 	if sc := span.SpanContext(); sc.IsValid() {
 		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.Attempt > 0 {
-		attrs = append(attrs, attribute.Int(AttrAttempt, e.Attempt))
-	}
-	if e.MaxAttempts > 0 {
-		attrs = append(attrs, attribute.Int(AttrMaxAttempts, e.MaxAttempts))
-	}
-	if e.LatencyMS > 0 {
-		attrs = append(attrs, attribute.Int64(AttrLatencyMS, e.LatencyMS))
-	}
-	if e.InputChars > 0 {
-		attrs = append(attrs, attribute.Int(AttrInputChars, e.InputChars))
-	}
-	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))
-	}
-	if e.ErrorKind != "" {
-		attrs = append(attrs, attribute.String(AttrErrorKind, e.ErrorKind))
-	}
-	return attrs
-}
-
 func (a *agentImpl) recordRunEvent(e RunEvent) {
 	if e.RunID == "" {
 		return

agent/otel_test.go

@@ -3,7 +3,6 @@ package agent
 import (
 	"context"
 	"encoding/json"
-	"errors"
 	"fmt"
 	"strings"
 	"testing"
@@ -12,13 +11,10 @@ import (
 	"go-micro.dev/v6/ai"
 	"go-micro.dev/v6/store"
 	"go.opentelemetry.io/otel/attribute"
-	"go.opentelemetry.io/otel/codes"
 	"go.opentelemetry.io/otel/sdk/trace"
 	"go.opentelemetry.io/otel/sdk/trace/tracetest"
 )
 
-const codesError = codes.Error
-
 type otelTestModel struct{ opts ai.Options }
 
 func (m *otelTestModel) Init(opts ...ai.Option) error {
@@ -75,16 +71,6 @@ 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
@@ -93,9 +79,6 @@ func TestAgentOpenTelemetrySpans(t *testing.T) {
 		if attrs[AttrRunID] != runID || attrs[AttrAgentName] != "runner" {
 			t.Fatalf("%s missing run correlation attributes: %#v", s.Name(), attrs)
 		}
-		if s.Name() == spanNameModelCall && (attrs[AttrAttempt] != "1" || attrs[AttrMaxAttempts] != "1") {
-			t.Fatalf("model span missing attempt attributes: %#v", attrs)
-		}
 	}
 	keys, err := store.Scope(st, "agent", "runner").List(store.ListPrefix("runs/"))
 	if err != nil {
@@ -132,159 +115,6 @@ func TestAgentOpenTelemetrySpans(t *testing.T) {
 	}
 }
 
-func TestAgentRunObservabilityRedactsInputByDefault(t *testing.T) {
-	secret := "deploy production with token sk-secret"
-	exp := tracetest.NewInMemoryExporter()
-	tp := trace.NewTracerProvider(trace.WithSyncer(exp))
-	st := store.NewMemoryStore()
-	a := New(Name("redactor"), Provider("oteltest"), WithStore(st), TraceProvider(tp))
-	if _, err := a.Ask(context.Background(), secret); err != nil {
-		t.Fatal(err)
-	}
-
-	spans := exp.GetSpans().Snapshots()
-	var sawInputChars bool
-	for _, s := range spans {
-		for _, event := range s.Events() {
-			attrs := spanAttributes(event.Attributes)
-			if attrs["agent.event.name"] == secret {
-				t.Fatalf("span event leaked raw input: %#v", event)
-			}
-			if attrs[AttrInputChars] == fmt.Sprint(len(secret)) {
-				sawInputChars = true
-			}
-		}
-	}
-	if !sawInputChars {
-		t.Fatal("run event missing redacted input length attribute")
-	}
-
-	summaries, err := ListRunSummaries(st, "redactor")
-	if err != nil {
-		t.Fatal(err)
-	}
-	events, err := LoadRunEvents(st, "redactor", summaries[0].RunID)
-	if err != nil {
-		t.Fatal(err)
-	}
-	for _, event := range events {
-		if event.Name == secret {
-			t.Fatalf("persisted run event leaked raw input: %#v", event)
-		}
-		if event.Kind == "run" && event.InputChars != len(secret) {
-			t.Fatalf("run event InputChars = %d, want %d", event.InputChars, len(secret))
-		}
-	}
-}
-
-func TestAgentTraceInputsOptInRecordsInput(t *testing.T) {
-	message := "operator-approved diagnostic prompt"
-	st := store.NewMemoryStore()
-	a := New(Name("input-opt-in"), Provider("oteltest"), WithStore(st), TraceInputs(true))
-	if _, err := a.Ask(context.Background(), message); err != nil {
-		t.Fatal(err)
-	}
-
-	summaries, err := ListRunSummaries(st, "input-opt-in")
-	if err != nil {
-		t.Fatal(err)
-	}
-	events, err := LoadRunEvents(st, "input-opt-in", summaries[0].RunID)
-	if err != nil {
-		t.Fatal(err)
-	}
-	for _, event := range events {
-		if event.Kind == "run" && event.Name == message {
-			return
-		}
-	}
-	t.Fatalf("opt-in run event did not record message: %#v", events)
-}
-
-type failingOtelModel struct{ opts ai.Options }
-
-func (m *failingOtelModel) Init(opts ...ai.Option) error {
-	for _, o := range opts {
-		o(&m.opts)
-	}
-	return nil
-}
-func (m *failingOtelModel) Options() ai.Options { return m.opts }
-func (m *failingOtelModel) String() string      { return "otelfail" }
-func (m *failingOtelModel) Stream(context.Context, *ai.Request, ...ai.GenerateOption) (ai.Stream, error) {
-	return nil, nil
-}
-func (m *failingOtelModel) Generate(context.Context, *ai.Request, ...ai.GenerateOption) (*ai.Response, error) {
-	return nil, errors.New("provider exploded")
-}
-
-func init() {
-	ai.Register("otelfail", func(opts ...ai.Option) ai.Model { return &failingOtelModel{opts: ai.NewOptions(opts...)} })
-}
-
-func TestAgentOpenTelemetrySpansModelFailure(t *testing.T) {
-	exp := tracetest.NewInMemoryExporter()
-	tp := trace.NewTracerProvider(trace.WithSyncer(exp))
-	st := store.NewMemoryStore()
-	a := New(Name("failing-runner"), Provider("otelfail"), WithStore(st), TraceProvider(tp))
-	if _, err := a.Ask(context.Background(), "hello"); err == nil {
-		t.Fatal("Ask succeeded, want provider error")
-	}
-
-	spans := exp.GetSpans().Snapshots()
-	var sawRunError, sawModelError bool
-	for _, s := range spans {
-		attrs := spanAttributes(s.Attributes())
-		switch s.Name() {
-		case spanNameRun:
-			if attrs[AttrAgentName] == "failing-runner" && s.Status().Code == codesError {
-				sawRunError = true
-			}
-		case spanNameModelCall:
-			if attrs[AttrAgentName] == "failing-runner" && attrs[AttrAttempt] == "1" && attrs[AttrErrorKind] == string(ai.ErrorKindUnknown) && s.Status().Code == codesError {
-				sawModelError = true
-			}
-		}
-	}
-	if !sawRunError || !sawModelError {
-		t.Fatalf("missing error spans: run=%v model=%v spans=%d", sawRunError, sawModelError, len(spans))
-	}
-
-	summaries, err := ListRunSummaries(st, "failing-runner")
-	if err != nil {
-		t.Fatal(err)
-	}
-	if len(summaries) != 1 || summaries[0].Status != "error" || summaries[0].LastError == "" {
-		t.Fatalf("unexpected failure summary: %#v", summaries)
-	}
-	events, err := LoadRunEvents(st, "failing-runner", summaries[0].RunID)
-	if err != nil {
-		t.Fatal(err)
-	}
-	var sawModelEvent bool
-	for _, event := range events {
-		if event.Kind == "model" && event.Attempt == 1 && event.MaxAttempts == 1 && event.Error != "" && event.ErrorKind == string(ai.ErrorKindUnknown) {
-			sawModelEvent = true
-		}
-	}
-	if !sawModelEvent {
-		t.Fatalf("missing failed model event with attempt metadata: %#v", events)
-	}
-}
-
-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 {

agent/resilience_test.go

@@ -102,30 +102,3 @@ func TestCanceledAskContextSkipsToolExecution(t *testing.T) {
 		t.Fatalf("plan persisted after canceled tool context: %q", plan)
 	}
 }
-
-func TestToolCallTimeoutPropagatesDeadlineToCustomTool(t *testing.T) {
-	var sawDeadline bool
-	a := newTestAgent(
-		Name("tool-timeout"),
-		ToolCallTimeout(10*time.Millisecond),
-		WithTool("slow", "slow tool", nil, func(ctx context.Context, input map[string]any) (string, error) {
-			if _, ok := ctx.Deadline(); ok {
-				sawDeadline = true
-			}
-			<-ctx.Done()
-			return "", ctx.Err()
-		}),
-	)
-
-	start := time.Now()
-	content := toolContent(a.toolHandler(), "slow", nil)
-	if !sawDeadline {
-		t.Fatal("custom tool did not receive a deadline")
-	}
-	if !strings.Contains(content, context.DeadlineExceeded.Error()) {
-		t.Fatalf("tool result = %q, want deadline exceeded", content)
-	}
-	if elapsed := time.Since(start); elapsed > 200*time.Millisecond {
-		t.Fatalf("tool call took %s, want bounded timeout", elapsed)
-	}
-}

agent/stream.go

@@ -1,240 +0,0 @@
-package agent
-
-import (
-	"context"
-	"encoding/json"
-	"errors"
-	"io"
-	"strings"
-	"sync"
-
-	"github.com/google/uuid"
-
-	"go-micro.dev/v6/ai"
-)
-
-// StreamEventType identifies an event emitted by a tool-aware agent stream.
-type StreamEventType string
-
-const (
-	// StreamEventToolStart is emitted immediately before a tool call runs.
-	StreamEventToolStart StreamEventType = "tool_start"
-	// StreamEventToolEnd is emitted after a tool call returns or is refused.
-	StreamEventToolEnd StreamEventType = "tool_end"
-	// StreamEventToken carries a chunk of the final answer.
-	StreamEventToken StreamEventType = "token"
-	// StreamEventDone carries the completed agent response.
-	StreamEventDone StreamEventType = "done"
-)
-
-// StreamEvent is one event from StreamAsk.
-type StreamEvent struct {
-	Type     StreamEventType
-	Token    string
-	ToolCall ai.ToolCall
-	Result   ai.ToolResult
-	Response *Response
-}
-
-// AgentStream is a stream of tool execution events followed by final-answer chunks.
-type AgentStream interface {
-	Recv() (*StreamEvent, error)
-	Close() error
-}
-
-// StreamAsk runs an agent Ask turn with tool start/end events and streams the final answer.
-// It is additive for callers that hold the public Agent interface; concrete agents also
-// expose the same method directly.
-func StreamAsk(ctx context.Context, ag Agent, message string) (AgentStream, error) {
-	streamer, ok := ag.(interface {
-		StreamAsk(context.Context, string) (AgentStream, error)
-	})
-	if !ok {
-		return nil, errors.New("agent: StreamAsk unsupported by implementation")
-	}
-	return streamer.StreamAsk(ctx, message)
-}
-
-// ResumeStreamAsk resumes a checkpointed agent run and emits the same event
-// shape as StreamAsk. Completed runs are streamed from the persisted response;
-// unfinished runs continue from their checkpoint and emit tool events for any
-// work that still needs to run. Tool calls already recorded as done in the
-// checkpoint are reused by the agent checkpoint wrapper and are not re-executed.
-func ResumeStreamAsk(ctx context.Context, ag Agent, runID string) (AgentStream, error) {
-	a, ok := ag.(*agentImpl)
-	if !ok {
-		return nil, errors.New("agent: ResumeStreamAsk unsupported by implementation")
-	}
-	return a.resumeStreamAsk(ctx, runID)
-}
-
-// StreamAsk runs tools like Ask, emits ToolStart/ToolEnd events as they execute,
-// then emits chunks of the final answer followed by a Done event.
-func (a *agentImpl) StreamAsk(ctx context.Context, message string) (AgentStream, error) {
-	events := make(chan *StreamEvent, 16)
-	done := make(chan struct{})
-	s := &agentStream{events: events, done: done}
-
-	go func() {
-		defer close(events)
-		defer close(done)
-		resp, err := a.askWithStreamEvents(ctx, message, events)
-		if err != nil {
-			s.setErr(err)
-			return
-		}
-		for _, tok := range splitStreamTokens(resp.Reply) {
-			if !sendStreamEvent(ctx, events, &StreamEvent{Type: StreamEventToken, Token: tok}) {
-				return
-			}
-		}
-		_ = sendStreamEvent(ctx, events, &StreamEvent{Type: StreamEventDone, Response: resp})
-	}()
-	return s, nil
-}
-
-func (a *agentImpl) resumeStreamAsk(ctx context.Context, runID string) (AgentStream, error) {
-	events := make(chan *StreamEvent, 16)
-	done := make(chan struct{})
-	s := &agentStream{events: events, done: done}
-
-	go func() {
-		defer close(events)
-		defer close(done)
-		resp, err := a.resumeWithStreamEvents(ctx, runID, events)
-		if err != nil {
-			s.setErr(err)
-			return
-		}
-		for _, tok := range splitStreamTokens(resp.Reply) {
-			if !sendStreamEvent(ctx, events, &StreamEvent{Type: StreamEventToken, Token: tok}) {
-				return
-			}
-		}
-		_ = sendStreamEvent(ctx, events, &StreamEvent{Type: StreamEventDone, Response: resp})
-	}()
-	return s, nil
-}
-
-func (a *agentImpl) askWithStreamEvents(ctx context.Context, message string, events chan<- *StreamEvent) (*Response, error) {
-	a.mu.Lock()
-	defer a.mu.Unlock()
-
-	if a.tools == nil {
-		a.tools = ai.NewTools(a.opts.Registry, ai.ToolClient(a.opts.Client))
-	}
-	base := a.toolHandler()
-	handler := func(ctx context.Context, call ai.ToolCall) ai.ToolResult {
-		_ = sendStreamEvent(ctx, events, &StreamEvent{Type: StreamEventToolStart, ToolCall: call})
-		result := base(ctx, call)
-		_ = sendStreamEvent(ctx, events, &StreamEvent{Type: StreamEventToolEnd, ToolCall: call, Result: result})
-		return result
-	}
-	a.setupWithToolHandler(handler)
-	defer a.setupWithToolHandler(nil)
-	return a.askLocked(ctx, uuid.New().String(), message, a.parentRunID, nil, true)
-}
-
-func (a *agentImpl) resumeWithStreamEvents(ctx context.Context, runID string, events chan<- *StreamEvent) (*Response, error) {
-	if a.opts.Checkpoint == nil {
-		return nil, errors.New("agent: ResumeStreamAsk requires a checkpoint")
-	}
-	run, ok, err := a.opts.Checkpoint.Load(ctx, runID)
-	if err != nil {
-		return nil, err
-	}
-	if !ok {
-		return nil, errors.New("agent: checkpointed run not found")
-	}
-	if run.Status == "done" {
-		var resp Response
-		if err := json.Unmarshal(run.State.Data, &resp); err != nil {
-			return nil, err
-		}
-		return &resp, nil
-	}
-	if terminalAgentRunStatus(run.Status) {
-		return nil, errors.New("agent: checkpointed run is terminal with status " + run.Status)
-	}
-
-	a.mu.Lock()
-	defer a.mu.Unlock()
-	if a.tools == nil {
-		a.tools = ai.NewTools(a.opts.Registry, ai.ToolClient(a.opts.Client))
-	}
-	base := a.toolHandler()
-	handler := func(ctx context.Context, call ai.ToolCall) ai.ToolResult {
-		_ = sendStreamEvent(ctx, events, &StreamEvent{Type: StreamEventToolStart, ToolCall: call})
-		result := base(ctx, call)
-		_ = sendStreamEvent(ctx, events, &StreamEvent{Type: StreamEventToolEnd, ToolCall: call, Result: result})
-		return result
-	}
-	a.setupWithToolHandler(handler)
-	defer a.setupWithToolHandler(nil)
-	if run.Status == "paused" {
-		if run.State.Stage == agentInputStep {
-			return nil, errors.New("agent: checkpointed run is input-required; resume with ResumeInput")
-		}
-		run.Status = "running"
-		run.State.Stage = agentAskStep
-	}
-	return a.askLocked(ctx, run.ID, string(run.State.Data), run.ParentID, &run, false)
-}
-
-type agentStream struct {
-	events <-chan *StreamEvent
-	done   <-chan struct{}
-	mu     sync.Mutex
-	err    error
-}
-
-func (s *agentStream) Recv() (*StreamEvent, error) {
-	ev, ok := <-s.events
-	if ok {
-		return ev, nil
-	}
-	s.mu.Lock()
-	defer s.mu.Unlock()
-	if s.err != nil {
-		return nil, s.err
-	}
-	return nil, io.EOF
-}
-
-func (s *agentStream) Close() error {
-	<-s.done
-	return nil
-}
-
-func (s *agentStream) setErr(err error) {
-	s.mu.Lock()
-	defer s.mu.Unlock()
-	s.err = err
-}
-
-func sendStreamEvent(ctx context.Context, events chan<- *StreamEvent, ev *StreamEvent) bool {
-	select {
-	case events <- ev:
-		return true
-	case <-ctx.Done():
-		return false
-	}
-}
-
-func splitStreamTokens(reply string) []string {
-	if reply == "" {
-		return nil
-	}
-	parts := strings.Fields(reply)
-	if len(parts) == 0 {
-		return []string{reply}
-	}
-	out := make([]string, 0, len(parts))
-	for i, part := range parts {
-		if i > 0 {
-			part = " " + part
-		}
-		out = append(out, part)
-	}
-	return out
-}

agent/stream_test.go

@@ -1,175 +0,0 @@
-package agent
-
-import (
-	"context"
-	"errors"
-	"io"
-	"testing"
-
-	"go-micro.dev/v6/ai"
-	"go-micro.dev/v6/flow"
-	"go-micro.dev/v6/store"
-)
-
-func TestStreamAskEmitsToolEventsAndFinalTokens(t *testing.T) {
-	calls := 0
-	fakeGen = func(ctx context.Context, opts ai.Options, req *ai.Request) (*ai.Response, error) {
-		if opts.ToolHandler == nil {
-			t.Fatal("StreamAsk must configure a tool handler")
-		}
-		calls++
-		result := opts.ToolHandler(ctx, ai.ToolCall{ID: "call-1", Name: "echo", Input: map[string]any{"text": "hello"}})
-		return &ai.Response{
-			Reply:     "planning",
-			Answer:    "final answer",
-			ToolCalls: []ai.ToolCall{{ID: "call-1", Name: "echo", Input: map[string]any{"text": "hello"}, Result: result.Content}},
-		}, nil
-	}
-	defer func() { fakeGen = nil }()
-
-	a := newTestAgent(Name("streamer"), WithTool("echo", "echo text", nil, func(ctx context.Context, input map[string]any) (string, error) {
-		return input["text"].(string), nil
-	}))
-	stream, err := a.StreamAsk(context.Background(), "say hello")
-	if err != nil {
-		t.Fatalf("StreamAsk: %v", err)
-	}
-
-	var types []StreamEventType
-	var tokens string
-	var done *Response
-	for {
-		event, err := stream.Recv()
-		if errors.Is(err, io.EOF) {
-			break
-		}
-		if err != nil {
-			t.Fatalf("Recv: %v", err)
-		}
-		types = append(types, event.Type)
-		if event.Type == StreamEventToken {
-			tokens += event.Token
-		}
-		if event.Type == StreamEventDone {
-			done = event.Response
-		}
-	}
-
-	want := []StreamEventType{StreamEventToolStart, StreamEventToolEnd, StreamEventToken, StreamEventToken, StreamEventToken, StreamEventDone}
-	if len(types) != len(want) {
-		t.Fatalf("event types = %v, want %v", types, want)
-	}
-	for i := range want {
-		if types[i] != want[i] {
-			t.Fatalf("event types = %v, want %v", types, want)
-		}
-	}
-	if tokens != "planning final answer" {
-		t.Fatalf("tokens = %q", tokens)
-	}
-	if done == nil || done.Reply != "planning\n\nfinal answer" {
-		t.Fatalf("done response = %#v", done)
-	}
-	if calls != 1 {
-		t.Fatalf("Generate calls = %d, want 1", calls)
-	}
-}
-
-func TestStreamAskHelperRejectsUnsupportedAgent(t *testing.T) {
-	_, err := StreamAsk(context.Background(), unsupportedAgent{}, "hello")
-	if err == nil {
-		t.Fatal("StreamAsk helper should reject unsupported implementations")
-	}
-}
-
-func TestResumeStreamAskDoesNotReplayCompletedTool(t *testing.T) {
-	ctx := context.Background()
-	cp := flow.StoreCheckpoint(store.NewStore(), "stream-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: "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("stream disconnected after tool")
-		}
-		return &ai.Response{Reply: "finished from streamed checkpoint"}, nil
-	}
-	defer func() { fakeGen = nil }()
-
-	a := newTestAgent(Name("stream-resume-agent"), WithCheckpoint(cp),
-		WithTool("charge", "charge once", nil, func(context.Context, map[string]any) (string, error) {
-			toolRuns++
-			return "charged", nil
-		}))
-	stream, err := a.StreamAsk(ctx, "charge order 42")
-	if err != nil {
-		t.Fatalf("StreamAsk: %v", err)
-	}
-	for {
-		_, err := stream.Recv()
-		if errors.Is(err, io.EOF) {
-			break
-		}
-		if err != nil {
-			break
-		}
-	}
-	if toolRuns != 1 {
-		t.Fatalf("tool executions after failed StreamAsk = %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))
-	}
-
-	resumed, err := ResumeStreamAsk(ctx, a, runs[0].ID)
-	if err != nil {
-		t.Fatalf("ResumeStreamAsk: %v", err)
-	}
-	var toolEvents int
-	var done *Response
-	for {
-		event, err := resumed.Recv()
-		if errors.Is(err, io.EOF) {
-			break
-		}
-		if err != nil {
-			t.Fatalf("resumed Recv: %v", err)
-		}
-		if event.Type == StreamEventToolStart || event.Type == StreamEventToolEnd {
-			toolEvents++
-		}
-		if event.Type == StreamEventDone {
-			done = event.Response
-		}
-	}
-	if toolRuns != 1 {
-		t.Fatalf("tool executions after ResumeStreamAsk = %d, want completed tool was not replayed", toolRuns)
-	}
-	if toolEvents != 2 {
-		t.Fatalf("resumed tool events = %d, want start/end for replayed checkpoint result", toolEvents)
-	}
-	if done == nil || done.Reply != "finished from streamed checkpoint" || done.RunID != runs[0].ID {
-		t.Fatalf("done response = %#v", done)
-	}
-}
-
-type unsupportedAgent struct{}
-
-func (unsupportedAgent) Name() string                                      { return "unsupported" }
-func (unsupportedAgent) Init(...Option)                                    {}
-func (unsupportedAgent) Options() Options                                  { return Options{} }
-func (unsupportedAgent) Ask(context.Context, string) (*Response, error)    { return nil, nil }
-func (unsupportedAgent) Stream(context.Context, string) (ai.Stream, error) { return nil, nil }
-func (unsupportedAgent) Run() error                                        { return nil }
-func (unsupportedAgent) Stop() error                                       { return nil }
-func (unsupportedAgent) String() string                                    { return "unsupported" }

ai/anthropic/anthropic.go

@@ -77,9 +77,11 @@ 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": anthropicMaxTokens(p.opts),
+		"max_tokens": 8192,
 		"system":     req.SystemPrompt,
-		"messages":   threadAnthropicMessages(req),
+		"messages": []map[string]any{
+			{"role": "user", "content": req.Prompt},
+		},
 	}
 
 	if len(anthropicTools) > 0 {
@@ -99,9 +101,10 @@ 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 := append(threadAnthropicMessages(req),
-		map[string]any{"role": "assistant", "content": cleanContent(rawContent)},
-	)
+	messages := []map[string]any{
+		{"role": "user", "content": req.Prompt},
+		{"role": "assistant", "content": cleanContent(rawContent)},
+	}
 
 	pendingCalls := resp.ToolCalls
 
@@ -124,7 +127,7 @@ func (p *Provider) Generate(ctx context.Context, req *ai.Request, opts ...ai.Gen
 
 		followUpReq := map[string]any{
 			"model":      p.opts.Model,
-			"max_tokens": anthropicMaxTokens(p.opts),
+			"max_tokens": 8192,
 			"system":     req.SystemPrompt,
 			"messages":   messages,
 		}
@@ -267,24 +270,3 @@ 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,7 +20,6 @@
 package atlascloud
 
 import (
-	"bufio"
 	"bytes"
 	"context"
 	"encoding/json"
@@ -43,7 +42,6 @@ func init() {
 	ai.RegisterVideo("atlascloud", func(opts ...ai.Option) ai.VideoModel {
 		return NewProvider(opts...)
 	})
-	ai.RegisterStream("atlascloud")
 }
 
 // Provider implements the ai.Model interface for Atlas Cloud.
@@ -93,21 +91,13 @@ func (p *Provider) Generate(ctx context.Context, req *ai.Request, opts ...ai.Gen
 
 	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})
+		{"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
@@ -152,111 +142,8 @@ 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) {
-	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()
+	return nil, fmt.Errorf("%w: atlascloud provider", ai.ErrStreamingUnsupported)
 }
 
 func (p *Provider) callAPI(ctx context.Context, req map[string]any) (*ai.Response, map[string]any, error) {

ai/atlascloud/atlascloud_test.go

@@ -2,11 +2,7 @@ package atlascloud
 
 import (
 	"context"
-	"encoding/json"
 	"errors"
-	"io"
-	"net/http"
-	"net/http/httptest"
 	"testing"
 
 	"go-micro.dev/v6/ai"
@@ -85,58 +81,16 @@ func TestProvider_Generate_NoAPIKey(t *testing.T) {
 	}
 }
 
-func TestProvider_Stream(t *testing.T) {
-	var sawStream, sawIncludeUsage bool
-	ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
-		if r.URL.Path != "/v1/chat/completions" {
-			t.Errorf("path = %s, want /v1/chat/completions", r.URL.Path)
-		}
-		var body map[string]any
-		if err := json.NewDecoder(r.Body).Decode(&body); err != nil {
-			t.Fatalf("decode request: %v", err)
-		}
-		sawStream, _ = body["stream"].(bool)
-		if so, ok := body["stream_options"].(map[string]any); ok {
-			sawIncludeUsage, _ = so["include_usage"].(bool)
-		}
-		w.Header().Set("Content-Type", "text/event-stream")
-		_, _ = w.Write([]byte("data: {\"choices\":[{\"delta\":{\"content\":\"hel\"}}]}\n\n"))
-		_, _ = w.Write([]byte("data: {\"choices\":[{\"delta\":{\"content\":\"lo\"}}]}\n\n"))
-		_, _ = w.Write([]byte("data: {\"choices\":[],\"usage\":{\"prompt_tokens\":7,\"completion_tokens\":2,\"total_tokens\":9}}\n\n"))
-		_, _ = w.Write([]byte("data: [DONE]\n\n"))
-	}))
-	defer ts.Close()
+func TestProvider_Stream_NotImplemented(t *testing.T) {
+	p := NewProvider()
 
-	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 !sawStream {
-		t.Fatal("stream request did not set stream=true")
-	}
-	if !sawIncludeUsage {
-		t.Fatal("stream request did not set stream_options.include_usage=true")
+	req := &ai.Request{
+		Prompt: "Hello",
 	}
 
-	first, err := stream.Recv()
-	if err != nil || first.Reply != "hel" {
-		t.Fatalf("first chunk = %#v, %v; want hel", first, err)
-	}
-	second, err := stream.Recv()
-	if err != nil || second.Reply != "lo" {
-		t.Fatalf("second chunk = %#v, %v; want lo", second, err)
-	}
-	usage, err := stream.Recv()
-	if err != nil {
-		t.Fatalf("usage chunk error: %v", err)
-	}
-	if usage.Usage.TotalTokens != 9 || usage.Usage.InputTokens != 7 || usage.Usage.OutputTokens != 2 {
-		t.Fatalf("usage = %#v; want input=7 output=2 total=9", usage.Usage)
-	}
-	if _, err := stream.Recv(); !errors.Is(err, io.EOF) {
-		t.Fatalf("final error = %v, want EOF", err)
+	_, err := p.Stream(context.Background(), req)
+	if !errors.Is(err, ai.ErrStreamingUnsupported) {
+		t.Fatalf("Stream error = %v, want ErrStreamingUnsupported", err)
 	}
 }
 

ai/capabilities_test.go

@@ -34,7 +34,7 @@ func TestRegisteredProviders(t *testing.T) {
 	}
 
 	got = ai.RegisteredProviders("stream")
-	want = []string{"atlascloud", "groq", "mistral", "openai", "together"}
+	want = []string{"openai"}
 	if !reflect.DeepEqual(got, want) {
 		t.Fatalf("RegisteredProviders(stream) = %#v, want %#v", got, want)
 	}
@@ -44,12 +44,12 @@ func TestCapabilityRows(t *testing.T) {
 	got := ai.CapabilityRows()
 	want := []ai.CapabilityRow{
 		{Provider: "anthropic", Capabilities: ai.Capabilities{Model: true}},
-		{Provider: "atlascloud", Capabilities: ai.Capabilities{Model: true, Image: true, Video: true, Stream: true}},
+		{Provider: "atlascloud", Capabilities: ai.Capabilities{Model: true, Image: true, Video: true}},
 		{Provider: "gemini", Capabilities: ai.Capabilities{Model: true}},
-		{Provider: "groq", Capabilities: ai.Capabilities{Model: true, Stream: true}},
-		{Provider: "mistral", Capabilities: ai.Capabilities{Model: true, Stream: true}},
+		{Provider: "groq", Capabilities: ai.Capabilities{Model: true}},
+		{Provider: "mistral", Capabilities: ai.Capabilities{Model: true}},
 		{Provider: "openai", Capabilities: ai.Capabilities{Model: true, Image: true, Stream: true}},
-		{Provider: "together", Capabilities: ai.Capabilities{Model: true, Stream: true}},
+		{Provider: "together", Capabilities: ai.Capabilities{Model: true}},
 	}
 	if !reflect.DeepEqual(got, want) {
 		t.Fatalf("CapabilityRows() = %#v, want %#v", got, want)
@@ -72,7 +72,7 @@ func TestCapabilityMatrix(t *testing.T) {
 	if caps := ai.ProviderCapabilities("openai"); caps != (ai.Capabilities{Model: true, Image: true, Stream: true}) {
 		t.Fatalf("ProviderCapabilities(openai) = %#v", caps)
 	}
-	if caps := ai.ProviderCapabilities("atlascloud"); caps != (ai.Capabilities{Model: true, Image: true, Video: true, Stream: true}) {
+	if caps := ai.ProviderCapabilities("atlascloud"); caps != (ai.Capabilities{Model: true, Image: true, Video: true}) {
 		t.Fatalf("ProviderCapabilities(atlascloud) = %#v", caps)
 	}
 	if caps := ai.ProviderCapabilities("missing"); caps != (ai.Capabilities{}) {
@@ -88,7 +88,7 @@ func TestRegisterStream(t *testing.T) {
 	}
 
 	got := ai.RegisteredProviders("stream")
-	want := []string{"atlascloud", "groq", "mistral", "openai", "test-stream", "together"}
+	want := []string{"openai", "test-stream"}
 	if !reflect.DeepEqual(got, want) {
 		t.Fatalf("RegisteredProviders(stream) = %#v, want %#v", got, want)
 	}

ai/groq/groq.go

@@ -22,14 +22,12 @@ import (
 	"strings"
 
 	"go-micro.dev/v6/ai"
-	"go-micro.dev/v6/ai/internal/openaiapi"
 )
 
 func init() {
 	ai.Register("groq", func(opts ...ai.Option) ai.Model {
 		return NewProvider(opts...)
 	})
-	ai.RegisterStream("groq")
 }
 
 type Provider struct {
@@ -121,7 +119,7 @@ func (p *Provider) Generate(ctx context.Context, req *ai.Request, opts ...ai.Gen
 }
 
 func (p *Provider) Stream(ctx context.Context, req *ai.Request, opts ...ai.GenerateOption) (ai.Stream, error) {
-	return openaiapi.Stream(ctx, p.opts, req, "/v1/chat/completions")
+	return nil, fmt.Errorf("%w: groq provider", ai.ErrStreamingUnsupported)
 }
 
 func (p *Provider) callAPI(ctx context.Context, req map[string]any) (*ai.Response, map[string]any, error) {

ai/groq/groq_test.go

@@ -2,11 +2,7 @@ package groq
 
 import (
 	"context"
-	"encoding/json"
 	"errors"
-	"io"
-	"net/http"
-	"net/http/httptest"
 	"testing"
 
 	"go-micro.dev/v6/ai"
@@ -44,44 +40,9 @@ func TestProvider_Generate_NoAPIKey(t *testing.T) {
 	}
 }
 
-func TestProvider_Stream(t *testing.T) {
-	var sawStream bool
-	ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
-		if r.URL.Path != "/v1/chat/completions" {
-			t.Fatalf("path = %s, want /v1/chat/completions", r.URL.Path)
-		}
-		var body map[string]any
-		if err := json.NewDecoder(r.Body).Decode(&body); err != nil {
-			t.Fatalf("decode request: %v", err)
-		}
-		sawStream, _ = body["stream"].(bool)
-		w.Header().Set("Content-Type", "text/event-stream")
-		_, _ = w.Write([]byte("data: {\"choices\":[{\"delta\":{\"content\":\"hel\"}}]}\n\n"))
-		_, _ = w.Write([]byte("data: {\"choices\":[{\"delta\":{\"content\":\"lo\"}}]}\n\n"))
-		_, _ = w.Write([]byte("data: [DONE]\n\n"))
-	}))
-	defer ts.Close()
-
-	p := NewProvider(ai.WithAPIKey("test-key"), ai.WithBaseURL(ts.URL))
-	stream, err := p.Stream(context.Background(), &ai.Request{Prompt: "Hello"})
-	if err != nil {
-		t.Fatalf("Stream returned error: %v", err)
-	}
-	defer stream.Close()
-	if !sawStream {
-		t.Fatal("stream request did not set stream=true")
-	}
-
-	first, err := stream.Recv()
-	if err != nil || first.Reply != "hel" {
-		t.Fatalf("first chunk = %#v, %v; want hel", first, err)
-	}
-	second, err := stream.Recv()
-	if err != nil || second.Reply != "lo" {
-		t.Fatalf("second chunk = %#v, %v; want lo", second, err)
-	}
-	if _, err := stream.Recv(); !errors.Is(err, io.EOF) {
-		t.Fatalf("final error = %v, want EOF", err)
+func TestProvider_Stream_NotImplemented(t *testing.T) {
+	if _, err := NewProvider().Stream(context.Background(), &ai.Request{Prompt: "hi"}); !errors.Is(err, ai.ErrStreamingUnsupported) {
+		t.Fatalf("Stream error = %v, want ErrStreamingUnsupported", err)
 	}
 }
 

ai/internal/openaiapi/stream.go

@@ -1,117 +0,0 @@
-package openaiapi
-
-import (
-	"bufio"
-	"bytes"
-	"context"
-	"encoding/json"
-	"fmt"
-	"io"
-	"net/http"
-	"strings"
-
-	"go-micro.dev/v6/ai"
-)
-
-// Stream opens an OpenAI-compatible chat completions SSE stream.
-func Stream(ctx context.Context, opts ai.Options, req *ai.Request, basePath string) (ai.Stream, error) {
-	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":          opts.Model,
-		"messages":       messages,
-		"stream":         true,
-		"stream_options": map[string]any{"include_usage": true},
-	}
-	if opts.MaxTokens > 0 {
-		apiReq["max_tokens"] = opts.MaxTokens
-	}
-	reqBody, err := json.Marshal(apiReq)
-	if err != nil {
-		return nil, fmt.Errorf("failed to marshal stream request: %w", err)
-	}
-	apiURL := strings.TrimRight(opts.BaseURL, "/") + basePath
-	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 "+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 &StreamReader{body: httpResp.Body, scanner: bufio.NewScanner(httpResp.Body)}, nil
-}
-
-// StreamReader reads OpenAI-compatible server-sent event chunks.
-type StreamReader struct {
-	body    io.ReadCloser
-	scanner *bufio.Scanner
-	closed  bool
-}
-
-func (s *StreamReader) Recv() (*ai.Response, error) {
-	for s.scanner.Scan() {
-		line := strings.TrimSpace(s.scanner.Text())
-		if line == "" || strings.HasPrefix(line, ":") {
-			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
-		}
-		if chunk.Usage != nil {
-			return &ai.Response{Usage: ai.Usage{
-				InputTokens:  chunk.Usage.PromptTokens,
-				OutputTokens: chunk.Usage.CompletionTokens,
-				TotalTokens:  chunk.Usage.TotalTokens,
-			}}, nil
-		}
-	}
-	if err := s.scanner.Err(); err != nil {
-		return nil, err
-	}
-	return nil, io.EOF
-}
-
-func (s *StreamReader) Close() error {
-	if s.closed {
-		return nil
-	}
-	s.closed = true
-	return s.body.Close()
-}

ai/mistral/mistral.go

@@ -22,14 +22,12 @@ import (
 	"strings"
 
 	"go-micro.dev/v6/ai"
-	"go-micro.dev/v6/ai/internal/openaiapi"
 )
 
 func init() {
 	ai.Register("mistral", func(opts ...ai.Option) ai.Model {
 		return NewProvider(opts...)
 	})
-	ai.RegisterStream("mistral")
 }
 
 type Provider struct {
@@ -121,7 +119,7 @@ func (p *Provider) Generate(ctx context.Context, req *ai.Request, opts ...ai.Gen
 }
 
 func (p *Provider) Stream(ctx context.Context, req *ai.Request, opts ...ai.GenerateOption) (ai.Stream, error) {
-	return openaiapi.Stream(ctx, p.opts, req, "/v1/chat/completions")
+	return nil, fmt.Errorf("%w: mistral provider", ai.ErrStreamingUnsupported)
 }
 
 func (p *Provider) callAPI(ctx context.Context, req map[string]any) (*ai.Response, map[string]any, error) {

ai/mistral/mistral_test.go

@@ -2,11 +2,7 @@ package mistral
 
 import (
 	"context"
-	"encoding/json"
 	"errors"
-	"io"
-	"net/http"
-	"net/http/httptest"
 	"testing"
 
 	"go-micro.dev/v6/ai"
@@ -44,44 +40,9 @@ func TestProvider_Generate_NoAPIKey(t *testing.T) {
 	}
 }
 
-func TestProvider_Stream(t *testing.T) {
-	var sawStream bool
-	ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
-		if r.URL.Path != "/v1/chat/completions" {
-			t.Fatalf("path = %s, want /v1/chat/completions", r.URL.Path)
-		}
-		var body map[string]any
-		if err := json.NewDecoder(r.Body).Decode(&body); err != nil {
-			t.Fatalf("decode request: %v", err)
-		}
-		sawStream, _ = body["stream"].(bool)
-		w.Header().Set("Content-Type", "text/event-stream")
-		_, _ = w.Write([]byte("data: {\"choices\":[{\"delta\":{\"content\":\"hel\"}}]}\n\n"))
-		_, _ = w.Write([]byte("data: {\"choices\":[{\"delta\":{\"content\":\"lo\"}}]}\n\n"))
-		_, _ = w.Write([]byte("data: [DONE]\n\n"))
-	}))
-	defer ts.Close()
-
-	p := NewProvider(ai.WithAPIKey("test-key"), ai.WithBaseURL(ts.URL))
-	stream, err := p.Stream(context.Background(), &ai.Request{Prompt: "Hello"})
-	if err != nil {
-		t.Fatalf("Stream returned error: %v", err)
-	}
-	defer stream.Close()
-	if !sawStream {
-		t.Fatal("stream request did not set stream=true")
-	}
-
-	first, err := stream.Recv()
-	if err != nil || first.Reply != "hel" {
-		t.Fatalf("first chunk = %#v, %v; want hel", first, err)
-	}
-	second, err := stream.Recv()
-	if err != nil || second.Reply != "lo" {
-		t.Fatalf("second chunk = %#v, %v; want lo", second, err)
-	}
-	if _, err := stream.Recv(); !errors.Is(err, io.EOF) {
-		t.Fatalf("final error = %v, want EOF", err)
+func TestProvider_Stream_NotImplemented(t *testing.T) {
+	if _, err := NewProvider().Stream(context.Background(), &ai.Request{Prompt: "hi"}); !errors.Is(err, ai.ErrStreamingUnsupported) {
+		t.Fatalf("Stream error = %v, want ErrStreamingUnsupported", err)
 	}
 }
 

ai/openai/openai.go

@@ -85,12 +85,7 @@ func (p *Provider) Generate(ctx context.Context, req *ai.Request, opts ...ai.Gen
 	// Build messages
 	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})
+		{"role": "user", "content": req.Prompt},
 	}
 
 	// Build initial request
@@ -98,9 +93,6 @@ 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
@@ -154,21 +146,12 @@ 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},
-	}
-	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})
+		{"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
+		"model":    p.opts.Model,
+		"messages": messages,
+		"stream":   true,
 	}
 	reqBody, err := json.Marshal(apiReq)
 	if err != nil {
@@ -220,27 +203,14 @@ 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 != "" {
-			return &ai.Response{Reply: chunk.Choices[0].Delta.Content}, nil
+		if len(chunk.Choices) == 0 || chunk.Choices[0].Delta.Content == "" {
+			continue
 		}
-		// 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
+		return &ai.Response{Reply: chunk.Choices[0].Delta.Content}, nil
 	}
 	if err := s.scanner.Err(); err != nil {
 		return nil, err

ai/openai/openai_test.go

@@ -8,7 +8,6 @@ import (
 	"net/http"
 	"net/http/httptest"
 	"testing"
-	"time"
 
 	"go-micro.dev/v6/ai"
 )
@@ -127,58 +126,6 @@ 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,8 +16,6 @@ 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
@@ -93,11 +91,3 @@ 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,34 +13,9 @@ type StatusCoder interface {
 	StatusCode() int
 }
 
-// RetryAfterCoder is implemented by provider errors that expose a server
-// supplied retry delay, such as HTTP Retry-After on a 429/503 response.
-type RetryAfterCoder interface {
-	RetryAfter() time.Duration
-}
-
-// ErrorKind classifies provider-boundary failures into stable buckets callers
-// can inspect without parsing provider-specific error strings.
-type ErrorKind string
-
-const (
-	ErrorKindUnknown     ErrorKind = "unknown"
-	ErrorKindCanceled    ErrorKind = "canceled"
-	ErrorKindTimeout     ErrorKind = "timeout"
-	ErrorKindRateLimited ErrorKind = "rate_limited"
-	ErrorKindUnavailable ErrorKind = "unavailable"
-	ErrorKindProvider    ErrorKind = "provider"
-)
-
-// ClassifiedError is implemented by errors that expose a stable ErrorKind.
-type ClassifiedError interface {
-	ErrorKind() ErrorKind
-}
-
 // RetryError is returned when Generate is retried and still fails.
 type RetryError struct {
 	Attempts int
-	Kind     ErrorKind
 	Err      error
 }
 
@@ -48,7 +23,7 @@ func (e *RetryError) Error() string {
 	if e == nil {
 		return ""
 	}
-	return fmt.Sprintf("ai generate failed after %d attempt(s) (%s): %v", e.Attempts, e.ErrorKind(), e.Err)
+	return fmt.Sprintf("ai generate failed after %d attempt(s): %v", e.Attempts, e.Err)
 }
 
 func (e *RetryError) Unwrap() error {
@@ -58,13 +33,6 @@ func (e *RetryError) Unwrap() error {
 	return e.Err
 }
 
-func (e *RetryError) ErrorKind() ErrorKind {
-	if e == nil || e.Kind == "" {
-		return ErrorKindUnknown
-	}
-	return e.Kind
-}
-
 // GeneratePolicy controls timeout and retry behavior for a model call.
 type GeneratePolicy struct {
 	Timeout     time.Duration
@@ -108,10 +76,9 @@ func GenerateWithRetry(ctx context.Context, m Model, req *Request, policy Genera
 		if ctx.Err() != nil {
 			return nil, ctx.Err()
 		}
-		transient := IsTransientError(err)
-		if attempt == policy.MaxAttempts || !transient {
-			if attempt > 1 || transient {
-				return nil, &RetryError{Attempts: attempt, Kind: ClassifyError(err), Err: err}
+		if attempt == policy.MaxAttempts || !IsTransientError(err) {
+			if attempt > 1 || IsTransientError(err) {
+				return nil, &RetryError{Attempts: attempt, Err: err}
 			}
 			return nil, err
 		}
@@ -119,7 +86,16 @@ 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 := 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
+		}
 		t := time.NewTimer(backoff)
 		select {
 		case <-ctx.Done():
@@ -130,81 +106,25 @@ func GenerateWithRetry(ctx context.Context, m Model, req *Request, policy Genera
 		case <-t.C:
 		}
 	}
-	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 {
-		return ""
-	}
-	var classified ClassifiedError
-	if errors.As(err, &classified) {
-		if kind := classified.ErrorKind(); kind != "" {
-			return kind
-		}
-	}
-	if errors.Is(err, context.Canceled) {
-		return ErrorKindCanceled
-	}
-	if errors.Is(err, context.DeadlineExceeded) {
-		return ErrorKindTimeout
-	}
-	var sc StatusCoder
-	if errors.As(err, &sc) {
-		code := sc.StatusCode()
-		switch {
-		case code == 429:
-			return ErrorKindRateLimited
-		case code >= 500:
-			return ErrorKindUnavailable
-		case code > 0:
-			return ErrorKindProvider
-		}
-	}
-	msg := strings.ToLower(err.Error())
-	switch {
-	case strings.Contains(msg, "rate limit") || strings.Contains(msg, "too many requests"):
-		return ErrorKindRateLimited
-	case strings.Contains(msg, "timeout") || strings.Contains(msg, "deadline"):
-		return ErrorKindTimeout
-	case strings.Contains(msg, "temporar") || strings.Contains(msg, "unavailable"):
-		return ErrorKindUnavailable
-	default:
-		return ErrorKindUnknown
-	}
+	return nil, &RetryError{Attempts: policy.MaxAttempts, Err: last}
 }
 
 // IsTransientError reports whether err is worth retrying at the provider boundary.
 func IsTransientError(err error) bool {
-	switch ClassifyError(err) {
-	case ErrorKindTimeout, ErrorKindRateLimited, ErrorKindUnavailable:
-		return true
-	default:
+	if err == nil {
 		return false
 	}
+	if errors.Is(err, context.Canceled) {
+		return false
+	}
+	if errors.Is(err, context.DeadlineExceeded) {
+		return true
+	}
+	var sc StatusCoder
+	if errors.As(err, &sc) {
+		code := sc.StatusCode()
+		return code == 429 || code >= 500
+	}
+	msg := strings.ToLower(err.Error())
+	return strings.Contains(msg, "rate limit") || strings.Contains(msg, "too many requests") || strings.Contains(msg, "timeout") || strings.Contains(msg, "temporar")
 }

ai/retry_test.go

@@ -133,91 +133,3 @@ func TestGenerateWithRetryAddsAttemptMetadataToRunInfo(t *testing.T) {
 		}
 	}
 }
-
-type statusErr int
-
-func (e statusErr) Error() string   { return "provider status" }
-func (e statusErr) StatusCode() int { return int(e) }
-
-type retryAfterErr struct {
-	delay time.Duration
-}
-
-func (e retryAfterErr) Error() string             { return "rate limit exceeded" }
-func (e retryAfterErr) StatusCode() int           { return 429 }
-func (e retryAfterErr) RetryAfter() time.Duration { return e.delay }
-
-func TestClassifyErrorDistinguishesOperationalOutcomes(t *testing.T) {
-	tests := []struct {
-		name string
-		err  error
-		want ErrorKind
-	}{
-		{name: "canceled", err: context.Canceled, want: ErrorKindCanceled},
-		{name: "timeout", err: context.DeadlineExceeded, want: ErrorKindTimeout},
-		{name: "rate limit status", err: statusErr(429), want: ErrorKindRateLimited},
-		{name: "unavailable status", err: statusErr(503), want: ErrorKindUnavailable},
-		{name: "provider status", err: statusErr(400), want: ErrorKindProvider},
-		{name: "rate limit text", err: errors.New("rate limit exceeded"), want: ErrorKindRateLimited},
-	}
-	for _, tt := range tests {
-		t.Run(tt.name, func(t *testing.T) {
-			if got := ClassifyError(tt.err); got != tt.want {
-				t.Fatalf("ClassifyError() = %q, want %q", got, tt.want)
-			}
-		})
-	}
-}
-
-func TestGenerateWithRetryExposesRetryErrorKind(t *testing.T) {
-	model := retryModel{generate: func(context.Context, *Request, ...GenerateOption) (*Response, error) {
-		return nil, statusErr(429)
-	}}
-
-	_, err := GenerateWithRetry(context.Background(), model, &Request{Prompt: "hi"}, GeneratePolicy{
-		MaxAttempts: 2,
-		Backoff:     time.Millisecond,
-	})
-	var retryErr *RetryError
-	if !errors.As(err, &retryErr) {
-		t.Fatalf("error = %T %[1]v, want RetryError", err)
-	}
-	if retryErr.ErrorKind() != ErrorKindRateLimited {
-		t.Fatalf("retry kind = %q, want %q", retryErr.ErrorKind(), ErrorKindRateLimited)
-	}
-	if !errors.Is(err, statusErr(429)) {
-		t.Fatalf("retry error does not unwrap provider status: %v", err)
-	}
-}
-
-func TestGenerateWithRetryHonorsRetryAfterWhenLongerThanBackoff(t *testing.T) {
-	attempts := 0
-	model := retryModel{generate: func(context.Context, *Request, ...GenerateOption) (*Response, error) {
-		attempts++
-		if attempts == 1 {
-			return nil, retryAfterErr{delay: 25 * time.Millisecond}
-		}
-		return &Response{Reply: "ok"}, nil
-	}}
-
-	start := time.Now()
-	resp, err := GenerateWithRetry(context.Background(), model, &Request{Prompt: "hi"}, GeneratePolicy{
-		MaxAttempts: 2,
-		Backoff:     time.Millisecond,
-	})
-	if err != nil {
-		t.Fatalf("GenerateWithRetry returned error: %v", err)
-	}
-	if resp.Reply != "ok" {
-		t.Fatalf("reply = %q, want ok", resp.Reply)
-	}
-	if elapsed := time.Since(start); elapsed < 20*time.Millisecond {
-		t.Fatalf("retry delay = %s, want RetryAfter delay to dominate base backoff", elapsed)
-	}
-}
-
-func TestGenerateWithRetryCapsRetryAfter(t *testing.T) {
-	if got := retryBackoff(retryAfterErr{delay: time.Minute}, 1, time.Millisecond); got != 30*time.Second {
-		t.Fatalf("retryBackoff() = %s, want 30s cap", got)
-	}
-}

ai/together/together.go

@@ -22,14 +22,12 @@ import (
 	"strings"
 
 	"go-micro.dev/v6/ai"
-	"go-micro.dev/v6/ai/internal/openaiapi"
 )
 
 func init() {
 	ai.Register("together", func(opts ...ai.Option) ai.Model {
 		return NewProvider(opts...)
 	})
-	ai.RegisterStream("together")
 }
 
 type Provider struct {
@@ -121,7 +119,7 @@ func (p *Provider) Generate(ctx context.Context, req *ai.Request, opts ...ai.Gen
 }
 
 func (p *Provider) Stream(ctx context.Context, req *ai.Request, opts ...ai.GenerateOption) (ai.Stream, error) {
-	return openaiapi.Stream(ctx, p.opts, req, "/v1/chat/completions")
+	return nil, fmt.Errorf("%w: together provider", ai.ErrStreamingUnsupported)
 }
 
 func (p *Provider) callAPI(ctx context.Context, req map[string]any) (*ai.Response, map[string]any, error) {

ai/together/together_test.go

@@ -2,11 +2,7 @@ package together
 
 import (
 	"context"
-	"encoding/json"
 	"errors"
-	"io"
-	"net/http"
-	"net/http/httptest"
 	"testing"
 
 	"go-micro.dev/v6/ai"
@@ -44,44 +40,9 @@ func TestProvider_Generate_NoAPIKey(t *testing.T) {
 	}
 }
 
-func TestProvider_Stream(t *testing.T) {
-	var sawStream bool
-	ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
-		if r.URL.Path != "/v1/chat/completions" {
-			t.Fatalf("path = %s, want /v1/chat/completions", r.URL.Path)
-		}
-		var body map[string]any
-		if err := json.NewDecoder(r.Body).Decode(&body); err != nil {
-			t.Fatalf("decode request: %v", err)
-		}
-		sawStream, _ = body["stream"].(bool)
-		w.Header().Set("Content-Type", "text/event-stream")
-		_, _ = w.Write([]byte("data: {\"choices\":[{\"delta\":{\"content\":\"hel\"}}]}\n\n"))
-		_, _ = w.Write([]byte("data: {\"choices\":[{\"delta\":{\"content\":\"lo\"}}]}\n\n"))
-		_, _ = w.Write([]byte("data: [DONE]\n\n"))
-	}))
-	defer ts.Close()
-
-	p := NewProvider(ai.WithAPIKey("test-key"), ai.WithBaseURL(ts.URL))
-	stream, err := p.Stream(context.Background(), &ai.Request{Prompt: "Hello"})
-	if err != nil {
-		t.Fatalf("Stream returned error: %v", err)
-	}
-	defer stream.Close()
-	if !sawStream {
-		t.Fatal("stream request did not set stream=true")
-	}
-
-	first, err := stream.Recv()
-	if err != nil || first.Reply != "hel" {
-		t.Fatalf("first chunk = %#v, %v; want hel", first, err)
-	}
-	second, err := stream.Recv()
-	if err != nil || second.Reply != "lo" {
-		t.Fatalf("second chunk = %#v, %v; want lo", second, err)
-	}
-	if _, err := stream.Recv(); !errors.Is(err, io.EOF) {
-		t.Fatalf("final error = %v, want EOF", err)
+func TestProvider_Stream_NotImplemented(t *testing.T) {
+	if _, err := NewProvider().Stream(context.Background(), &ai.Request{Prompt: "hi"}); !errors.Is(err, ai.ErrStreamingUnsupported) {
+		t.Fatalf("Stream error = %v, want ErrStreamingUnsupported", err)
 	}
 }
 

cmd/micro/cli/deploy/deploy.go

@@ -42,31 +42,14 @@ func Deploy(c *cli.Context) error {
 		return showDeployHelp()
 	}
 
-	target, remotePath := resolveDeployTarget(c, target, cfg)
-	if c.Bool("dry-run") {
-		return printDeployPlan(c, target, cfg, remotePath)
-	}
-
-	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 target, remotePath
+	return deploySSH(c, target, cfg)
 }
 
 func showDeployHelp() error {
@@ -99,82 +82,7 @@ func showDeployTargets(cfg *config.Config) error {
 	return fmt.Errorf("%s", sb.String())
 }
 
-func printDeployPlan(c *cli.Context, target string, cfg *config.Config, remotePath string) error {
-	dir := c.Args().Get(1)
-	if dir == "" {
-		dir = "."
-	}
-
-	absDir, err := filepath.Abs(dir)
-	if err != nil {
-		return fmt.Errorf("failed to get absolute path: %w", err)
-	}
-
-	if cfg == nil {
-		cfg, _ = config.Load(absDir)
-	}
-	if remotePath == "" {
-		remotePath = defaultRemotePath
-	}
-
-	services, err := deployServices(absDir, cfg, c.String("service"))
-	if err != nil {
-		return err
-	}
-
-	fmt.Println()
-	fmt.Println("  \033[1mmicro deploy --dry-run\033[0m")
-	fmt.Println()
-	fmt.Printf("  Target      \033[36m%s\033[0m\n", target)
-	fmt.Printf("  Remote path %s\n", remotePath)
-	fmt.Printf("  Services    %s\n", strings.Join(services, ", "))
-	fmt.Println()
-	fmt.Println("  Plan:")
-	fmt.Println("    1. Build linux/amd64 service binaries")
-	fmt.Printf("    2. Copy binaries to %s/bin/\n", remotePath)
-	fmt.Println("    3. Enable and restart micro@<service> systemd units")
-	fmt.Println("    4. Check service health")
-	fmt.Println()
-	fmt.Println("  No SSH, rsync, systemd, or remote deployment was performed.")
-	return nil
-}
-
-func deployServices(absDir string, cfg *config.Config, filterService string) ([]string, error) {
-	if filterService != "" && cfg != nil {
-		found := false
-		for _, svc := range cfg.Services {
-			if svc.Name == filterService {
-				found = true
-				break
-			}
-		}
-		if !found && len(cfg.Services) > 0 {
-			return nil, fmt.Errorf("service '%s' not found in configuration", filterService)
-		}
-	}
-
-	if cfg != nil && len(cfg.Services) > 0 {
-		sorted, err := cfg.TopologicalSort()
-		if err != nil {
-			return nil, err
-		}
-		services := make([]string, 0, len(sorted))
-		for _, svc := range sorted {
-			if filterService == "" || svc.Name == filterService {
-				services = append(services, svc.Name)
-			}
-		}
-		return services, nil
-	}
-
-	services := []string{filepath.Base(absDir)}
-	if filterService != "" && filterService != services[0] {
-		return nil, fmt.Errorf("service '%s' not found (only '%s' available)", filterService, services[0])
-	}
-	return services, nil
-}
-
-func deploySSH(c *cli.Context, target string, cfg *config.Config, remotePath string) error {
+func deploySSH(c *cli.Context, target string, cfg *config.Config) error {
 	dir := c.Args().Get(1)
 	if dir == "" {
 		dir = "."
@@ -190,6 +98,7 @@ func deploySSH(c *cli.Context, target string, cfg *config.Config, remotePath str
 		cfg, _ = config.Load(absDir)
 	}
 
+	remotePath := c.String("path")
 	if remotePath == "" {
 		remotePath = defaultRemotePath
 	}
@@ -199,10 +108,19 @@ func deploySSH(c *cli.Context, target string, cfg *config.Config, remotePath str
 	fmt.Println()
 	fmt.Printf("  Target     \033[36m%s\033[0m\n\n", target)
 
-	// Early validation: resolve services before SSH checks.
-	services, err := deployServices(absDir, cfg, c.String("service"))
-	if err != nil {
-		return err
+	// Early validation: Check if the requested service exists before SSH checks
+	filterService := c.String("service")
+	if filterService != "" && cfg != nil {
+		found := false
+		for _, svc := range cfg.Services {
+			if svc.Name == filterService {
+				found = true
+				break
+			}
+		}
+		if !found && len(cfg.Services) > 0 {
+			return fmt.Errorf("service '%s' not found in configuration", filterService)
+		}
 	}
 
 	// Step 1: Check SSH connectivity
@@ -222,6 +140,28 @@ func deploySSH(c *cli.Context, target string, cfg *config.Config, remotePath str
 	fmt.Println("\u2713")
 
 	// Step 3: Build binaries
+	var services []string
+	if cfg != nil && len(cfg.Services) > 0 {
+		sorted, err := cfg.TopologicalSort()
+		if err != nil {
+			return err
+		}
+		for _, svc := range sorted {
+			// If --service flag is provided, only include that service
+			if filterService == "" || svc.Name == filterService {
+				services = append(services, svc.Name)
+			}
+		}
+	} else {
+		// Single service project
+		services = []string{filepath.Base(absDir)}
+
+		// If --service flag was provided for a single-service project, validate it matches
+		if filterService != "" && filterService != services[0] {
+			return fmt.Errorf("service '%s' not found (only '%s' available)", filterService, services[0])
+		}
+	}
+
 	fmt.Printf("  Building binaries...       ")
 	if err := buildBinaries(absDir, cfg, c.Bool("build"), services); err != nil {
 		fmt.Println("\u2717")
@@ -540,10 +480,6 @@ The deploy process:
 				Name:  "service",
 				Usage: "Deploy only a specific service (for multi-service projects)",
 			},
-			&cli.BoolFlag{
-				Name:  "dry-run",
-				Usage: "Print the deployment plan without building, connecting, copying, or restarting services",
-			},
 		},
 	})
 }

cmd/micro/cli/deploy/deploy_test.go

@@ -1,167 +0,0 @@
-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, "")
-	set.Bool("dry-run", 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)
-	}
-}
-
-func TestDeployDryRunPlansConfiguredTargetWithoutRemoteSideEffects(t *testing.T) {
-	dir := t.TempDir()
-	if err := os.WriteFile(dir+"/micro.mu", []byte(`service api
-    path ./api
-
-deploy prod
-    ssh deploy@prod.example.com
-    path /srv/micro
-`), 0644); err != nil {
-		t.Fatalf("write config: %v", err)
-	}
-
-	oldwd, err := os.Getwd()
-	if err != nil {
-		t.Fatalf("getwd: %v", err)
-	}
-	if err := os.Chdir(dir); err != nil {
-		t.Fatalf("chdir: %v", err)
-	}
-	t.Cleanup(func() {
-		if err := os.Chdir(oldwd); err != nil {
-			t.Errorf("restore cwd: %v", err)
-		}
-	})
-
-	ctx := newDeployTestContext(t, "--dry-run", "prod")
-	if err := Deploy(ctx); err != nil {
-		t.Fatalf("dry-run deploy: %v", err)
-	}
-}
-
-func TestDeployDryRunValidatesRequestedService(t *testing.T) {
-	ctx := newDeployTestContext(t, "--dry-run", "--service", "missing", "prod")
-	cfg := &config.Config{Services: map[string]*config.Service{
-		"api": {Name: "api", Path: "./api"},
-	}}
-
-	err := printDeployPlan(ctx, "deploy@prod.example.com", cfg, defaultRemotePath)
-	if err == nil {
-		t.Fatal("expected dry-run to validate service names")
-	}
-	if !strings.Contains(err.Error(), "service 'missing' not found in configuration") {
-		t.Fatalf("unexpected error: %v", err)
-	}
-}

cmd/micro/cli/new/contract_test.go

@@ -1,23 +1,20 @@
 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, 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.
+// toolchain can build. 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`.
@@ -32,8 +29,6 @@ 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
@@ -53,8 +48,6 @@ func TestZeroToOneNoMCPContract(t *testing.T) {
 
 	generated.replaceModule(t)
 	generated.build(t)
-	generated.run(t)
-	generated.call(t, "Bob", "Hello Bob")
 }
 
 type generatedService struct {
@@ -128,74 +121,3 @@ 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

@@ -1,169 +0,0 @@
-// 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

@@ -1,64 +0,0 @@
-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,7 +12,6 @@ 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

@@ -1,50 +0,0 @@
-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", "deploy"} {
-		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")
-	}
-
-	var hasDeployDryRun bool
-	for _, command := range microcmd.DefaultCmd.App().Commands {
-		if command.Name != "deploy" {
-			continue
-		}
-		for _, flag := range command.Flags {
-			for _, name := range flag.Names() {
-				if name == "dry-run" {
-					hasDeployDryRun = true
-				}
-			}
-		}
-	}
-	if !hasDeployDryRun {
-		t.Fatal("missing deploy boundary: deploy --dry-run")
-	}
-}

events/memory.go

@@ -19,11 +19,7 @@ func NewStream(opts ...Option) (Stream, error) {
 	for _, o := range opts {
 		o(&options)
 	}
-	st := options.Store
-	if st == nil {
-		st = store.NewMemoryStore()
-	}
-	return &mem{store: st}, nil
+	return &mem{store: store.NewMemoryStore()}, nil
 }
 
 type subscriber struct {

events/options.go

@@ -1,25 +1,11 @@
 package events
 
-import (
-	"time"
+import "time"
 
-	"go-micro.dev/v6/store"
-)
-
-type Options struct {
-	// Store persists published events for durability and replay. If nil, an
-	// in-memory store is used and events do not survive a restart.
-	Store store.Store
-}
+type Options struct{}
 
 type Option func(o *Options)
 
-// WithStore backs the stream with a durable store (e.g. the file store), so
-// published events persist and can be replayed across restarts.
-func WithStore(s store.Store) Option {
-	return func(o *Options) { o.Store = s }
-}
-
 type StoreOptions struct {
 	TTL    time.Duration
 	Backup Backup

examples/README.md

@@ -80,11 +80,10 @@ A workflow as ordered, checkpointed steps that survives a crash and resumes wher
 - **Checkpoint** — each step is persisted; on `Resume`, completed steps are not re-run (no duplicate side effects)
 
 ### [support](./support/)
-A maintained 0-to-hero reference path in one runnable file:
-- **scaffold** typed `customers`, `tickets`, and `notify` services
-- **run/chat** with a support agent that uses those services as tools
-- **inspect** the event-driven `intake` flow and approval gate
-- **CI** keeps the deterministic mock-model journey runnable with `go test ./examples/support`
+A real-world support desk — the "zero to hero" shape in one runnable file:
+- **services** (`customers`, `tickets`, `notify`) become the agent's tools automatically
+- **flow** turns a `ticket.created` event into work for the agent (the event is the prompt)
+- **guardrail** — the agent triages freely but can't email a customer without passing the approval gate
 
 ## Coming Soon
 

examples/agent-durable/README.md

@@ -1,32 +0,0 @@
-# Durable agent run resume
-
-This example shows the agent-side counterpart to `examples/flow-durable`: an
-agent run is checkpointed with the same `Checkpoint` interface used by flows,
-then resumed after an interruption without repeating a completed side effect.
-The sample uses an in-memory store to keep repeated local runs deterministic;
-use your service store for process-restart recovery.
-
-Run it with:
-
-```sh
-go run ./examples/agent-durable
-```
-
-The demo model calls `inventory.reserve`, then fails to mimic a process dying
-after the tool call was checkpointed. `micro.AgentPending` finds the unfinished
-run and `micro.AgentResume` continues it from the saved checkpoint. The final
-`tool executions: 1` line is the important bit: the reservation tool was not
-called a second time during resume.
-
-In a service, use the same pattern at startup:
-
-```go
-pending, _ := micro.AgentPending(ctx, agent)
-for _, run := range pending {
-    _, _ = micro.AgentResume(ctx, agent, run.ID)
-}
-```
-
-`context.Context` cancellation and deadlines are still honored by checkpoint
-loads/saves, model calls, and tool calls. Runs with terminal statuses such as
-`done`, `canceled`, and `expired` are not returned by `AgentPending`.

examples/agent-durable/main.go

@@ -1,88 +0,0 @@
-// Package main demonstrates durable agent runs: a checkpointed agent can
-// resume after a crash without re-executing completed tool calls.
-package main
-
-import (
-	"context"
-	"errors"
-	"fmt"
-	"sync/atomic"
-
-	micro "go-micro.dev/v6"
-	"go-micro.dev/v6/ai"
-	"go-micro.dev/v6/store"
-)
-
-func main() {
-	ctx := context.Background()
-	checkpoint := micro.StoreCheckpoint(store.NewMemoryStore(), "durable-agent-demo")
-	model := &demoModel{failFirst: true}
-	ai.Register("durable-demo", func(opts ...ai.Option) ai.Model {
-		_ = model.Init(opts...)
-		return model
-	})
-	var reservations atomic.Int32
-
-	ag := micro.NewAgent("durable-agent-demo",
-		micro.AgentWithCheckpoint(checkpoint),
-		micro.AgentProvider("durable-demo"),
-		micro.AgentTool("inventory.reserve", "reserve inventory exactly once", map[string]any{
-			"sku": map[string]any{"type": "string"},
-		}, func(ctx context.Context, input map[string]any) (string, error) {
-			count := reservations.Add(1)
-			return fmt.Sprintf("reserved %s (execution %d)", input["sku"], count), nil
-		}),
-	)
-
-	_, err := ag.Ask(ctx, "reserve sku-123 and confirm")
-	fmt.Println("initial run:", err)
-
-	pending, err := micro.AgentPending(ctx, ag)
-	if err != nil {
-		panic(err)
-	}
-	if len(pending) == 0 {
-		panic("expected a checkpointed run to resume")
-	}
-
-	resp, err := micro.AgentResume(ctx, ag, pending[0].ID)
-	if err != nil {
-		panic(err)
-	}
-	fmt.Println("resumed reply:", resp.Reply)
-	fmt.Println("tool executions:", reservations.Load())
-}
-
-type demoModel struct {
-	failFirst bool
-	opts      ai.Options
-}
-
-func (m *demoModel) Init(opts ...ai.Option) error {
-	m.opts = ai.NewOptions(opts...)
-	return nil
-}
-func (m *demoModel) Options() ai.Options { return m.opts }
-func (m *demoModel) String() string      { return "durable-demo" }
-
-func (m *demoModel) Generate(ctx context.Context, req *ai.Request, opts ...ai.GenerateOption) (*ai.Response, error) {
-	if m.opts.ToolHandler != nil {
-		res := m.opts.ToolHandler(ctx, ai.ToolCall{
-			ID:    "reserve-1",
-			Name:  "inventory.reserve",
-			Input: map[string]any{"sku": "sku-123"},
-		})
-		if res.Content == "" {
-			return nil, errors.New("reservation tool returned no content")
-		}
-	}
-	if m.failFirst {
-		m.failFirst = false
-		return nil, errors.New("simulated process interruption after checkpointed tool call")
-	}
-	return &ai.Response{Reply: "sku-123 is reserved; no duplicate reservation was made"}, nil
-}
-
-func (m *demoModel) Stream(context.Context, *ai.Request, ...ai.GenerateOption) (ai.Stream, error) {
-	return nil, ai.ErrStreamingUnsupported
-}

examples/agent-human-input/README.md

@@ -1,44 +0,0 @@
-# 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.
-
-## Cancellation and deadlines
-
-`ResumeInput` uses the caller's `context.Context` for checkpoint reads, writes,
-and the resumed model/tool turn. If the context is canceled or its deadline
-expires before the resume is committed, the call returns the context error and
-the checkpointed run remains `paused` at `input-required`; list it with
-`agent.Pending` and retry with a fresh context once the operator is ready.

examples/support/README.md

@@ -1,25 +1,9 @@
-# Zero-to-hero support desk
+# Support desk
 
-A maintained 0-to-hero reference for the Go Micro lifecycle: scaffold a few
-typed services, run them in one process, let an agent chat with those services
-as tools, then inspect the durable flow that triggered the work. It is one
-runnable file and one CI smoke test, so the reference path stays honest as the
-framework evolves.
-
-## The path
-
-1. **Scaffold services** — `customers`, `tickets`, and `notify` are ordinary
-   typed Go Micro services. Their request/response structs and method comments
-   become the tool contract the agent sees.
-2. **Run the harness** — the example starts an in-memory registry, broker,
-   client, store, services, agent, and flow in one process; no external
-   dependencies or API key are required for the default run.
-3. **Chat through an agent** — the `support` agent receives the ticket event as
-   a prompt and calls service tools to look up the customer, triage the ticket,
-   and draft a reply.
-4. **Inspect the workflow** — the `intake` flow records the event-driven run and
-   prints the agent result, showing the service → agent → workflow lifecycle as
-   one runtime.
+A real-world agent built the Go Micro way: a few services, an agent that
+manages them, an event that triggers it, and a human-in-the-loop gate on the
+one action that touches a customer. It's the "zero to hero" shape in one
+runnable file.
 
 ## The scenario
 
@@ -60,12 +44,6 @@ agent, which:
 go run main.go            # mock model — deterministic, no API key
 ```
 
-The maintained check is the same deterministic path:
-
-```bash
-go test ./examples/support
-```
-
 Against a live model, the agent reasons about the ticket itself instead of
 following the script:
 

examples/support/main.go

@@ -212,39 +212,37 @@ func waitFor(reg registry.Registry, names ...string) {
 	}
 }
 
-func runSupport(provider string) error {
+func main() {
+	provider := flag.String("provider", "mock", "LLM provider: mock (default), anthropic, openai, ...")
+	flag.Parse()
+
 	apiKey := ""
-	if provider == "mock" {
+	if *provider == "mock" {
 		ai.Register("mock", newMock)
-	} 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)
+	} 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)
 	}
 
-	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 {
-		return fmt.Errorf("broker connect: %w", err)
+		fmt.Println("broker connect:", err)
+		os.Exit(1)
 	}
 	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.Address("127.0.0.1:0"), service.Registry(reg), service.Client(cl), service.HandleSignal(false))
+		svc := service.New(service.Name(name), service.Registry(reg), service.Client(cl))
 		_ = 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
@@ -252,11 +250,10 @@ func runSupport(provider string) error {
 	// 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"])
@@ -278,7 +275,8 @@ func runSupport(provider string) error {
 		flow.Prompt("A new support ticket arrived: {{.Data}}. Handle it."),
 	)
 	if err := intake.Register(reg, br, cl); err != nil {
-		return fmt.Errorf("flow register: %w", err)
+		fmt.Println("flow register:", err)
+		os.Exit(1)
 	}
 	defer intake.Stop()
 
@@ -289,7 +287,8 @@ func runSupport(provider string) error {
 	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 {
-		return fmt.Errorf("publish: %w", err)
+		fmt.Println("publish:", err)
+		os.Exit(1)
 	}
 
 	// Wait for the agent to act.
@@ -306,18 +305,7 @@ func runSupport(provider string) error {
 	}
 	if notify.sent >= 1 {
 		fmt.Println("\n\033[32m✓ ticket triaged and the customer was replied to — triggered by an event\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)
+	} else {
+		fmt.Println("\n\033[31m✗ the agent did not complete the triage\033[0m")
 	}
 }

examples/support/main_test.go

@@ -1,32 +0,0 @@
-package main
-
-import (
-	"os"
-	"strings"
-	"testing"
-)
-
-func TestRunSupportMockSmoke(t *testing.T) {
-	if err := runSupport("mock"); err != nil {
-		t.Fatalf("support example failed: %v", err)
-	}
-}
-
-func TestZeroToHeroReadmeDocumentsLifecycle(t *testing.T) {
-	b, err := os.ReadFile("README.md")
-	if err != nil {
-		t.Fatalf("read README.md: %v", err)
-	}
-	doc := string(b)
-	for _, want := range []string{
-		"Scaffold services",
-		"Run the harness",
-		"Chat through an agent",
-		"Inspect the workflow",
-		"go test ./examples/support",
-	} {
-		if !strings.Contains(doc, want) {
-			t.Fatalf("README.md missing zero-to-hero step %q", want)
-		}
-	}
-}

gateway/a2a/a2a.go

@@ -21,14 +21,13 @@
 // 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, input-required handoffs, `tasks/resubscribe`,
-// and Agent Card discovery.
+// notification delivery, and Agent Card discovery. `input-required` and
+// `tasks/resubscribe` are advertised as unsupported and are follow-ups.
 package a2a
 
 import (
 	"context"
 	"encoding/json"
-	"errors"
 	"fmt"
 	"io"
 	"log"
@@ -141,11 +140,8 @@ func (g *Gateway) Handler() http.Handler {
 	// Per-agent card (served at the agent's url and at its well-known path).
 	mux.HandleFunc("GET /agents/{name}", g.handleCard)
 	mux.HandleFunc("GET /agents/{name}/.well-known/agent.json", g.handleCard)
-	mux.HandleFunc("GET /agents/{name}/skills/{skill}", g.handleSkillCard)
-	mux.HandleFunc("GET /agents/{name}/skills/{skill}/.well-known/agent.json", g.handleSkillCard)
 	// Per-agent JSON-RPC endpoint.
 	mux.HandleFunc("POST /agents/{name}", g.handleRPC)
-	mux.HandleFunc("POST /agents/{name}/skills/{skill}", g.handleSkillRPC)
 	// Top-level well-known: serve the single agent's card if there's
 	// exactly one, otherwise point to the directory.
 	mux.HandleFunc("GET /.well-known/agent.json", g.handleWellKnown)
@@ -240,10 +236,9 @@ type PushNotificationConfig struct {
 
 // Task states (JSON-RPC binding wire values).
 const (
-	stateCompleted     = "completed"
-	stateFailed        = "failed"
-	stateWorking       = "working"
-	stateInputRequired = "input-required"
+	stateCompleted = "completed"
+	stateFailed    = "failed"
+	stateWorking   = "working"
 )
 
 // JSON-RPC envelopes.
@@ -342,17 +337,23 @@ func Card(name, url, description string, services []string) AgentCard {
 			description = "Go Micro agent"
 		}
 	}
-	skills := skillsFromServices(services)
 	return AgentCard{
-		Name:               name,
-		Description:        description,
-		URL:                url,
-		Version:            "1.0.0",
-		ProtocolVersion:    protocolVersion,
-		Capabilities:       Capabilities{Streaming: true, PushNotifications: true},
+		Name:            name,
+		Description:     description,
+		URL:             url,
+		Version:         "1.0.0",
+		ProtocolVersion: protocolVersion,
+		Capabilities:    Capabilities{Streaming: true, PushNotifications: true},
+		// The agent converses over a single Chat endpoint; advertise that
+		// as one skill, tagged with the services it manages.
 		DefaultInputModes:  []string{"text/plain"},
 		DefaultOutputModes: []string{"text/plain"},
-		Skills:             skills,
+		Skills: []Skill{{
+			ID:          "chat",
+			Name:        "Chat",
+			Description: "Converse with the agent to operate its services.",
+			Tags:        services,
+		}},
 	}
 }
 
@@ -369,19 +370,6 @@ func (g *Gateway) lookupCard(name string) (AgentCard, bool) {
 	return g.card(name, meta), true
 }
 
-func (g *Gateway) lookupSkillCard(name, skillID string) (AgentCard, Skill, bool) {
-	card, ok := g.lookupCard(name)
-	if !ok {
-		return AgentCard{}, Skill{}, false
-	}
-	for _, skill := range card.Skills {
-		if skill.ID == skillID {
-			return card, skill, true
-		}
-	}
-	return AgentCard{}, Skill{}, false
-}
-
 // ---------------------------------------------------------------------------
 // HTTP handlers
 // ---------------------------------------------------------------------------
@@ -404,17 +392,6 @@ func (g *Gateway) handleCard(w http.ResponseWriter, r *http.Request) {
 	writeJSON(w, http.StatusOK, card)
 }
 
-func (g *Gateway) handleSkillCard(w http.ResponseWriter, r *http.Request) {
-	card, skill, ok := g.lookupSkillCard(r.PathValue("name"), r.PathValue("skill"))
-	if !ok {
-		http.NotFound(w, r)
-		return
-	}
-	card.URL = g.opts.BaseURL + "/agents/" + r.PathValue("name") + "/skills/" + skill.ID
-	card.Skills = []Skill{skill}
-	writeJSON(w, http.StatusOK, card)
-}
-
 func (g *Gateway) handleWellKnown(w http.ResponseWriter, r *http.Request) {
 	cards, err := g.cards()
 	if err != nil {
@@ -443,18 +420,6 @@ func (g *Gateway) handleRPC(w http.ResponseWriter, r *http.Request) {
 	})
 }
 
-func (g *Gateway) handleSkillRPC(w http.ResponseWriter, r *http.Request) {
-	name := r.PathValue("name")
-	_, skill, ok := g.lookupSkillCard(name, r.PathValue("skill"))
-	if !ok {
-		writeRPC(w, nil, nil, &rpcError{Code: errInvalidParams, Message: "unknown agent skill: " + name + "/" + r.PathValue("skill")})
-		return
-	}
-	g.disp.serve(w, r, func(ctx context.Context, text string) (string, error) {
-		return g.callAgent(ctx, name, skillPrompt(skill, text))
-	})
-}
-
 // dispatcher handles A2A JSON-RPC requests against an Invoke function and
 // retains recent tasks for tasks/get. It is shared by the gateway (one
 // per registry) and embedded agents (one per agent).
@@ -462,12 +427,11 @@ 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{}, watchers: map[string]map[chan *Task]struct{}{}}
+	return &dispatcher{tasks: map[string]*Task{}, pushConfigs: map[string]PushNotificationConfig{}}
 }
 
 func (d *dispatcher) serve(w http.ResponseWriter, r *http.Request, invoke Invoke) {
@@ -490,7 +454,7 @@ func (d *dispatcher) serveWithStream(w http.ResponseWriter, r *http.Request, inv
 		d.send(requestContext(r.Context()), w, req, invoke)
 	case "message/stream":
 		if streamInvoke != nil {
-			d.streamChunks(requestContext(r.Context()), w, req, streamInvoke, invoke)
+			d.streamChunks(requestContext(r.Context()), w, req, streamInvoke)
 			return
 		}
 		d.stream(requestContext(r.Context()), w, req, invoke)
@@ -504,7 +468,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":
-		d.resubscribe(requestContext(r.Context()), w, req)
+		writeRPC(w, req.ID, nil, &rpcError{Code: errMethodNotFound, Message: "resubscribe is not supported"})
 	default:
 		writeRPC(w, req.ID, nil, &rpcError{Code: errMethodNotFound, Message: "method not found: " + req.Method})
 	}
@@ -539,7 +503,7 @@ func (d *dispatcher) stream(ctx context.Context, w http.ResponseWriter, req rpcR
 	}
 }
 
-func (d *dispatcher) streamChunks(ctx context.Context, w http.ResponseWriter, req rpcRequest, invoke StreamInvoke, fallback Invoke) {
+func (d *dispatcher) streamChunks(ctx context.Context, w http.ResponseWriter, req rpcRequest, invoke StreamInvoke) {
 	var p sendParams
 	if err := json.Unmarshal(req.Params, &p); err != nil {
 		writeRPC(w, req.ID, nil, &rpcError{Code: errInvalidParams, Message: "invalid params"})
@@ -552,10 +516,6 @@ func (d *dispatcher) streamChunks(ctx context.Context, w http.ResponseWriter, re
 	}
 	stream, err := invoke(ctx, text)
 	if err != nil {
-		if errors.Is(err, ai.ErrStreamingUnsupported) && fallback != nil {
-			d.stream(ctx, w, req, fallback)
-			return
-		}
 		writeRPC(w, req.ID, nil, &rpcError{Code: errInternal, Message: err.Error()})
 		return
 	}
@@ -597,7 +557,6 @@ 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()
 	}
@@ -618,10 +577,6 @@ 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)
@@ -632,49 +587,6 @@ 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 == "" {
@@ -760,59 +672,18 @@ 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
-	if !exists {
-		d.order = append(d.order, t.ID)
-	}
+	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
@@ -891,68 +762,6 @@ func (d *dispatcher) deliverPush(taskID string, task *Task) {
 	}
 }
 
-func skillsFromServices(services []string) []Skill {
-	if len(services) == 0 {
-		return []Skill{{ID: "chat", Name: "Chat", Description: "Converse with the agent to operate its services."}}
-	}
-	seen := map[string]bool{}
-	var skills []Skill
-	for _, service := range services {
-		service = strings.TrimSpace(service)
-		if service == "" {
-			continue
-		}
-		id := skillID(service)
-		if id == "" || seen[id] {
-			continue
-		}
-		seen[id] = true
-		skills = append(skills, Skill{
-			ID:          id,
-			Name:        skillName(service),
-			Description: fmt.Sprintf("Operate the %s service through this agent.", service),
-			Tags:        []string{service},
-		})
-	}
-	if len(skills) == 0 {
-		return []Skill{{ID: "chat", Name: "Chat", Description: "Converse with the agent to operate its services."}}
-	}
-	return skills
-}
-
-func skillID(service string) string {
-	service = strings.ToLower(strings.TrimSpace(service))
-	var b strings.Builder
-	dash := false
-	for _, r := range service {
-		if (r >= 'a' && r <= 'z') || (r >= '0' && r <= '9') {
-			b.WriteRune(r)
-			dash = false
-			continue
-		}
-		if !dash && b.Len() > 0 {
-			b.WriteByte('-')
-			dash = true
-		}
-	}
-	return strings.Trim(b.String(), "-")
-}
-
-func skillName(service string) string {
-	parts := strings.FieldsFunc(service, func(r rune) bool { return r == '-' || r == '_' || r == '.' || r == '/' || r == ' ' })
-	for i, part := range parts {
-		if part == "" {
-			continue
-		}
-		parts[i] = strings.ToUpper(part[:1]) + part[1:]
-	}
-	return strings.Join(parts, " ")
-}
-
-func skillPrompt(skill Skill, text string) string {
-	return fmt.Sprintf("Use the %q skill (%s) for this request.\n\n%s", skill.Name, skill.ID, text)
-}
-
 func textOf(parts []Part) string {
 	var b strings.Builder
 	for _, p := range parts {

gateway/a2a/a2a_test.go

@@ -4,7 +4,6 @@ import (
 	"bytes"
 	"context"
 	"encoding/json"
-	"errors"
 	"fmt"
 	"io"
 	"net/http"
@@ -51,7 +50,7 @@ func newGatewayWithAgent(t *testing.T) (*httptest.Server, func()) {
 		server.Name("echo"),
 		server.Address("127.0.0.1:0"),
 		server.Registry(reg),
-		server.Metadata(map[string]string{"type": "agent", "services": "task,project"}),
+		server.Metadata(map[string]string{"type": "agent", "services": ""}),
 	)
 	if err := pb.RegisterAgentHandler(srv, echoAgent{}); err != nil {
 		t.Fatalf("register agent handler: %v", err)
@@ -88,40 +87,8 @@ func TestAgentCardFromRegistry(t *testing.T) {
 	if card.URL != "http://gw/agents/echo" {
 		t.Errorf("card url = %q", card.URL)
 	}
-	if card.ProtocolVersion == "" {
-		t.Errorf("card missing protocolVersion: %+v", card)
-	}
-	if got := skillIDs(card.Skills); strings.Join(got, ",") != "task,project" {
-		t.Errorf("skill IDs = %v, want [task project]", got)
-	}
-}
-
-func TestSkillEndpointServesFocusedCardAndRoutesRPC(t *testing.T) {
-	ts, cleanup := newGatewayWithAgent(t)
-	defer cleanup()
-
-	resp, err := http.Get(ts.URL + "/agents/echo/skills/task/.well-known/agent.json")
-	if err != nil {
-		t.Fatalf("get skill card: %v", err)
-	}
-	defer resp.Body.Close()
-	if resp.StatusCode != http.StatusOK {
-		t.Fatalf("skill card status = %d", resp.StatusCode)
-	}
-	var card AgentCard
-	if err := json.NewDecoder(resp.Body).Decode(&card); err != nil {
-		t.Fatalf("decode skill card: %v", err)
-	}
-	if card.URL != "http://gw/agents/echo/skills/task" || len(card.Skills) != 1 || card.Skills[0].ID != "task" {
-		t.Fatalf("skill card = %+v, want task-only card at skill URL", card)
-	}
-
-	task := rpcTask(t, ts.URL+"/agents/echo/skills/task", `{
-		"jsonrpc":"2.0","id":1,"method":"message/send",
-		"params":{"message":{"role":"user","kind":"message","messageId":"m1",
-			"parts":[{"kind":"text","text":"ping"}]}}}`)
-	if task.Status.State != stateCompleted || textOf(task.Artifacts[0].Parts) != "pong" {
-		t.Fatalf("skill task = %+v, want completed pong", task)
+	if card.ProtocolVersion == "" || len(card.Skills) == 0 {
+		t.Errorf("card missing protocolVersion or skills: %+v", card)
 	}
 }
 
@@ -392,248 +359,6 @@ 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 TestMessageStreamChunksFallsBackWhenUnsupported(t *testing.T) {
-	d := newDispatcher()
-	body := `{"jsonrpc":"2.0","id":1,"method":"message/stream","params":{"message":{"role":"user","parts":[{"kind":"text","text":"ping"}],"kind":"message"}}}`
-	req := httptest.NewRequest(http.MethodPost, "/", bytes.NewBufferString(body))
-	rr := httptest.NewRecorder()
-	var streamed bool
-	var fallbackText string
-
-	d.serveWithStream(rr, req, func(ctx context.Context, text string) (string, error) {
-		fallbackText = text
-		return "pong", nil
-	}, func(ctx context.Context, text string) (ai.Stream, error) {
-		streamed = true
-		return nil, fmt.Errorf("%w: test provider", ai.ErrStreamingUnsupported)
-	})
-
-	if !streamed {
-		t.Fatal("stream invoke was not attempted")
-	}
-	if fallbackText != "ping" {
-		t.Fatalf("fallback text = %q, want ping", fallbackText)
-	}
-	if ct := rr.Result().Header.Get("Content-Type"); !strings.HasPrefix(ct, "text/event-stream") {
-		t.Fatalf("content-type = %q, want text/event-stream", ct)
-	}
-	var events []struct {
-		Result Task      `json:"result"`
-		Error  *rpcError `json:"error"`
-	}
-	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())
-	}
-	if events[0].Error != nil {
-		t.Fatalf("fallback event error: %+v", events[0].Error)
-	}
-	if events[0].Result.Status.State != stateCompleted || textOf(events[0].Result.Artifacts[0].Parts) != "pong" {
-		t.Fatalf("fallback task = %+v, want completed pong", events[0].Result)
-	}
-}
-
-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)
@@ -683,9 +408,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":"unknown","params":{}}`, &resp)
+	rpc(t, ts.URL+"/agents/echo", `{"jsonrpc":"2.0","id":1,"method":"tasks/resubscribe","params":{}}`, &resp)
 	if resp.Error == nil || resp.Error.Code != errMethodNotFound {
-		t.Errorf("expected method-not-found, got %+v", resp.Error)
+		t.Errorf("expected method-not-found for resubscribe, got %+v", resp.Error)
 	}
 }
 
@@ -731,11 +456,3 @@ func rpcTask(t *testing.T, url, body string) Task {
 	}
 	return resp.Result
 }
-
-func skillIDs(skills []Skill) []string {
-	ids := make([]string, 0, len(skills))
-	for _, skill := range skills {
-		ids = append(ids, skill.ID)
-	}
-	return ids
-}

gateway/mcp/httpjsonrpc.go

@@ -1,129 +0,0 @@
-package mcp
-
-import (
-	"encoding/json"
-	"net/http"
-	"strings"
-)
-
-// HandlerOption configures NewHandler.
-type HandlerOption func(*handlerOptions)
-
-type handlerOptions struct {
-	serverName, serverVersion, protocolVersion string
-}
-
-// WithServerInfo sets the name/version advertised in the initialize response.
-func WithServerInfo(name, version string) HandlerOption {
-	return func(o *handlerOptions) { o.serverName, o.serverVersion = name, version }
-}
-
-// WithProtocolVersion sets the MCP protocol version advertised in initialize.
-func WithProtocolVersion(v string) HandlerOption {
-	return func(o *handlerOptions) { o.protocolVersion = v }
-}
-
-// NewHandler returns an http.Handler serving the MCP protocol over HTTP as
-// JSON-RPC 2.0 (initialize, ping, notifications/*, tools/list, tools/call),
-// backed by the resolver. Mount it on your own server (e.g. POST /mcp): the
-// gateway provides the protocol; you keep your routes, middleware and any
-// human-facing docs page.
-func NewHandler(r Resolver, opts ...HandlerOption) http.Handler {
-	o := handlerOptions{serverName: "go-micro-mcp", serverVersion: "1.0.0", protocolVersion: "2024-11-05"}
-	for _, fn := range opts {
-		fn(&o)
-	}
-	return http.HandlerFunc(func(w http.ResponseWriter, req *http.Request) {
-		if req.Method != http.MethodPost {
-			http.Error(w, "method not allowed", http.StatusMethodNotAllowed)
-			return
-		}
-		var rpc struct {
-			JSONRPC string          `json:"jsonrpc"`
-			ID      json.RawMessage `json:"id"`
-			Method  string          `json:"method"`
-			Params  json.RawMessage `json:"params"`
-		}
-		if err := json.NewDecoder(req.Body).Decode(&rpc); err != nil {
-			writeRPCError(w, nil, ParseError, "Parse error", err.Error())
-			return
-		}
-
-		// Notifications (and any id-less request) expect no response body.
-		if strings.HasPrefix(rpc.Method, "notifications/") || len(rpc.ID) == 0 {
-			w.WriteHeader(http.StatusNoContent)
-			return
-		}
-
-		ctx := req.Context()
-		switch rpc.Method {
-		case "initialize":
-			writeRPCResult(w, rpc.ID, map[string]interface{}{
-				"protocolVersion": o.protocolVersion,
-				"capabilities":    map[string]interface{}{"tools": map[string]interface{}{}},
-				"serverInfo":      map[string]interface{}{"name": o.serverName, "version": o.serverVersion},
-			})
-		case "ping":
-			writeRPCResult(w, rpc.ID, map[string]interface{}{})
-		case "tools/list":
-			tools, err := r.List(ctx)
-			if err != nil {
-				writeRPCError(w, rpc.ID, InternalError, "Failed to list tools", err.Error())
-				return
-			}
-			list := make([]map[string]interface{}, 0, len(tools))
-			for _, t := range tools {
-				list = append(list, map[string]interface{}{
-					"name": t.Name, "description": t.Description, "inputSchema": t.InputSchema,
-				})
-			}
-			writeRPCResult(w, rpc.ID, map[string]interface{}{"tools": list})
-		case "tools/call":
-			var p struct {
-				Name      string                 `json:"name"`
-				Arguments map[string]interface{} `json:"arguments"`
-			}
-			if err := json.Unmarshal(rpc.Params, &p); err != nil {
-				writeRPCError(w, rpc.ID, InvalidParams, "Invalid params", err.Error())
-				return
-			}
-			res, err := r.Call(ctx, p.Name, p.Arguments)
-			if err != nil {
-				// Protocol/pre-check failure -> JSON-RPC error. An *RPCError
-				// carries a specific code; anything else is InternalError.
-				if rpcErr, ok := err.(*RPCError); ok {
-					writeRPCError(w, rpc.ID, rpcErr.Code, rpcErr.Message, rpcErr.Data)
-				} else {
-					writeRPCError(w, rpc.ID, InternalError, "Tool call failed", err.Error())
-				}
-				return
-			}
-			result := map[string]interface{}{
-				"content": []map[string]interface{}{{"type": "text", "text": res.Text}},
-			}
-			if res.IsError {
-				result["isError"] = true
-			}
-			writeRPCResult(w, rpc.ID, result)
-		default:
-			writeRPCError(w, rpc.ID, MethodNotFound, "Method not found", rpc.Method)
-		}
-	})
-}
-
-func writeRPCResult(w http.ResponseWriter, id json.RawMessage, result interface{}) {
-	w.Header().Set("Content-Type", "application/json")
-	_ = json.NewEncoder(w).Encode(map[string]interface{}{"jsonrpc": "2.0", "id": rawOrNull(id), "result": result})
-}
-
-func writeRPCError(w http.ResponseWriter, id json.RawMessage, code int, msg string, data interface{}) {
-	w.Header().Set("Content-Type", "application/json")
-	_ = json.NewEncoder(w).Encode(map[string]interface{}{"jsonrpc": "2.0", "id": rawOrNull(id), "error": map[string]interface{}{"code": code, "message": msg, "data": data}})
-}
-
-func rawOrNull(id json.RawMessage) interface{} {
-	if len(id) == 0 {
-		return nil
-	}
-	return id
-}

gateway/mcp/resolver.go

@@ -1,131 +0,0 @@
-package mcp
-
-import (
-	"context"
-	"sync"
-
-	"go-micro.dev/v6/ai"
-	"go-micro.dev/v6/client"
-	"go-micro.dev/v6/registry"
-)
-
-// CallResult is the outcome of a successful tool dispatch. A tool that ran but
-// produced an error sets IsError — per the MCP spec this is returned as a
-// tools/call result with isError:true, not a JSON-RPC protocol error.
-type CallResult struct {
-	Text    string
-	IsError bool
-}
-
-// Error lets the package's RPCError (see stdio.go) be returned by a resolver
-// to signal a protocol/pre-check failure with a specific JSON-RPC code; the
-// handler maps it straight to the JSON-RPC error.
-func (e *RPCError) Error() string { return e.Message }
-
-// ToolFunc executes a manually-registered tool. Return a *CallResult for tool
-// outcomes (set IsError for tool-level failures); return a non-nil error — an
-// *RPCError for a specific code — for protocol/pre-check failures.
-type ToolFunc func(ctx context.Context, args map[string]any) (*CallResult, error)
-
-// Resolver supplies the gateway's tools and executes calls. Swapping the
-// resolver changes where tools come from without touching the MCP protocol or
-// transport:
-//
-//   - NewManualResolver:   tools you register explicitly (full product control,
-//     including tools that are not go-micro services, executed via your own
-//     logic — auth, metering, …).
-//   - NewRegistryResolver: tools auto-discovered from registered services.
-//
-// The built-in store/broker tools are intentionally NOT exposed by any
-// resolver — they remain a development convenience on the legacy Serve() path.
-type Resolver interface {
-	// List returns the current tool catalog.
-	List(ctx context.Context) ([]Tool, error)
-	// Call executes a tool by name with JSON arguments.
-	Call(ctx context.Context, name string, args map[string]any) (*CallResult, error)
-}
-
-// ManualResolver exposes an explicitly-registered set of tools.
-type ManualResolver struct {
-	mu    sync.RWMutex
-	order []Tool
-	funcs map[string]ToolFunc
-}
-
-// NewManualResolver returns an empty manual resolver.
-func NewManualResolver() *ManualResolver {
-	return &ManualResolver{funcs: map[string]ToolFunc{}}
-}
-
-// Add registers (or replaces) a tool and its handler. Returns the resolver for
-// chaining.
-func (m *ManualResolver) Add(t Tool, fn ToolFunc) *ManualResolver {
-	m.mu.Lock()
-	defer m.mu.Unlock()
-	if _, ok := m.funcs[t.Name]; ok {
-		for i := range m.order {
-			if m.order[i].Name == t.Name {
-				m.order[i] = t
-			}
-		}
-	} else {
-		m.order = append(m.order, t)
-	}
-	m.funcs[t.Name] = fn
-	return m
-}
-
-// List returns the registered tools.
-func (m *ManualResolver) List(_ context.Context) ([]Tool, error) {
-	m.mu.RLock()
-	defer m.mu.RUnlock()
-	out := make([]Tool, len(m.order))
-	copy(out, m.order)
-	return out, nil
-}
-
-// Call runs the handler registered for name.
-func (m *ManualResolver) Call(ctx context.Context, name string, args map[string]any) (*CallResult, error) {
-	m.mu.RLock()
-	fn, ok := m.funcs[name]
-	m.mu.RUnlock()
-	if !ok {
-		return nil, &RPCError{Code: InvalidParams, Message: "Tool not found: " + name, Data: name}
-	}
-	return fn(ctx, args)
-}
-
-// RegistryResolver auto-discovers tools from registered go-micro services and
-// executes them over RPC. It exposes only services — never the internal
-// store/broker tools.
-type RegistryResolver struct {
-	tools *ai.Tools
-}
-
-// NewRegistryResolver discovers services from reg and calls them with cl.
-func NewRegistryResolver(reg registry.Registry, cl client.Client) *RegistryResolver {
-	return &RegistryResolver{tools: ai.NewTools(reg, ai.ToolClient(cl))}
-}
-
-// List discovers the current service tools.
-func (r *RegistryResolver) List(_ context.Context) ([]Tool, error) {
-	discovered, err := r.tools.Discover()
-	if err != nil {
-		return nil, err
-	}
-	out := make([]Tool, 0, len(discovered))
-	for _, t := range discovered {
-		out = append(out, Tool{
-			Name:        t.Name,
-			Description: t.Description,
-			InputSchema: map[string]interface{}{"type": "object", "properties": t.Properties},
-		})
-	}
-	return out, nil
-}
-
-// Call executes a discovered service tool.
-func (r *RegistryResolver) Call(ctx context.Context, name string, args map[string]any) (*CallResult, error) {
-	res := r.tools.Handler()(ctx, ai.ToolCall{ID: "1", Name: name, Input: args})
-	return &CallResult{Text: res.Content}, nil
-}

gateway/mcp/resolver_test.go

@@ -1,64 +0,0 @@
-package mcp
-
-import (
-	"context"
-	"encoding/json"
-	"net/http"
-	"net/http/httptest"
-	"strings"
-	"testing"
-)
-
-func TestManualResolverHandler(t *testing.T) {
-	res := NewManualResolver().
-		Add(Tool{Name: "echo", Description: "echoes text"},
-			func(_ context.Context, args map[string]interface{}) (*CallResult, error) {
-				s, _ := args["text"].(string)
-				return &CallResult{Text: "you said: " + s}, nil
-			}).
-		Add(Tool{Name: "boom", Description: "errors"},
-			func(_ context.Context, _ map[string]interface{}) (*CallResult, error) {
-				return &CallResult{Text: "kaboom", IsError: true}, nil
-			}).
-		Add(Tool{Name: "blocked", Description: "coded error"},
-			func(_ context.Context, _ map[string]interface{}) (*CallResult, error) {
-				return nil, &RPCError{Code: -32000, Message: "insufficient credits"}
-			})
-
-	ts := httptest.NewServer(NewHandler(res))
-	defer ts.Close()
-	rpc := func(body string) (int, map[string]interface{}) {
-		resp, err := http.Post(ts.URL, "application/json", strings.NewReader(body))
-		if err != nil {
-			t.Fatalf("post rpc: %v", err)
-		}
-		defer resp.Body.Close()
-		var out map[string]interface{}
-		json.NewDecoder(resp.Body).Decode(&out)
-		return resp.StatusCode, out
-	}
-
-	if _, out := rpc(`{"jsonrpc":"2.0","id":1,"method":"tools/list"}`); len(out["result"].(map[string]interface{})["tools"].([]interface{})) != 3 {
-		t.Fatalf("tools/list: %v", out)
-	}
-	// tool result
-	_, out := rpc(`{"jsonrpc":"2.0","id":2,"method":"tools/call","params":{"name":"echo","arguments":{"text":"hi"}}}`)
-	if out["result"].(map[string]interface{})["content"].([]interface{})[0].(map[string]interface{})["text"] != "you said: hi" {
-		t.Fatalf("echo: %v", out)
-	}
-	// tool-level error -> isError result, NOT protocol error
-	_, out = rpc(`{"jsonrpc":"2.0","id":3,"method":"tools/call","params":{"name":"boom","arguments":{}}}`)
-	if out["error"] != nil || out["result"].(map[string]interface{})["isError"] != true {
-		t.Fatalf("boom should be isError result: %v", out)
-	}
-	// coded protocol error -> JSON-RPC error with the code
-	_, out = rpc(`{"jsonrpc":"2.0","id":4,"method":"tools/call","params":{"name":"blocked","arguments":{}}}`)
-	if out["error"] == nil || int(out["error"].(map[string]interface{})["code"].(float64)) != -32000 {
-		t.Fatalf("blocked should be -32000: %v", out)
-	}
-	// notification -> 204, no body
-	code, _ := rpc(`{"jsonrpc":"2.0","method":"notifications/initialized"}`)
-	if code != http.StatusNoContent {
-		t.Fatalf("notification status = %d, want 204", code)
-	}
-}

internal/docs/AGENT_CONFORMANCE.md

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

internal/docs/AGENT_DESIGN.md

@@ -122,19 +122,6 @@ database "agent", table "{name}":
   history    — conversation history
 ```
 
-## Durable Ask / StreamAsk runs
-
-Agents can opt into the same checkpoint backend used by flows with
-`micro.AgentWithCheckpoint(...)`. When enabled, each `Ask` or `StreamAsk` run is
-persisted with its input, terminal status, response, and tool-call records. If
-the process or transport drops after a tool has completed but before the model
-returns a final answer, restart the agent with the same checkpoint store and
-call `micro.AgentResume(ctx, ag, runID)` or
-`micro.AgentResumeStreamAsk(ctx, ag, runID)`.
-Completed tool calls are served from the checkpoint instead of being executed
-again, while guardrails such as `MaxSteps`, loop detection, approval pauses, and
-`request_input` pauses continue to apply to the resumed run.
-
 ## Built-in Capabilities
 
 Beyond its scoped service tools, every agent gets two built-in tools. They are not service endpoints — they are capabilities the agent has over itself and over other agents. They are plain tools wired into the agent's tool handler; there is no separate harness, loop engine, or graph. The LLM calls them exactly like any other tool.

internal/docs/PRIORITIES.md

@@ -19,9 +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.
 
-## Work queue (ranked)
+## Later (ranked)
 
-1. **Emit OpenTelemetry spans for agent runs** ([#3403](https://github.com/micro/go-micro/issues/3403)) — durable checkpoint resume landed via #3414, streaming landed via #3410, and the next highest-value gap is making long-running agent work operable: lifecycle boundaries, model/tool calls, approvals, retries, failures, and cancellation should be correlated by run ID in OpenTelemetry without leaking sensitive payloads.
+No open queued items. The previous top item, **A2A push notifications and
+multi-turn task support** (#3212), is closed; the next architecture-review pass
+should seed a fresh issue-backed item from the roadmap and improvement radar.
 
 _Seeded by Claude Code from the roadmap + open issues; thereafter maintained by the
 architecture-review pass._

internal/harness/provider-conformance/README.md

@@ -1,86 +0,0 @@
-# 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 provider-facing
-agent test and the harnesses in `internal/harness`:
-
-- `agent` — provider tool-call conformance through `agent.Ask`, including run metadata propagation.
-- `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, including `provider-conformance -providers mock`. On the daily schedule and manual dispatch it also runs the live
-provider conformance job. That job:
-
-1. runs the same `agent`, `universe`, `agent-flow`, and `plan-delegate` harness list,
-2. reads the provider keys from repository secrets,
-3. skips providers whose secrets are absent,
-4. fails when any configured provider fails a harness, and
-5. 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/provider-conformance/main.go

@@ -46,7 +46,7 @@ var providerEnv = map[string]string{
 
 func main() {
 	providersFlag := flag.String("providers", "anthropic,openai,gemini,groq,mistral,together,atlascloud", "comma-separated providers to check; use mock for deterministic local checks")
-	harnessesFlag := flag.String("harnesses", "agent,universe,agent-flow,plan-delegate", "comma-separated harness names under internal/harness; agent runs the provider tool-call conformance test")
+	harnessesFlag := flag.String("harnesses", "universe,agent-flow,plan-delegate", "comma-separated harness names under internal/harness")
 	timeoutFlag := flag.Duration("timeout", 10*time.Minute, "timeout per provider/harness run")
 	requireConfiguredFlag := flag.Bool("require-configured", false, "fail when a selected live provider is missing an API key")
 	capabilitiesFlag := flag.Bool("capabilities", true, "print the registered provider capability matrix before running conformance")
@@ -244,9 +244,6 @@ func validateSelection(providers, harnesses []string) error {
 	}
 
 	for _, harness := range harnesses {
-		if harness == "agent" {
-			continue
-		}
 		if strings.Contains(harness, string(os.PathSeparator)) || harness == "." || harness == ".." {
 			return fmt.Errorf("invalid harness name %q", harness)
 		}
@@ -326,10 +323,6 @@ func localRPCEnv(env []string) []string {
 }
 
 func runHarness(provider, harness string, timeout time.Duration) error {
-	if harness == "agent" {
-		return runAgentConformance(provider, timeout)
-	}
-
 	ctx, cancel := context.WithTimeout(context.Background(), timeout)
 	defer cancel()
 
@@ -365,28 +358,3 @@ func runHarness(provider, harness string, timeout time.Duration) error {
 	}
 	return nil
 }
-
-func runAgentConformance(provider string, timeout time.Duration) error {
-	ctx, cancel := context.WithTimeout(context.Background(), timeout)
-	defer cancel()
-
-	testProvider := provider
-	if provider == "mock" {
-		testProvider = "fake"
-	}
-	cmd := exec.CommandContext(ctx, "go", "test", "./agent", "-run", "TestAgentProviderConformanceMatrix", "-count=1", "-v")
-	cmd.Dir = repoRoot()
-	cmd.Stdout = os.Stdout
-	cmd.Stderr = os.Stderr
-	cmd.Env = localRPCEnv(append(os.Environ(),
-		"GO_MICRO_AGENT_CONFORMANCE_LIVE=1",
-		"GO_MICRO_AGENT_CONFORMANCE_PROVIDERS="+testProvider,
-	))
-	if err := cmd.Run(); err != nil {
-		if ctx.Err() == context.DeadlineExceeded {
-			return fmt.Errorf("timed out after %s", timeout)
-		}
-		return err
-	}
-	return nil
-}

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

@@ -1,38 +0,0 @@
-# 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 four 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.
-4. **Deploy** — `micro deploy --dry-run <target>` remains available as the
-   deployment-boundary checkpoint. The dry run resolves configured deploy targets
-   and services and prints the remote build/copy/systemd/health plan without
-   building binaries, opening SSH connections, running `rsync`, or touching
-   remote infrastructure.
-
-After the CLI boundary smoke checks, the script runs the deterministic harnesses
-that boot real services, agents, workflows, store-backed run history, plan/delegate,
-and A2A with only the LLM mocked.
-
-## Local and CI entry points
-
-The default GitHub harness workflow runs this script on every push and pull
-request after the 0→1 scaffold contract. Developers can run the same no-secret
-contract locally with:
-
-```sh
-make harness
-```
-
-That target intentionally exercises both 0→1 scaffold variants, the 0→hero
-scenario, the event-driven agent-flow harness, and mock provider conformance, so
-the public scaffold → run/chat → inspect → deploy lifecycle stays executable
-outside CI as well. Live provider checks remain separate and gated by configured
-API keys (`make provider-conformance` or the scheduled/manual CI job).

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

@@ -1,14 +0,0 @@
-#!/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
-go test ./cmd/micro/cli/deploy -run TestDeployDryRun -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

@@ -69,18 +69,10 @@ A card looks like:
   "capabilities": { "streaming": true, "pushNotifications": true },
   "defaultInputModes": ["text/plain"],
   "defaultOutputModes": ["text/plain"],
-  "skills": [
-    { "id": "task", "name": "Task", "tags": ["task"] },
-    { "id": "project", "name": "Project", "tags": ["project"] }
-  ]
+  "skills": [{ "id": "chat", "name": "Chat", "tags": ["task", "project"] }]
 }
 ```
 
-Each managed service is advertised as its own typed skill. Clients can call the
-whole agent at `/agents/task-mgr`, or address one skill directly at
-`/agents/task-mgr/skills/task`; the skill endpoint serves a focused card and
-routes the request to the same agent with that skill selected.
-
 ## Calling an agent
 
 A2A uses JSON-RPC 2.0 over HTTP. Send a message with `message/send`; the gateway runs the agent and returns a completed `Task`:
@@ -180,11 +172,17 @@ 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. 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.
+
+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.
 
 ## See also
 

internal/website/docs/guides/agent-harness.md

@@ -36,46 +36,13 @@ your stack — the harness *is* the stack.
 | Discovery & RPC | Registry + client; agents and services find and call each other | Shipped |
 | Interop | MCP (tools), A2A (agents), x402 (paid tools) | Shipped |
 | Resilience | Per-call timeout with context propagation; opt-in retry/backoff (`ModelRetry`) across the loop | Shipped |
-| Durable runs | Checkpoint and resume an agent run with the same checkpoint backend flows use | Shipped |
+| Durable runs | Checkpoint and resume an agent run (flows already do) | In progress |
 | Observability | `RunInfo` → OpenTelemetry spans for runs, model calls, tools, delegation, and failures; persisted run history | Shipped |
 | Streaming | `ai.Stream` through chat, agent, and A2A | In progress |
 
 The "in progress" rows are exactly the roadmap's [Now and Next](/docs/roadmap.html),
 and the work is happening in the open.
 
-## Durable agent runs
-
-Agents can persist their execution history to the same `Checkpoint` backend as
-flows. A checkpointed `Ask` records the run id, original prompt, model result,
-and completed tool calls. If the process restarts after a tool succeeds but
-before the model finishes, `AgentResume` continues the same run and returns the
-recorded tool result instead of re-running the side effect. If a run already
-completed, resume returns the persisted response without calling the model.
-
-```go
-agent := micro.NewAgent("conductor",
-    micro.AgentProvider("anthropic"),
-    micro.AgentWithCheckpoint(checkpoint),
-)
-
-resp, err := agent.Ask(ctx, "charge order 42 and send a receipt")
-if err != nil {
-    // On startup, or after a transient failure, discover unfinished work:
-    pending, _ := micro.AgentPending(ctx, agent)
-    for _, run := range pending {
-        _, _ = micro.AgentResume(ctx, agent, run.ID)
-    }
-}
-_ = resp
-```
-
-For human-in-the-loop runs that pause through the built-in `request_input` tool,
-resume with the operator's response:
-
-```go
-_, err := micro.AgentResumeInput(ctx, agent, runID, "Deploy to us-east-1")
-```
-
 ## Observing agent runs
 
 Pass an OpenTelemetry tracer provider when you construct an agent to turn the

internal/website/docs/guides/agents-and-workflows.md

@@ -83,44 +83,6 @@ a := micro.NewAgent("conductor",
 a.Ask(ctx, "Plan the launch, create the tasks, and have comms notify the owner.")
 ```
 
-### Long-running memory
-
-Agents use store-backed conversation memory by default, scoped under the agent's
-name. That makes short restarts boring: the next `Ask` reloads the retained
-history from the same store backend you already use for services and flows.
-Long-running agents can also keep model context bounded without losing useful
-prior context:
-
-```go
-a := micro.NewAgent("conductor",
-    micro.AgentServices("task"),
-    micro.AgentProvider("anthropic"),
-    micro.AgentCompactMemory(40, 12),      // max active messages, recent messages kept verbatim
-    micro.AgentMemoryRecallLimit(5),       // archived turns recalled per Ask
-)
-```
-
-`AgentCompactMemory(maxMessages, keepRecent)` switches the default memory to a
-deterministic compactor. Once active history grows past `maxMessages`, older
-turns move into the durable archive, a provider-neutral summary is injected into
-active context, and the newest `keepRecent` messages stay verbatim. On future
-asks, archived turns whose text matches the current request are recalled ahead of
-the active context. The built-in retrieval is intentionally simple and
-credential-free for CI; teams that need embeddings or a vector database can still
-provide their own `AgentMemory` implementation.
-
-This is harness memory, not prompt-layer orchestration: services remain the
-capabilities, agents remain the dynamic decision makers, and flows remain the
-durable predefined paths. Compaction only keeps a scheduled or looping agent from
-turning every past turn into model context while still letting it remember facts
-that matter to the current service → agent → workflow run.
-
-Checkpointed agent runs and compacted memory share the same store-backed shape.
-If a provider call fails after the prompt has been recorded, `agent.Resume` uses
-the checkpointed run id and does not append that same user turn a second time;
-completed tool results and recalled archived memory remain available for the
-retry.
-
 ## The patterns — most are already here
 
 Anthropic lists five workflow patterns. Go Micro implements the two richest ones natively, as services and tools, and the rest are ordinary compositions:

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

@@ -39,12 +39,12 @@ The built-in providers currently register these capability interfaces:
 | Provider | Chat/text (`ai.Model`) | Image (`ai.ImageModel`) | Video (`ai.VideoModel`) | Streaming (`ai.Stream`) |
 | --- | --- | --- | --- | --- |
 | `anthropic` | Yes | No | No | No |
-| `atlascloud` | Yes | Yes | Yes | Yes |
+| `atlascloud` | Yes | Yes | Yes | No |
 | `gemini` | Yes | No | No | No |
-| `groq` | Yes | No | No | Yes |
-| `mistral` | Yes | No | No | Yes |
+| `groq` | Yes | No | No | No |
+| `mistral` | Yes | No | No | No |
 | `openai` | Yes | Yes | No | Yes |
-| `together` | Yes | No | No | Yes |
+| `together` | Yes | No | No | No |
 
 ## Step 1: Implement the `ai.Model` Interface
 

internal/website/docs/roadmap.md

@@ -36,13 +36,14 @@ The priority is that what exists works everywhere, under real conditions.
 
 ## Next — agentic depth
 
+- **Durable agent loop.** Flows resume; the agent's own loop does not yet. Reuse `Checkpoint` so a long-running agent survives a restart and continues.
 - **Streaming.** Broaden provider-backed `ai.Stream` coverage and keep chat plus A2A `message/stream` working end to end for real chat and long-task UX.
 - **Agent observability.** Wire the new `RunInfo` into OpenTelemetry spans so a run — steps, tool calls, delegation — is traceable. This is also what anyone running it in production will need.
 
 ## Later
 
 - **Memory management** — summarization and retrieval (RAG) beyond a fixed buffer.
-- **Human-in-the-loop** — broaden pause/resume UX around `input-required` runs and approvals.
+- **Human-in-the-loop** — pause and resume mid-run (`input-required`), beyond the binary `ApproveTool` gate.
 - **A2A** — richer live-stream reconnection (`tasks/resubscribe`) and `input-required` handoffs.
 
 ## Developer experience (ongoing)

micro.go

@@ -23,21 +23,12 @@ type Service = service.Service
 // Agent is the interface for an AI agent that manages services.
 type Agent = agent.Agent
 
-// AgentResponse is what an agent returns from Ask or a resumed run.
-type AgentResponse = agent.Response
-
-// AgentStream is a stream of tool execution events followed by final-answer chunks.
-type AgentStream = agent.AgentStream
-
 // AgentOption configures an Agent.
 type AgentOption = agent.Option
 
 // Flow is an event-driven LLM orchestration unit.
 type Flow = flow.Flow
 
-// FlowRun is a checkpointed flow or agent run record.
-type FlowRun = flow.Run
-
 // FlowOption configures a Flow.
 type FlowOption = flow.Option
 
@@ -173,32 +164,6 @@ func AgentWrapTool(w ...ai.ToolWrapper) AgentOption {
 // tool calls, delegation, and failures.
 func AgentTraceProvider(tp trace.TracerProvider) AgentOption { return agent.TraceProvider(tp) }
 
-// AgentWithCheckpoint sets the durability backend for agent Ask runs.
-// It uses the same Checkpoint interface as flows so services, agents,
-// and workflows can share one execution history backend.
-func AgentWithCheckpoint(c Checkpoint) AgentOption { return agent.WithCheckpoint(c) }
-
-// AgentPending returns checkpointed agent runs that have not completed.
-// Use it at process startup to discover agent work that should be resumed.
-func AgentPending(ctx context.Context, a Agent) ([]FlowRun, error) { return agent.Pending(ctx, a) }
-
-// AgentResume resumes a checkpointed agent run by id. Completed runs return
-// the persisted response without calling the model or replaying tool calls.
-func AgentResume(ctx context.Context, a Agent, runID string) (*AgentResponse, error) {
-	return agent.Resume(ctx, a, runID)
-}
-
-// AgentResumeInput resumes a checkpointed agent run waiting for human input.
-func AgentResumeInput(ctx context.Context, a Agent, runID, input string) (*AgentResponse, error) {
-	return agent.ResumeInput(ctx, a, runID, input)
-}
-
-// AgentResumeStreamAsk resumes a checkpointed agent run by id and streams the
-// resulting tool events and final answer.
-func AgentResumeStreamAsk(ctx context.Context, a Agent, runID string) (AgentStream, error) {
-	return agent.ResumeStreamAsk(ctx, a, runID)
-}
-
 // NewFlow creates an event-driven LLM orchestration unit.
 //
 //	f := micro.NewFlow("onboard-user",