Add A2A completed-task streaming

.github/workflows/architecture-review.yml

@@ -1,24 +1,21 @@
 name: Architecture Review
 
-# Continuous high-altitude oversight of the whole framework and harness — the
-# "founder lens" of the autonomous loop (internal/docs/CONTINUOUS_IMPROVEMENT.md).
-# Where DevRel watches the public story and the increment loop ships code, the
-# architect watches the SYSTEM and runs alongside the builders: it tracks what is
-# in flight and what just merged, keeps the roadmap priorities live, and judges
-# cohesion (harness <-> framework <-> dev UX), missing pieces, and realignment.
+# Periodic high-altitude review of the whole framework and harness against the
+# North Star (internal/docs/THESIS.md) — part of the autonomous loop
+# (internal/docs/CONTINUOUS_IMPROVEMENT.md). Where DevRel watches the public
+# story, the architect watches the system: API coherence, lifecycle gaps, and
+# whether recent increments are converging on the thesis or sprawling.
 #
-# Its OUTPUT is the ranked queue in internal/docs/PRIORITIES.md plus an assessment
-# — NOT large refactors. Breaking public-API and architectural changes stay with
-# the human (see CONTINUOUS_IMPROVEMENT.md).
+# The architect's OUTPUT is an assessment plus scoped follow-up issues that feed
+# the hourly increment loop — NOT large refactors. Breaking public-API and
+# architectural changes stay with the human (see CONTINUOUS_IMPROVEMENT.md).
 #
-# Runs hourly, offset before the increment loop (:29) so it re-prioritizes and
-# THEN the loop builds the new top of the queue. Opens a fresh issue and
-# dispatches Codex via CODEX_TRIGGER_TOKEN.
+# Opens a fresh issue and dispatches Codex via CODEX_TRIGGER_TOKEN.
 
 on:
   workflow_dispatch: {}
   schedule:
-    - cron: "59 * * * *" # hourly at :59, just before the :29 increment run (tunable)
+    - cron: "0 8 */3 * *" # roughly every 3 days, 08:00 UTC (tunable)
 
 permissions:
   issues: write
@@ -44,8 +41,8 @@ jobs:
           fi
           ISSUE_URL=$(gh issue create --repo "$REPO" \
             --title "Architecture review #$RUN_NUMBER" \
-            --body "Continuous architecture / harness oversight against the North Star in internal/docs/THESIS.md. Output: a re-ranked internal/docs/PRIORITIES.md (only if it changed) plus an assessment.")
+            --body "Periodic architecture / harness review against the North Star in internal/docs/THESIS.md. Output: an assessment plus scoped follow-up issues for the increment loop.")
           ISSUE_NUM="${ISSUE_URL##*/}"
           echo "Opened issue #$ISSUE_NUM — dispatching Codex (Architect)."
           gh issue comment "$ISSUE_NUM" --repo "$REPO" --body \
-          "@codex Act as the architect — the founder lens — for go-micro, running continuously alongside the builders. Hold the whole picture: how the harness, the framework, and the developer UX fit together cohesively, what is in flight and what just merged, what to prioritize next on the roadmap, and what is missing or has drifted. Each run: (1) TRACK STATE — scan recently merged PRs and open codex PRs/issues to see what shipped and what is being built right now, so the queue reflects reality (drop done items, don't re-queue in-flight work). (2) ASSESS against the North Star in internal/docs/THESIS.md — lead with its Mission (*the problem we solve: make building an agent as easy as building a service, on one runtime*) and re-derive alignment from the CANON it names (the blog under internal/website/blog, the README, and the website — read these, don't rely on THESIS.md alone), then ROADMAP.md (Now → Next → Later). Judge every priority against the mission: does it make the services → agents → workflows lifecycle simpler, more cohesive, and more operable? Look at coherence and seams across the core packages (agent, ai, flow, gateway/mcp, gateway/a2a, model, server, store, registry), the dev inner loop (scaffold → run → chat → inspect → deploy), missing pieces, duplication/drift, and realignment. Flag drift in EITHER direction: work drifting from the mission, or the North Star/website drifting from the lived story in the blog (which needs re-grounding in the canon). (3) MAINTAIN THE QUEUE in internal/docs/PRIORITIES.md — a SINGLE ordered list, highest-value first, each item linking a scoped CI-verifiable issue (#N); roadmap phase is the primary ordering, internal findings (cohesion gaps, DX friction, missing pieces) interleaved by value. For any prioritized gap that has no issue yet, file one: \`gh issue create --label codex --label enhancement --title \"<scoped task>\" --body \"<goal, scope, acceptance criteria>\"\`. OUTPUT: post a concise assessment as a comment on this issue (#$ISSUE_NUM) — what shipped, what's in flight, the top risks/gaps/missing pieces, and the reasoning behind the ranking. If the ranking actually changed, open ONE PR for PRIORITIES.md: \`git switch -c codex/architect-$ISSUE_NUM\`, \`git push -u origin codex/architect-$ISSUE_NUM\`, \`gh pr create --base master --label codex --title \"<title>\" --body \"<summary, Closes #$ISSUE_NUM>\"\`, then \`gh pr merge --squash --auto --delete-branch\`. If the queue is already accurate and correctly ranked, do NOT open a PR — just close this issue (\`gh issue close $ISSUE_NUM\`). Do NOT make breaking public-API or architectural changes yourself — surface those in the assessment as notes for the human, never as auto-merged changes. Do not use the make_pr tool (it is a no-op stub)."
+          "@codex Act as the architect for go-micro. Review the overall framework and harness against the North Star in internal/docs/THESIS.md (services → agents → workflows as one runtime) and the roadmap in ROADMAP.md. Assess: API coherence and consistency across the core packages (agent, ai, flow, gateway/mcp, gateway/a2a, model, server, store, registry), gaps or missing pieces in the services → agents → workflows lifecycle, duplication or drift, and whether recent increments (scan recently merged PRs) are converging on the thesis or sprawling. Then: (A) post a concise architectural assessment as a comment on this issue (#$ISSUE_NUM) — strengths, the top risks/gaps, and a recommended direction for the next increments; (B) file concrete, scoped follow-up issues for the highest-value gaps so the hourly increment loop can pick them up — \`gh issue create --label codex --label enhancement --title \"<scoped task>\" --body \"<goal, scope, acceptance criteria>\"\` (each must be a single, self-contained, CI-verifiable chunk). Do NOT make breaking public-API or architectural changes yourself — your output is the assessment and the issues. A small, safe doc/comment correction may be a PR (\`git switch -c codex/architect-$ISSUE_NUM\` … \`gh pr create --base master --label codex …\` … \`gh pr merge --squash --auto --delete-branch\`). Do not use the make_pr tool (it is a no-op stub)."

.github/workflows/continuous-improvement.yml

@@ -62,4 +62,4 @@ jobs:
           ISSUE_NUM="${ISSUE_URL##*/}"
           echo "Opened issue #$ISSUE_NUM — dispatching Codex."
           gh issue comment "$ISSUE_NUM" --repo "$REPO" --body \
-          "@codex Run one continuous-improvement increment per internal/docs/CONTINUOUS_IMPROVEMENT.md, aligned to the North Star in internal/docs/THESIS.md (the holistic services → agents → workflows lifecycle). PICK THE WORK FROM THE QUEUE: read internal/docs/PRIORITIES.md and take the highest-ranked item whose linked issue is still OPEN — that is your task, and its issue number is the one you close. (If PRIORITIES.md is missing or every listed item's issue is already closed, fall back to picking the single highest-value roadmap/issue/improvement-radar item yourself.) Implement it, and verify \`go build ./...\`, \`go test ./...\`, and \`golangci-lint run ./...\`. Then open the PR YOURSELF from the shell — do NOT use the make_pr tool (in this environment it only records metadata and never creates a PR). Create a uniquely-named branch under the codex/ prefix and open the PR from it: \`git switch -c codex/increment-$ISSUE_NUM\`, then \`git push -u origin codex/increment-$ISSUE_NUM\`, then \`gh pr create --base master --label codex --title \"<title>\" --body \"<body; include 'Closes #<the priority issue you built>' so it leaves the queue, and 'Closes #$ISSUE_NUM' for this run's tracker>\"\`. Finally enable auto-merge so GitHub merges it once CI is green: \`gh pr merge --squash --auto --delete-branch\`. The gh CLI is installed and authenticated and origin points to $REPO. One concern per PR; stay out of brand/positioning copy and breaking public API."
+          "@codex Run one continuous-improvement increment per internal/docs/CONTINUOUS_IMPROVEMENT.md, aligned to the North Star in internal/docs/THESIS.md (the holistic services → agents → workflows lifecycle). Pick the single highest-value roadmap/issue/improvement-radar item that advances that thesis, implement it, and verify \`go build ./...\`, \`go test ./...\`, and \`golangci-lint run ./...\`. Then open the PR YOURSELF from the shell — do NOT use the make_pr tool (in this environment it only records metadata and never creates a PR). Create a uniquely-named branch under the codex/ prefix and open the PR from it: \`git switch -c codex/increment-$ISSUE_NUM\`, then \`git push -u origin codex/increment-$ISSUE_NUM\`, then \`gh pr create --base master --label codex --title \"<title>\" --body \"<body, including 'Closes #$ISSUE_NUM'>\"\`. Finally enable auto-merge so GitHub merges it once CI is green: \`gh pr merge --squash --auto --delete-branch\`. The gh CLI is installed and authenticated and origin points to $REPO. One concern per PR; stay out of brand/positioning copy and breaking public API."

.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:
@@ -33,8 +33,8 @@ jobs:
         run: go run ./internal/harness/universe
       - name: Agent-flow harness
         run: go run ./internal/harness/agent-flow
-      - name: 0→hero run/chat/inspect reference scenario
-        run: ./internal/harness/zero-to-hero-ci/run.sh
+      - name: 0→hero plan-delegate workflow harness
+        run: go run ./internal/harness/plan-delegate
 
   harness-live:
     name: Provider harnesses (live LLM conformance)
@@ -50,17 +50,6 @@ jobs:
         with:
           go-version: stable
           cache: true
-      - name: Agent provider conformance matrix
-        env:
-          GO_MICRO_AGENT_CONFORMANCE_LIVE: "1"
-          ANTHROPIC_API_KEY: ${{ secrets.ANTHROPIC_API_KEY }}
-          OPENAI_API_KEY: ${{ secrets.OPENAI_API_KEY }}
-          GEMINI_API_KEY: ${{ secrets.GEMINI_API_KEY }}
-          GROQ_API_KEY: ${{ secrets.GROQ_API_KEY }}
-          MISTRAL_API_KEY: ${{ secrets.MISTRAL_API_KEY }}
-          TOGETHER_API_KEY: ${{ secrets.TOGETHER_API_KEY }}
-          ATLASCLOUD_API_KEY: ${{ secrets.ATLASCLOUD_API_KEY }}
-        run: go test ./agent -run TestAgentProviderConformanceMatrix -count=1 -v
       - name: Provider conformance against live models
         env:
           ANTHROPIC_API_KEY: ${{ secrets.ANTHROPIC_API_KEY }}
@@ -72,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. v1 is the synchronous JSON-RPC binding (`message/send`, `tasks/get`, card discovery); streaming and push notifications are advertised as 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)

README.md

@@ -237,7 +237,7 @@ Just as a service composes pluggable abstractions (registry, broker, store), an
 ```go
 agent := micro.NewAgent("assistant",
     micro.AgentProvider("anthropic"),                 // model — swap the provider
-    micro.AgentCompactMemory(40, 12),                 // memory — durable, summarized, recallable
+    micro.AgentMemory(micro.NewInMemory(50)),         // memory — default is store-backed & durable
     micro.AgentTool("weather", "Get the weather for a city",
         map[string]any{"city": map[string]any{"type": "string"}},
         func(ctx context.Context, in map[string]any) (string, error) {
@@ -247,7 +247,7 @@ agent := micro.NewAgent("assistant",
 )
 ```
 
-**Memory** is durable and store-backed by default (Postgres, NATS KV, or file), so an agent picks up where it left off after a restart — or supply your own with `AgentMemory`. Long-running agents can opt into `AgentCompactMemory(maxMessages, keepRecent)`: older turns are collapsed into a deterministic summary, recent turns stay verbatim, and relevant archived turns are recalled on future asks without replaying the whole conversation. **Tools** are your services automatically, plus any function you register with `AgentTool`.
+**Memory** is durable and store-backed by default (Postgres, NATS KV, or file), so an agent picks up where it left off after a restart — or supply your own with `AgentMemory`. **Tools** are your services automatically, plus any function you register with `AgentTool`.
 
 ### Paid tools (x402)
 

ROADMAP.md

@@ -15,8 +15,7 @@ The full, current roadmap lives at **[go-micro.dev/docs/roadmap](https://go-micr
 ## Where we are (v6)
 
 Services, agents (`plan`/`delegate`, guardrails, memory, tool middleware), durable
-flows, the MCP and A2A gateways (both directions, including A2A streaming,
-push notifications, and multi-turn continuation), x402 paid tools, secure by
+flows, the MCP and A2A gateways (both directions), x402 paid tools, secure by
 default.
 
 ## Principles
@@ -42,14 +41,14 @@ default.
 ## Next — agentic depth
 
 - **Durable agent loop** — resume a long run via `Checkpoint` (flows already do).
-- **Streaming** — broaden provider-backed `ai.Stream` coverage and keep chat/A2A streaming end to end.
+- **Streaming** — `ai.Stream` + A2A `message/stream`, end to end.
 - **Agent observability** — `RunInfo` → OpenTelemetry spans.
 
 ## Later
 
 - Memory management (summarization, retrieval/RAG); human-in-the-loop pause/resume;
-  richer A2A live-stream reconnection (`tasks/resubscribe`) and `input-required`
-  handoffs.
+  x402 live-facilitator conformance and paid remote tools with spend caps; A2A
+  streaming, push notifications, and multi-turn tasks.
 
 ## Developer experience (ongoing)
 

agent/agent.go

@@ -23,7 +23,6 @@ import (
 	"github.com/google/uuid"
 	pb "go-micro.dev/v6/agent/proto"
 	"go-micro.dev/v6/ai"
-	"go-micro.dev/v6/flow"
 	"go-micro.dev/v6/gateway/a2a"
 	"go-micro.dev/v6/server"
 	"go-micro.dev/v6/store"
@@ -43,7 +42,6 @@ type Agent interface {
 	Init(...Option)
 	Options() Options
 	Ask(ctx context.Context, message string) (*Response, error)
-	Stream(ctx context.Context, message string) (ai.Stream, error)
 	Run() error
 	Stop() error
 	String() string
@@ -87,16 +85,6 @@ type agentImpl struct {
 	// Both are surfaced to tool wrappers via ai.RunInfo on the context.
 	runID       string
 	parentRunID string
-
-	// pause records a guardrail approval pause raised during the current
-	// 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.
@@ -160,8 +148,6 @@ func (a *agentImpl) setup() {
 		a.mem = a.opts.Memory
 	case a.ephemeral:
 		a.mem = NewInMemory(a.opts.HistoryLimit)
-	case a.opts.MemoryCompaction.MaxMessages > 0:
-		a.mem = NewCompactingMemory(a.stateStore(), "history", a.opts.MemoryCompaction.MaxMessages, a.opts.MemoryCompaction.KeepRecent)
 	default:
 		a.mem = NewMemory(a.stateStore(), "history", a.opts.HistoryLimit)
 	}
@@ -185,38 +171,6 @@ func (a *agentImpl) Ask(ctx context.Context, message string) (*Response, error)
 	return a.ask(ctx, message, a.parentRunID)
 }
 
-// Stream sends a message and returns a streaming model response. Tool-calling
-// agent runs still use Ask; Stream is for chat turns where immediate token
-// delivery is more important than tool orchestration.
-func (a *agentImpl) Stream(ctx context.Context, message string) (ai.Stream, error) {
-	a.mu.Lock()
-	defer a.mu.Unlock()
-	if a.model == nil {
-		a.setup()
-	}
-	toolList, err := a.discoverTools()
-	if err != nil {
-		return nil, fmt.Errorf("discover tools: %w", err)
-	}
-	a.mem.Add("user", message)
-	return a.model.Stream(ctx, &ai.Request{
-		Prompt:       message,
-		SystemPrompt: a.buildPrompt(),
-		Tools:        toolList,
-		Messages:     a.mem.Messages(),
-	})
-}
-
-// 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) {
-	a, ok := ag.(*agentImpl)
-	if !ok {
-		return nil, fmt.Errorf("agent pending: unsupported agent implementation %T", ag)
-	}
-	return a.pending(ctx)
-}
-
 func (a *agentImpl) ask(ctx context.Context, message, parentRunID string) (*Response, error) {
 	a.mu.Lock()
 	defer a.mu.Unlock()
@@ -225,10 +179,6 @@ func (a *agentImpl) ask(ctx context.Context, message, parentRunID string) (*Resp
 		a.setup()
 	}
 
-	return a.askLocked(ctx, uuid.New().String(), message, parentRunID, nil)
-}
-
-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)
@@ -237,70 +187,30 @@ func (a *agentImpl) askLocked(ctx context.Context, runID, message, parentRunID s
 	a.mem.Add("user", message)
 	a.steps = 0
 	a.calls = map[string]int{}
-	a.pause = nil
 
 	// Correlate this run's tool calls and surface lineage to wrappers.
-	a.runID = runID
+	a.runID = uuid.New().String()
 	ctx = ai.WithRunInfo(ctx, ai.RunInfo{
 		RunID:    a.runID,
 		ParentID: parentRunID,
 		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
-	}
 	ctx, endRun := a.startRun(ctx, message)
 	defer func() { endRun(err) }()
 
-	messages := a.mem.Messages()
-	if recall, ok := a.mem.(MemoryRecall); ok && a.opts.MemoryRecallLimit > 0 {
-		if recalled := recall.Recall(message, a.opts.MemoryRecallLimit); len(recalled) > 0 {
-			messages = append([]ai.Message{{
-				Role:    "system",
-				Content: "Relevant recalled memory follows; use it as durable prior context without assuming the whole conversation was replayed.",
-			}}, append(recalled, messages...)...)
-		}
-	}
-
 	resp, err := ai.GenerateWithRetry(ctx, a.model, &ai.Request{
 		Prompt:       message,
 		SystemPrompt: a.buildPrompt(),
 		Tools:        toolList,
-		Messages:     messages,
+		Messages:     a.mem.Messages(),
 	}, ai.GeneratePolicy{
 		Timeout:     a.opts.ModelTimeout,
 		MaxAttempts: a.opts.ModelMaxAttempts,
 		Backoff:     a.opts.ModelRetryBackoff,
 	})
 	if err != nil {
-		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
 	}
-	if a.pause != nil && a.opts.Checkpoint != nil {
-		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
-		if err := a.saveRun(ctx, run); err != nil {
-			return nil, err
-		}
-		return nil, fmt.Errorf("agent run %s paused for approval: %s", run.ID, a.pause.Message)
-	}
 
 	if resp.Reply != "" {
 		a.mem.Add("assistant", resp.Reply)
@@ -317,31 +227,13 @@ func (a *agentImpl) askLocked(ctx context.Context, runID, message, parentRunID s
 		reply += resp.Answer
 	}
 
-	res := &Response{
+	return &Response{
 		Reply:     reply,
 		ToolCalls: resp.ToolCalls,
 		Agent:     a.opts.Name,
 		RunID:     a.runID,
 		ParentID:  parentRunID,
-	}
-	run.Status = "done"
-	run.State.Stage = ""
-	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
-	if err := a.saveRun(ctx, run); err != nil {
-		return nil, err
-	}
-	return res, nil
+	}, nil
 }
 
 // Chat implements the proto AgentHandler interface for RPC.
@@ -395,13 +287,13 @@ func (a *agentImpl) Run() error {
 	// Ask in-process — no separate gateway needed to be queried by URL.
 	if a.opts.A2AAddress != "" {
 		card := a2a.Card(a.opts.Name, "http://localhost"+a.opts.A2AAddress, "", a.opts.Services)
-		handler := a2a.NewAgentStreamHandler(card, func(ctx context.Context, text string) (string, error) {
+		handler := a2a.NewAgentHandler(card, func(ctx context.Context, text string) (string, error) {
 			resp, err := a.Ask(ctx, text)
 			if err != nil {
 				return "", err
 			}
 			return resp.Reply, nil
-		}, a.Stream)
+		})
 		go func() {
 			if err := http.ListenAndServe(a.opts.A2AAddress, handler); err != nil {
 				fmt.Printf("agent %s A2A server: %v\n", a.opts.Name, err)

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
@@ -118,9 +102,8 @@ func (a *agentImpl) toolHandler() ai.ToolHandler {
 
 	// Innermost first: base, then guardrails (approve → loop → step →
 	// plan), then developer wrappers outermost. Wrapping reverses order,
-	// so the result runs plan → step → loop → approve → checkpoint → base.
+	// so the result runs plan → step → loop → approve → base.
 	h := a.baseHandler()
-	h = a.checkpointToolWrap(h)
 	h = a.approveWrap(h)
 	h = a.loopWrap(h)
 	h = a.stepWrap(h)
@@ -162,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)
 		}
@@ -220,16 +200,6 @@ func (a *agentImpl) loopWrap(next ai.ToolHandler) ai.ToolHandler {
 }
 
 // approveWrap gates each action before it runs (ApproveTool).
-type approvalPause struct {
-	Tool    string
-	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 {
@@ -238,7 +208,6 @@ func (a *agentImpl) approveWrap(next ai.ToolHandler) ai.ToolHandler {
 				if reason != "" {
 					msg += ": " + reason
 				}
-				a.pause = &approvalPause{Tool: call.Name, Message: msg}
 				return refused(call.ID, ai.RefusedApproval, msg)
 			}
 		}
@@ -257,17 +226,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

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

@@ -1,218 +0,0 @@
-package agent
-
-import (
-	"context"
-	"encoding/json"
-	"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 {
-	now := time.Now()
-	run := flow.Run{
-		ID:       runID,
-		ParentID: parentRunID,
-		Flow:     a.opts.Name,
-		State:    flow.State{Stage: agentAskStep, Data: []byte(message)},
-		Steps:    []flow.StepRecord{{Name: agentAskStep, Status: "in_progress"}},
-		Status:   "running",
-		Started:  now,
-		Updated:  now,
-	}
-	if existing != nil {
-		run = *existing
-		run.Status = "running"
-		run.State.Stage = agentAskStep
-		if len(run.Steps) == 0 {
-			run.Steps = []flow.StepRecord{{Name: agentAskStep}}
-		}
-		run.Steps[0].Status = "in_progress"
-		run.Steps[0].Error = ""
-		run.Steps[0].Result = ""
-	}
-	return run
-}
-
-func (a *agentImpl) saveRun(ctx context.Context, run flow.Run) error {
-	if a.opts.Checkpoint == nil {
-		return nil
-	}
-	if err := a.opts.Checkpoint.Save(ctx, run); err != nil {
-		return fmt.Errorf("agent %s checkpoint save: %w", a.opts.Name, err)
-	}
-	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)
-	}
-	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" {
-		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
-	}
-	if run.Status == "done" {
-		var resp Response
-		if err := json.Unmarshal(run.State.Data, &resp); err != nil {
-			return nil, fmt.Errorf("agent run %s response decode: %w", runID, err)
-		}
-		return &resp, nil
-	}
-	message := string(run.State.Data)
-	parentID := run.ParentID
-	a.mu.Lock()
-	defer a.mu.Unlock()
-	if a.model == nil {
-		a.setup()
-	}
-	return a.askLocked(ctx, run.ID, message, parentID, &run)
-}
-
-// ResumeInput resumes a checkpointed agent run that paused via the built-in
-// request_input tool. The supplied input is appended to the original request so
-// the same run can continue with durable checkpoint and completed tool history.
-func ResumeInput(ctx context.Context, ag Agent, runID, input string) (*Response, error) {
-	a, ok := ag.(*agentImpl)
-	if !ok {
-		return nil, fmt.Errorf("agent resume input: unsupported agent implementation %T", ag)
-	}
-	return a.resumeInput(ctx, runID, input)
-}
-
-func (a *agentImpl) resumeInput(ctx context.Context, runID, input string) (*Response, error) {
-	if a.opts.Checkpoint == nil {
-		return nil, fmt.Errorf("agent %s has no checkpoint configured", a.opts.Name)
-	}
-	run, ok, err := a.opts.Checkpoint.Load(ctx, runID)
-	if err != nil {
-		return nil, err
-	}
-	if !ok {
-		return nil, fmt.Errorf("agent run %s not found", runID)
-	}
-	if run.Status != "paused" || run.State.Stage != agentInputStep {
-		return nil, fmt.Errorf("agent run %s is not waiting for human input", runID)
-	}
-	var p inputPause
-	if err := run.State.Scan(&p); err != nil {
-		return nil, fmt.Errorf("agent run %s input state decode: %w", runID, err)
-	}
-	message := p.OriginalMessage
-	if message == "" {
-		message = string(run.State.Data)
-	}
-	message += "\n\nHuman input: " + input
-	run.Status = "running"
-	run.State.Stage = agentAskStep
-	run.State.Data = []byte(message)
-	a.mu.Lock()
-	defer a.mu.Unlock()
-	if a.model == nil {
-		a.setup()
-	}
-	return a.askLocked(ctx, run.ID, message, run.ParentID, &run)
-}
-
-func (a *agentImpl) pending(ctx context.Context) ([]flow.Run, error) {
-	if a.opts.Checkpoint == nil {
-		return nil, nil
-	}
-	runs, err := a.opts.Checkpoint.List(ctx)
-	if err != nil {
-		return nil, err
-	}
-	out := runs[:0]
-	for _, run := range runs {
-		if run.Flow == a.opts.Name && run.Status != "done" {
-			out = append(out, run)
-		}
-	}
-	return out, nil
-}
-
-func (a *agentImpl) checkpointToolWrap(next ai.ToolHandler) ai.ToolHandler {
-	return func(ctx context.Context, call ai.ToolCall) ai.ToolResult {
-		if a.opts.Checkpoint == nil || a.currentRun == nil {
-			return next(ctx, call)
-		}
-		name := toolCheckpointName(call)
-		if rec, ok := findStep(a.currentRun.Steps, name); ok && rec.Status == "done" {
-			return ai.ToolResult{ID: call.ID, Value: rec.Result, Content: rec.Result}
-		}
-
-		idx := upsertStep(&a.currentRun.Steps, flow.StepRecord{Name: name, Status: "in_progress"})
-		_ = a.saveRun(ctx, *a.currentRun)
-		res := next(ctx, call)
-		a.currentRun.Steps[idx].Attempts++
-		if res.Refused != "" {
-			a.currentRun.Steps[idx].Status = "failed"
-			a.currentRun.Steps[idx].Error = res.Content
-			_ = a.saveRun(ctx, *a.currentRun)
-			return res
-		}
-		a.currentRun.Steps[idx].Status = "done"
-		a.currentRun.Steps[idx].Result = res.Content
-		a.currentRun.Steps[idx].Error = ""
-		_ = a.saveRun(ctx, *a.currentRun)
-		return res
-	}
-}
-
-func toolCheckpointName(call ai.ToolCall) string {
-	b, _ := json.Marshal(call.Input)
-	return "tool:" + call.Name + ":" + string(b)
-}
-
-func findStep(steps []flow.StepRecord, name string) (flow.StepRecord, bool) {
-	for _, step := range steps {
-		if step.Name == name {
-			return step, true
-		}
-	}
-	return flow.StepRecord{}, false
-}
-
-func upsertStep(steps *[]flow.StepRecord, rec flow.StepRecord) int {
-	for i := range *steps {
-		if (*steps)[i].Name == rec.Name {
-			(*steps)[i].Status = rec.Status
-			(*steps)[i].Error = rec.Error
-			return i
-		}
-	}
-	if len(*steps) == 0 || (*steps)[0].Name != agentAskStep {
-		*steps = append([]flow.StepRecord{{Name: agentAskStep, Status: "in_progress"}}, (*steps)...)
-	}
-	*steps = append(*steps, rec)
-	return len(*steps) - 1
-}

agent/checkpoint_test.go

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

agent/conformance_test.go

@@ -1,158 +0,0 @@
-package agent
-
-import (
-	"context"
-	"errors"
-	"fmt"
-	"os"
-	"strings"
-	"testing"
-	"time"
-
-	"go-micro.dev/v6/ai"
-	"go-micro.dev/v6/registry"
-	"go-micro.dev/v6/store"
-)
-
-type conformanceProvider struct {
-	name  string
-	model string
-	key   string
-	live  bool
-}
-
-func TestAgentProviderConformanceMatrix(t *testing.T) {
-	providers := []conformanceProvider{
-		{name: "fake"},
-		{name: "openai", key: "OPENAI_API_KEY", model: "GO_MICRO_CONFORMANCE_OPENAI_MODEL", live: true},
-		{name: "anthropic", key: "ANTHROPIC_API_KEY", model: "GO_MICRO_CONFORMANCE_ANTHROPIC_MODEL", live: true},
-		{name: "atlascloud", key: "ATLASCLOUD_API_KEY", model: "GO_MICRO_CONFORMANCE_ATLASCLOUD_MODEL", live: true},
-		{name: "gemini", key: "GEMINI_API_KEY", model: "GO_MICRO_CONFORMANCE_GEMINI_MODEL", live: true},
-		{name: "groq", key: "GROQ_API_KEY", model: "GO_MICRO_CONFORMANCE_GROQ_MODEL", live: true},
-		{name: "mistral", key: "MISTRAL_API_KEY", model: "GO_MICRO_CONFORMANCE_MISTRAL_MODEL", live: true},
-		{name: "together", key: "TOGETHER_API_KEY", model: "GO_MICRO_CONFORMANCE_TOGETHER_MODEL", live: true},
-	}
-
-	for _, provider := range providers {
-		provider := provider
-		t.Run(provider.name, func(t *testing.T) {
-			runAgentConformanceScenario(t, provider)
-		})
-	}
-}
-
-func runAgentConformanceScenario(t *testing.T, provider conformanceProvider) {
-	t.Helper()
-	if provider.live {
-		if os.Getenv(provider.key) == "" {
-			t.Skipf("%s not set; skipping live %s conformance", provider.key, provider.name)
-		}
-		if os.Getenv("GO_MICRO_AGENT_CONFORMANCE_LIVE") == "" {
-			t.Skipf("GO_MICRO_AGENT_CONFORMANCE_LIVE not set; skipping live %s conformance", provider.name)
-		}
-	} else {
-		fakeGen = func(ctx context.Context, opts ai.Options, req *ai.Request) (*ai.Response, error) {
-			if req.Prompt == "" {
-				return nil, errors.New("missing prompt")
-			}
-			if len(req.Messages) == 0 || req.Messages[len(req.Messages)-1].Role != "user" {
-				return nil, fmt.Errorf("missing user history: %+v", req.Messages)
-			}
-			if len(req.Tools) == 0 {
-				return nil, errors.New("missing tools")
-			}
-			if opts.ToolHandler == nil {
-				return nil, errors.New("missing tool handler")
-			}
-			res := opts.ToolHandler(ctx, ai.ToolCall{
-				ID:    "fake-call-1",
-				Name:  "conformance_echo",
-				Input: map[string]any{"value": "agent-conformance"},
-			})
-			if res.Content == "" {
-				return nil, errors.New("empty tool result")
-			}
-			return &ai.Response{
-				Reply:     "used conformance_echo",
-				Answer:    res.Content,
-				ToolCalls: []ai.ToolCall{{ID: "fake-call-1", Name: "conformance_echo", Input: map[string]any{"value": "agent-conformance"}, Result: res.Content}},
-			}, nil
-		}
-		defer func() { fakeGen = nil }()
-	}
-
-	var sawTool bool
-	var sawRunInfo bool
-	agentOpts := []Option{
-		Name("conformance-" + provider.name),
-		Provider(provider.name),
-		APIKey(os.Getenv(provider.key)),
-		Prompt("You are a conformance test agent. Use the conformance_echo tool exactly once with input {\"value\":\"agent-conformance\"}, then answer with the tool result."),
-		WithRegistry(registry.NewMemoryRegistry()),
-		WithStore(store.NewMemoryStore()),
-		WithMemory(NewInMemory(8)),
-		ModelCallTimeout(45 * time.Second),
-		WithTool("conformance_echo", "Echo a conformance value and return a deterministic marker.", map[string]any{
-			"value": map[string]any{"type": "string", "description": "value to echo"},
-		}, func(ctx context.Context, input map[string]any) (string, error) {
-			sawTool = true
-			info, ok := ai.RunInfoFrom(ctx)
-			if !ok {
-				return "", errors.New("missing run info")
-			}
-			if info.RunID == "" || info.Agent != "conformance-"+provider.name {
-				return "", fmt.Errorf("unexpected run info: %+v", info)
-			}
-			sawRunInfo = true
-			if input["value"] != "agent-conformance" {
-				return "", fmt.Errorf("unexpected value %v", input["value"])
-			}
-			return `{"marker":"agent-conformance-ok"}`, nil
-		}),
-	}
-	if provider.model != "" {
-		if model := os.Getenv(provider.model); model != "" {
-			agentOpts = append(agentOpts, Model(model))
-		}
-	}
-
-	a := New(agentOpts...)
-	resp, err := a.Ask(context.Background(), "Run the provider conformance check.")
-	if err != nil {
-		t.Fatalf("Ask: %v", err)
-	}
-	if resp.RunID == "" {
-		t.Fatal("RunID is empty")
-	}
-	if resp.Agent != "conformance-"+provider.name {
-		t.Fatalf("Agent = %q", resp.Agent)
-	}
-	if !sawTool {
-		t.Fatal("provider did not request the conformance tool")
-	}
-	if !sawRunInfo {
-		t.Fatal("tool did not receive RunInfo")
-	}
-	if !strings.Contains(resp.Reply, "agent-conformance-ok") && !strings.Contains(resp.Reply, "agent-conformance") {
-		t.Fatalf("reply %q does not include conformance marker", resp.Reply)
-	}
-}
-
-func TestAgentProviderConformanceFakeError(t *testing.T) {
-	fakeGen = func(ctx context.Context, opts ai.Options, req *ai.Request) (*ai.Response, error) {
-		return nil, errors.New("conformance provider failure")
-	}
-	defer func() { fakeGen = nil }()
-
-	a := New(
-		Name("conformance-error"),
-		Provider("fake"),
-		WithRegistry(registry.NewMemoryRegistry()),
-		WithStore(store.NewMemoryStore()),
-		WithMemory(NewInMemory(4)),
-	)
-	_, err := a.Ask(context.Background(), "fail deterministically")
-	if err == nil || !strings.Contains(err.Error(), "conformance provider failure") {
-		t.Fatalf("Ask error = %v, want conformance provider failure", err)
-	}
-}

agent/integration_test.go

@@ -179,46 +179,3 @@ func TestDelegateRequiresTask(t *testing.T) {
 		t.Errorf("delegate with no task should error; got %q", content)
 	}
 }
-
-func TestCompactingMemorySummarizesAndRecallsArchivedContext(t *testing.T) {
-	var sawSummary, sawRecall bool
-	fakeGen = func(ctx context.Context, opts ai.Options, req *ai.Request) (*ai.Response, error) {
-		for _, msg := range req.Messages {
-			text := msg.Content.(string)
-			if strings.Contains(text, "Conversation memory summary") && strings.Contains(text, "alpha project") {
-				sawSummary = true
-			}
-			if strings.Contains(text, "alpha project budget is 42") {
-				sawRecall = true
-			}
-		}
-		return &ai.Response{Reply: "ok"}, nil
-	}
-	defer func() { fakeGen = nil }()
-
-	a := newTestAgent(Name("memory"), CompactMemory(4, 2), MemoryRecallLimit(3))
-	turns := []string{
-		"alpha project budget is 42",
-		"beta project owner is sam",
-		"gamma project deadline is monday",
-		"delta project status is green",
-		"epsilon project risk is low",
-	}
-	for _, turn := range turns {
-		if _, err := a.Ask(context.Background(), turn); err != nil {
-			t.Fatalf("Ask(%q): %v", turn, err)
-		}
-	}
-	if got := len(a.mem.Messages()); got > 4 {
-		t.Fatalf("compacted memory retained %d messages, want <= 4", got)
-	}
-	if _, err := a.Ask(context.Background(), "what was the alpha budget?"); err != nil {
-		t.Fatalf("Ask recall: %v", err)
-	}
-	if !sawSummary {
-		t.Error("model request did not include a deterministic compacted summary")
-	}
-	if !sawRecall {
-		t.Error("model request did not recall archived matching context")
-	}
-}

agent/memory.go

@@ -2,9 +2,6 @@ package agent
 
 import (
 	"encoding/json"
-	"fmt"
-	"sort"
-	"strings"
 	"sync"
 
 	"go-micro.dev/v6/ai"
@@ -24,21 +21,6 @@ type Memory interface {
 	Clear()
 }
 
-// MemoryCompaction configures deterministic, store-backed context compaction
-// for the default memory implementation. When the retained conversation grows
-// past MaxMessages, older turns are collapsed into a summary message while the
-// newest KeepRecent turns stay verbatim for provider-neutral continuity.
-type MemoryCompaction struct {
-	MaxMessages int
-	KeepRecent  int
-}
-
-// MemoryRecall is implemented by memory backends that can retrieve durable
-// prior context relevant to a new turn without replaying every stored message.
-type MemoryRecall interface {
-	Recall(query string, limit int) []ai.Message
-}
-
 // NewMemory returns the default store-backed memory: an in-process
 // conversation buffer (truncated to limit) that persists to the store
 // under key, so an agent picks up where it left off after a restart.
@@ -49,33 +31,6 @@ func NewMemory(s store.Store, key string, limit int) Memory {
 	return m
 }
 
-// NewCompactingMemory returns store-backed memory with explicit compaction and
-// retrieval controls. It keeps all messages in the backing store, compacts older
-// turns into a deterministic summary when the conversation exceeds maxMessages,
-// and lets callers recall relevant prior turns with Recall.
-func NewCompactingMemory(s store.Store, key string, maxMessages, keepRecent int) Memory {
-	if keepRecent <= 0 {
-		keepRecent = maxMessages / 2
-	}
-	if keepRecent < 1 {
-		keepRecent = 1
-	}
-	m := &storeMemory{
-		store: s,
-		key:   key,
-		// Use an unlimited buffer here; compaction, not truncation, decides
-		// what remains in active context so a summary can preserve older turns.
-		hist: ai.NewHistory(0),
-		compaction: MemoryCompaction{
-			MaxMessages: maxMessages,
-			KeepRecent:  keepRecent,
-		},
-	}
-	m.load()
-	m.compact()
-	return m
-}
-
 // NewInMemory returns conversation memory that is not persisted.
 func NewInMemory(limit int) Memory {
 	return &storeMemory{hist: ai.NewHistory(limit)}
@@ -84,19 +39,16 @@ func NewInMemory(limit int) Memory {
 // storeMemory is the default Memory: an ai.History buffer optionally
 // persisted to a store.
 type storeMemory struct {
-	mu         sync.Mutex
-	store      store.Store
-	key        string
-	hist       *ai.History
-	compaction MemoryCompaction
-	archive    []ai.Message
+	mu    sync.Mutex
+	store store.Store
+	key   string
+	hist  *ai.History
 }
 
 func (m *storeMemory) Add(role, content string) {
 	m.mu.Lock()
 	m.hist.Add(role, content)
 	m.mu.Unlock()
-	m.compact()
 	m.save()
 }
 
@@ -109,49 +61,10 @@ func (m *storeMemory) Messages() []ai.Message {
 func (m *storeMemory) Clear() {
 	m.mu.Lock()
 	m.hist.Reset()
-	m.archive = nil
 	m.mu.Unlock()
 	m.save()
 }
 
-// Recall returns archived messages whose content contains words from query.
-// It is deterministic and provider-neutral: no embeddings or model calls are
-// required, but semantic/vector stores can replace Memory for richer retrieval.
-func (m *storeMemory) Recall(query string, limit int) []ai.Message {
-	m.mu.Lock()
-	defer m.mu.Unlock()
-	if limit <= 0 {
-		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-- {
-		msg := m.archive[i]
-		if score := recallScore(msg, terms); score > 0 {
-			matches = append(matches, match{msg: msg, score: score, index: i})
-		}
-	}
-	sort.SliceStable(matches, func(i, j int) bool {
-		if matches[i].score != matches[j].score {
-			return matches[i].score > matches[j].score
-		}
-		return matches[i].index > matches[j].index
-	})
-	if len(matches) > limit {
-		matches = matches[:limit]
-	}
-	out := make([]ai.Message, 0, len(matches))
-	for _, match := range matches {
-		out = append(out, match.msg)
-	}
-	return out
-}
-
 func (m *storeMemory) load() {
 	if m.store == nil || m.key == "" {
 		return
@@ -160,17 +73,12 @@ func (m *storeMemory) load() {
 	if err != nil || len(recs) == 0 {
 		return
 	}
-	var state memoryState
-	if err := json.Unmarshal(recs[0].Value, &state); err != nil {
-		var msgs []ai.Message
-		if err := json.Unmarshal(recs[0].Value, &msgs); err != nil {
-			return
-		}
-		state.Messages = msgs
+	var msgs []ai.Message
+	if err := json.Unmarshal(recs[0].Value, &msgs); err != nil {
+		return
 	}
 	m.mu.Lock()
-	m.archive = state.Archive
-	for _, msg := range state.Messages {
+	for _, msg := range msgs {
 		m.hist.Add(msg.Role, msg.Content)
 	}
 	m.mu.Unlock()
@@ -181,94 +89,10 @@ func (m *storeMemory) save() {
 		return
 	}
 	m.mu.Lock()
-	data, err := json.Marshal(memoryState{
-		Messages: m.hist.Messages(),
-		Archive:  m.archive,
-	})
+	data, err := json.Marshal(m.hist.Messages())
 	m.mu.Unlock()
 	if err != nil {
 		return
 	}
 	_ = m.store.Write(&store.Record{Key: m.key, Value: data})
 }
-
-func (m *storeMemory) compact() {
-	if m.compaction.MaxMessages <= 0 {
-		return
-	}
-	m.mu.Lock()
-	defer m.mu.Unlock()
-	msgs := m.hist.Messages()
-	if len(msgs) <= m.compaction.MaxMessages {
-		return
-	}
-	keep := m.compaction.KeepRecent
-	if keep <= 0 || keep >= m.compaction.MaxMessages {
-		keep = m.compaction.MaxMessages - 1
-	}
-	if keep < 1 {
-		keep = 1
-	}
-	cut := len(msgs) - keep
-	older := msgs[:cut]
-	recent := msgs[cut:]
-	m.archive = append(m.archive, older...)
-	summary := ai.Message{
-		Role:    "system",
-		Content: fmt.Sprintf("Conversation memory summary: %s", summarizeMessages(older)),
-	}
-	m.hist.Reset()
-	m.hist.Add(summary.Role, summary.Content)
-	for _, msg := range recent {
-		m.hist.Add(msg.Role, msg.Content)
-	}
-}
-
-func summarizeMessages(msgs []ai.Message) string {
-	var b strings.Builder
-	for i, msg := range msgs {
-		if i > 0 {
-			b.WriteString(" | ")
-		}
-		fmt.Fprintf(&b, "%s: %s", msg.Role, compactText(fmt.Sprint(msg.Content), 120))
-	}
-	return b.String()
-}
-
-func compactText(s string, max int) string {
-	s = strings.Join(strings.Fields(s), " ")
-	if max > 0 && len(s) > max {
-		return s[:max] + "…"
-	}
-	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
-	for _, term := range strings.Fields(strings.ToLower(query)) {
-		term = strings.Trim(term, ".,!?;:\"'()[]{}")
-		if len(term) < 3 || seen[term] {
-			continue
-		}
-		seen[term] = true
-		terms = append(terms, term)
-	}
-	return terms
-}
-
-type memoryState struct {
-	Messages []ai.Message `json:"messages"`
-	Archive  []ai.Message `json:"archive,omitempty"`
-}

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

@@ -6,7 +6,6 @@ import (
 
 	"go-micro.dev/v6/ai"
 	"go-micro.dev/v6/client"
-	"go-micro.dev/v6/flow"
 	"go-micro.dev/v6/registry"
 	"go-micro.dev/v6/store"
 	"go.opentelemetry.io/otel/trace"
@@ -60,16 +59,6 @@ type Options struct {
 	// Memory is the agent's conversation memory. Nil = the default
 	// store-backed memory (durable across restarts).
 	Memory Memory
-	// MemoryCompaction enables deterministic compaction/retrieval on the
-	// default store-backed memory. Custom Memory implementations can expose
-	// retrieval by implementing MemoryRecall.
-	MemoryCompaction MemoryCompaction
-	// MemoryRecallLimit bounds recalled archived turns injected into a model
-	// request (0 disables recall injection).
-	MemoryRecallLimit int
-	// Checkpoint persists agent Ask runs so callers can resume by run id
-	// after a restart without replaying a run that already completed.
-	Checkpoint flow.Checkpoint
 
 	// MaxSteps bounds the number of tool executions per Ask (0 =
 	// unbounded). Once exceeded, further tool calls are refused and the
@@ -222,34 +211,6 @@ func WithMemory(m Memory) Option {
 	return func(o *Options) { o.Memory = m }
 }
 
-// CompactMemory enables deterministic, store-backed memory compaction for the
-// default agent memory. Older turns are summarized once active context exceeds
-// maxMessages, keepRecent newest turns remain verbatim, and recalled archived
-// turns are injected into matching future asks.
-func CompactMemory(maxMessages, keepRecent int) Option {
-	return func(o *Options) {
-		o.MemoryCompaction = MemoryCompaction{MaxMessages: maxMessages, KeepRecent: keepRecent}
-		if o.MemoryRecallLimit == 0 {
-			o.MemoryRecallLimit = 5
-		}
-	}
-}
-
-// MemoryRecallLimit sets how many archived turns a memory backend may inject
-// into a model request for the current Ask. Use 0 to disable retrieval.
-func MemoryRecallLimit(n int) Option {
-	return func(o *Options) { o.MemoryRecallLimit = n }
-}
-
-// WithCheckpoint sets the durability backend for agent Ask runs. The
-// Checkpoint interface is shared with flow so services, agents, and workflows
-// can use one execution history backend. When set, each Ask is saved as a
-// single-step run keyed by run id; Resume returns a completed run's persisted
-// response instead of calling the model again.
-func WithCheckpoint(c flow.Checkpoint) Option {
-	return func(o *Options) { o.Checkpoint = c }
-}
-
 // WrapTool registers a tool-execution wrapper, the tool-side analog of
 // a client/server middleware wrapper. Each wrapper takes the next handler
 // and returns a new one; code before the next(...) call runs before the

agent/otel.go

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

agent/otel_test.go

@@ -3,8 +3,6 @@ package agent
 import (
 	"context"
 	"encoding/json"
-	"fmt"
-	"strings"
 	"testing"
 	"time"
 
@@ -30,11 +28,7 @@ func (m *otelTestModel) Stream(context.Context, *ai.Request, ...ai.GenerateOptio
 }
 func (m *otelTestModel) Generate(ctx context.Context, req *ai.Request, opts ...ai.GenerateOption) (*ai.Response, error) {
 	if m.opts.ToolHandler != nil {
-		if strings.Contains(req.Prompt, "delegate") {
-			_ = m.opts.ToolHandler(ctx, ai.ToolCall{ID: "call-delegate", Name: toolDelegate, Input: map[string]any{"task": "subtask"}})
-		} else if !strings.Contains(req.Prompt, "subtask") {
-			_ = m.opts.ToolHandler(ctx, ai.ToolCall{ID: "call-1", Name: "probe", Input: map[string]any{"ok": true}})
-		}
+		_ = m.opts.ToolHandler(ctx, ai.ToolCall{ID: "call-1", Name: "probe", Input: map[string]any{"ok": true}})
 	}
 	return &ai.Response{Reply: "done", Usage: ai.Usage{InputTokens: 2, OutputTokens: 3, TotalTokens: 5}}, nil
 }
@@ -71,16 +65,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
@@ -125,78 +109,14 @@ func TestAgentOpenTelemetrySpans(t *testing.T) {
 	}
 }
 
-func spanEventHasRunInfo(events []trace.Event, name, runID, agentName string) bool {
-	for _, event := range events {
-		if event.Name != name {
-			continue
-		}
-		attrs := spanAttributes(event.Attributes)
-		if attrs[AttrRunID] == runID && attrs[AttrAgentName] == agentName {
-			return true
-		}
-	}
-	return false
-}
-
 func spanAttributes(attrs []attribute.KeyValue) map[string]string {
 	out := make(map[string]string, len(attrs))
 	for _, attr := range attrs {
-		out[string(attr.Key)] = fmt.Sprint(attr.Value.AsInterface())
+		out[string(attr.Key)] = attr.Value.AsString()
 	}
 	return out
 }
 
-func TestAgentOpenTelemetrySpansDelegateLineage(t *testing.T) {
-	exp := tracetest.NewInMemoryExporter()
-	tp := trace.NewTracerProvider(trace.WithSyncer(exp))
-	st := store.NewMemoryStore()
-	a := New(Name("conductor"), Provider("oteltest"), WithStore(st), TraceProvider(tp))
-	if _, err := a.Ask(context.Background(), "delegate please"); err != nil {
-		t.Fatal(err)
-	}
-
-	spans := exp.GetSpans().Snapshots()
-	var parentRunID string
-	var delegateSpanID string
-	var subRunSeen bool
-	for _, s := range spans {
-		attrs := spanAttributes(s.Attributes())
-		if s.Name() == spanNameRun && attrs[AttrAgentName] == "conductor" {
-			parentRunID = attrs[AttrRunID]
-		}
-	}
-	if parentRunID == "" {
-		t.Fatal("parent run span missing run id")
-	}
-	for _, s := range spans {
-		attrs := spanAttributes(s.Attributes())
-		if s.Name() == spanNameToolCall && attrs[AttrToolName] == toolDelegate {
-			if attrs[AttrDelegate] != "true" || attrs[AttrRunID] != parentRunID {
-				t.Fatalf("delegate span missing correlation attributes: %#v", attrs)
-			}
-			delegateSpanID = s.SpanContext().SpanID().String()
-		}
-	}
-	if delegateSpanID == "" {
-		t.Fatal("delegate tool span not emitted")
-	}
-	for _, s := range spans {
-		attrs := spanAttributes(s.Attributes())
-		if s.Name() == spanNameRun && attrs[AttrAgentName] == "conductor.sub" {
-			if attrs[AttrParentRunID] != parentRunID {
-				t.Fatalf("sub-agent run parent attr = %q, want %q", attrs[AttrParentRunID], parentRunID)
-			}
-			if s.Parent().SpanID().String() != delegateSpanID {
-				t.Fatalf("sub-agent run parent span = %s, want delegate span %s", s.Parent().SpanID(), delegateSpanID)
-			}
-			subRunSeen = true
-		}
-	}
-	if !subRunSeen {
-		t.Fatalf("sub-agent run span not emitted; got %d spans", len(spans))
-	}
-}
-
 func TestAgentRunTimelineRecordsModelAndToolWithoutTraceProvider(t *testing.T) {
 	st := store.NewMemoryStore()
 	a := New(Name("runner-noop"), Provider("oteltest"), WithStore(st), WithTool("probe", "probe", nil, func(context.Context, map[string]any) (string, error) { return "ok", nil }))

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"
@@ -92,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
@@ -151,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/capabilities_test.go

@@ -34,7 +34,7 @@ func TestRegisteredProviders(t *testing.T) {
 	}
 
 	got = ai.RegisteredProviders("stream")
-	want = []string{"openai"}
+	want = []string{}
 	if !reflect.DeepEqual(got, want) {
 		t.Fatalf("RegisteredProviders(stream) = %#v, want %#v", got, want)
 	}
@@ -48,7 +48,7 @@ func TestCapabilityRows(t *testing.T) {
 		{Provider: "gemini", Capabilities: ai.Capabilities{Model: 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: "openai", Capabilities: ai.Capabilities{Model: true, Image: true}},
 		{Provider: "together", Capabilities: ai.Capabilities{Model: true}},
 	}
 	if !reflect.DeepEqual(got, want) {
@@ -69,7 +69,7 @@ func TestCapabilityMatrix(t *testing.T) {
 		}
 	}
 
-	if caps := ai.ProviderCapabilities("openai"); caps != (ai.Capabilities{Model: true, Image: true, Stream: true}) {
+	if caps := ai.ProviderCapabilities("openai"); caps != (ai.Capabilities{Model: true, Image: true}) {
 		t.Fatalf("ProviderCapabilities(openai) = %#v", caps)
 	}
 	if caps := ai.ProviderCapabilities("atlascloud"); caps != (ai.Capabilities{Model: true, Image: true, Video: true}) {
@@ -88,7 +88,7 @@ func TestRegisterStream(t *testing.T) {
 	}
 
 	got := ai.RegisteredProviders("stream")
-	want := []string{"openai", "test-stream"}
+	want := []string{"test-stream"}
 	if !reflect.DeepEqual(got, want) {
 		t.Fatalf("RegisteredProviders(stream) = %#v, want %#v", got, want)
 	}

ai/model.go

@@ -113,21 +113,12 @@ const (
 // RunInfo describes the agent run a tool call belongs to. The agent
 // attaches it to the context passed to a ToolHandler, so a wrapper can
 // correlate calls within a run and across delegation without coupling to
-// the agent package. Flows also attach their name and current step so
-// tools and agents called from a workflow can be tied back to the
-// services → agents → workflows lifecycle that invoked them. Per-call
-// detail (tool name, id) is on the ToolCall. Attempt and MaxAttempts are
-// set while a model Generate call is in progress, so tools and wrappers can
-// tell which provider attempt produced the call and whether it is part of a
-// retry budget. They are zero when no model-attempt context is known.
+// the agent package. Per-call detail (tool name, id) is on the ToolCall;
+// step and attempt counts are naturally counted by the wrapper itself.
 type RunInfo struct {
-	RunID       string // correlation id for this agent or flow run
-	ParentID    string // the run that delegated to this one, if any
-	Agent       string // the agent's name
-	Flow        string // the flow's name, when the call is part of a workflow
-	Step        string // the flow step currently executing, when known
-	Attempt     int    // current model Generate attempt, starting at 1 when known
-	MaxAttempts int    // configured model Generate attempt budget when known
+	RunID    string // correlation id for this agent run (one per Ask)
+	ParentID string // the run that delegated to this one, if any
+	Agent    string // the agent's name
 }
 
 type runInfoKey struct{}

ai/openai/openai.go

@@ -2,7 +2,6 @@
 package openai
 
 import (
-	"bufio"
 	"bytes"
 	"context"
 	"encoding/json"
@@ -21,7 +20,6 @@ func init() {
 	ai.RegisterImage("openai", func(opts ...ai.Option) ai.ImageModel {
 		return NewProvider(opts...)
 	})
-	ai.RegisterStream("openai")
 }
 
 // Provider implements the ai.Model interface for OpenAI
@@ -85,12 +83,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 +91,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
@@ -150,110 +140,9 @@ func (p *Provider) Generate(ctx context.Context, req *ai.Request, opts ...ai.Gen
 	return resp, nil
 }
 
-// Stream generates a streaming response from the OpenAI chat completions API.
+// Stream generates a streaming response (not yet implemented)
 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 &openAIStream{body: httpResp.Body, scanner: bufio.NewScanner(httpResp.Body)}, nil
-}
-
-type openAIStream struct {
-	body    io.ReadCloser
-	scanner *bufio.Scanner
-	closed  bool
-}
-
-func (s *openAIStream) 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 *openAIStream) Close() error {
-	if s.closed {
-		return nil
-	}
-	s.closed = true
-	return s.body.Close()
+	return nil, fmt.Errorf("%w: openai provider", ai.ErrStreamingUnsupported)
 }
 
 // callAPI makes an HTTP request to the OpenAI API

ai/openai/openai_test.go

@@ -2,13 +2,8 @@ package openai
 
 import (
 	"context"
-	"encoding/json"
 	"errors"
-	"io"
-	"net/http"
-	"net/http/httptest"
 	"testing"
-	"time"
 
 	"go-micro.dev/v6/ai"
 )
@@ -86,96 +81,16 @@ 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()
+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")
+	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)
-	}
-	if _, err := stream.Recv(); !errors.Is(err, io.EOF) {
-		t.Fatalf("final error = %v, want EOF", err)
-	}
-}
-
-func TestProvider_StreamPropagatesMalformedChunk(t *testing.T) {
-	ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
-		w.Header().Set("Content-Type", "text/event-stream")
-		_, _ = w.Write([]byte("data: {bad json}\n\n"))
-	}))
-	defer ts.Close()
-
-	p := NewProvider(ai.WithAPIKey("test-key"), ai.WithBaseURL(ts.URL))
-	stream, err := p.Stream(context.Background(), &ai.Request{Prompt: "Hello"})
-	if err != nil {
-		t.Fatalf("Stream returned error: %v", err)
-	}
-	defer stream.Close()
-
-	if _, err := stream.Recv(); err == nil {
-		t.Fatal("Recv returned nil error for malformed chunk")
-	}
-}
-
-func TestProvider_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")
+	_, err := p.Stream(context.Background(), req)
+	if !errors.Is(err, ai.ErrStreamingUnsupported) {
+		t.Fatalf("Stream error = %v, want ErrStreamingUnsupported", err)
 	}
 }
 

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,28 +13,9 @@ type StatusCoder interface {
 	StatusCode() int
 }
 
-// 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
 }
 
@@ -42,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 {
@@ -52,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
@@ -86,11 +60,6 @@ func GenerateWithRetry(ctx context.Context, m Model, req *Request, policy Genera
 		if policy.Timeout > 0 {
 			callCtx, cancel = context.WithTimeout(ctx, policy.Timeout)
 		}
-		if info, ok := RunInfoFrom(callCtx); ok {
-			info.Attempt = attempt
-			info.MaxAttempts = policy.MaxAttempts
-			callCtx = WithRunInfo(callCtx, info)
-		}
 		resp, err := m.Generate(callCtx, req, opts...)
 		cancel()
 		if err == nil {
@@ -102,10 +71,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
 		}
@@ -133,57 +101,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}
-}
-
-// 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

@@ -94,90 +94,3 @@ func TestGenerateWithRetryHonorsPerAttemptTimeout(t *testing.T) {
 		t.Fatalf("attempts = %d, want 2", attempts)
 	}
 }
-
-func TestGenerateWithRetryAddsAttemptMetadataToRunInfo(t *testing.T) {
-	var got []RunInfo
-	model := retryModel{generate: func(ctx context.Context, _ *Request, _ ...GenerateOption) (*Response, error) {
-		info, ok := RunInfoFrom(ctx)
-		if !ok {
-			t.Fatal("RunInfo missing from attempt context")
-		}
-		got = append(got, info)
-		if info.Attempt == 1 {
-			return nil, errors.New("temporary provider outage")
-		}
-		return &Response{Reply: "ok"}, nil
-	}}
-
-	ctx := WithRunInfo(context.Background(), RunInfo{RunID: "run-1", Agent: "worker"})
-	_, err := GenerateWithRetry(ctx, model, &Request{Prompt: "hi"}, GeneratePolicy{
-		MaxAttempts: 2,
-		Backoff:     time.Millisecond,
-	})
-	if err != nil {
-		t.Fatalf("GenerateWithRetry returned error: %v", err)
-	}
-	if len(got) != 2 {
-		t.Fatalf("attempt contexts = %d, want 2", len(got))
-	}
-	for i, info := range got {
-		wantAttempt := i + 1
-		if info.Attempt != wantAttempt {
-			t.Fatalf("attempt %d RunInfo.Attempt = %d, want %d", i, info.Attempt, wantAttempt)
-		}
-		if info.MaxAttempts != 2 {
-			t.Fatalf("attempt %d RunInfo.MaxAttempts = %d, want 2", i, info.MaxAttempts)
-		}
-		if info.RunID != "run-1" || info.Agent != "worker" {
-			t.Fatalf("attempt %d RunInfo identity = (%q, %q), want (run-1, worker)", i, info.RunID, info.Agent)
-		}
-	}
-}
-
-type statusErr int
-
-func (e statusErr) Error() string   { return "provider status" }
-func (e statusErr) StatusCode() int { return int(e) }
-
-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)
-	}
-}

cmd/micro/chat/chat.go

@@ -10,9 +10,7 @@ import (
 	"bufio"
 	"context"
 	"encoding/json"
-	"errors"
 	"fmt"
-	"io"
 	"os"
 	"os/exec"
 	"path/filepath"
@@ -81,7 +79,6 @@ Examples:
 			&cli.StringFlag{Name: "model", Usage: "Model name (uses provider default if unset)", EnvVars: []string{"MICRO_AI_MODEL"}},
 			&cli.StringFlag{Name: "base_url", Usage: "Override the provider's base URL", EnvVars: []string{"MICRO_AI_BASE_URL"}},
 			&cli.StringFlag{Name: "prompt", Usage: "Send a single prompt and exit (non-interactive)"},
-			&cli.BoolFlag{Name: "stream", Usage: "Stream model output as it is generated when the provider supports it"},
 		},
 		Action: run,
 	})
@@ -105,7 +102,6 @@ type session struct {
 	sysPrompt string
 	procs     []*exec.Cmd
 	agents    map[string]agentInfo
-	stream    bool
 
 	// Built-in agent capabilities (plan, delegate), shared with the
 	// agent package so the direct-service fallback has the same tools a
@@ -315,7 +311,6 @@ func run(c *cli.Context) error {
 	modelName := c.String("model")
 	baseURL := c.String("base_url")
 	singlePrompt := c.String("prompt")
-	streamOutput := c.Bool("stream")
 
 	if provider == "" {
 		provider = ai.AutoDetectProvider(baseURL)
@@ -352,7 +347,6 @@ func run(c *cli.Context) error {
 		hist:          ai.NewHistory(50),
 		builtinTools:  builtinTools,
 		builtinHandle: builtinHandle,
-		stream:        streamOutput,
 	}
 	s.refreshTools()
 
@@ -455,21 +449,12 @@ func (s *session) ask(ctx context.Context, prompt string) error {
 	// Fallback: direct service access (no agents)
 	s.hist.Add("user", prompt)
 
-	req := &ai.Request{
+	resp, err := s.model.Generate(ctx, &ai.Request{
 		Prompt:       prompt,
 		SystemPrompt: s.sysPrompt,
 		Tools:        s.toolList,
 		Messages:     s.hist.Messages(),
-	}
-	if s.stream {
-		if err := s.askStream(ctx, req); err == nil {
-			return nil
-		} else if !errors.Is(err, ai.ErrStreamingUnsupported) {
-			return err
-		}
-	}
-
-	resp, err := s.model.Generate(ctx, req)
+	})
 	if err != nil {
 		return err
 	}
@@ -504,34 +489,6 @@ func (s *session) ask(ctx context.Context, prompt string) error {
 	return nil
 }
 
-func (s *session) askStream(ctx context.Context, req *ai.Request) error {
-	stream, err := s.model.Stream(ctx, req)
-	if err != nil {
-		return err
-	}
-	defer stream.Close()
-	var reply strings.Builder
-	for {
-		chunk, err := stream.Recv()
-		if errors.Is(err, io.EOF) {
-			break
-		}
-		if err != nil {
-			return err
-		}
-		if chunk == nil || chunk.Reply == "" {
-			continue
-		}
-		fmt.Print(chunk.Reply)
-		reply.WriteString(chunk.Reply)
-	}
-	if reply.Len() > 0 {
-		fmt.Println()
-		s.hist.Add("assistant", reply.String())
-	}
-	return nil
-}
-
 // routeToAgent dispatches a message to the right agent.
 // If there's only one agent, sends directly. Otherwise uses the
 // LLM to classify intent and route.

cmd/micro/cli/deploy/deploy.go

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

cmd/micro/cli/deploy/deploy_test.go

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

cmd/micro/cli/new/contract_test.go

@@ -1,70 +1,24 @@
 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`.
 func TestZeroToOneContract(t *testing.T) {
-	generated := generateService(t, "helloworld")
-
-	for _, rel := range []string{"go.mod", "main.go", "handler/helloworld.go", "README.md", "Makefile"} {
-		if _, err := os.Stat(filepath.Join(generated.dir, rel)); err != nil {
-			t.Fatalf("generated file %s: %v", rel, err)
-		}
-	}
-
-	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
-// intentionally run without the local MCP listener, but that variant must still
-// satisfy the same 0→1 contract: scaffold, tidy, and build without additional
-// toolchain dependencies.
-func TestZeroToOneNoMCPContract(t *testing.T) {
-	generated := generateService(t, "worker", "--no-mcp")
-
-	main, err := os.ReadFile(filepath.Join(generated.dir, "main.go"))
-	if err != nil {
-		t.Fatal(err)
-	}
-	if strings.Contains(string(main), "gateway/mcp") || strings.Contains(string(main), "WithMCP") {
-		t.Fatalf("--no-mcp generated main.go with MCP wiring:\n%s", main)
-	}
-
-	generated.replaceModule(t)
-	generated.build(t)
-	generated.run(t)
-	generated.call(t, "Bob", "Hello Bob")
-}
-
-type generatedService struct {
-	dir      string
-	repoRoot string
-}
-
-func generateService(t *testing.T, name string, args ...string) generatedService {
-	t.Helper()
-
 	if testing.Short() {
 		t.Skip("contract test shells out to the Go toolchain; skipped with -short")
 	}
@@ -85,117 +39,37 @@ func generateService(t *testing.T, name string, args ...string) generatedService
 	defer os.Chdir(oldwd)
 
 	set := flag.NewFlagSet("micro-new", flag.ContinueOnError)
-	set.Bool("no-mcp", false, "")
-	set.Bool("proto", false, "")
-	set.String("template", "", "")
-	set.String("prompt", "", "")
-	set.String("provider", "", "")
-	set.String("api_key", "", "")
-	if err := set.Parse(append(args, name)); err != nil {
+	if err := set.Parse([]string{"helloworld"}); err != nil {
 		t.Fatal(err)
 	}
 	ctx := cli.NewContext(cli.NewApp(), set, nil)
 
 	if err := Run(ctx); err != nil {
-		t.Fatalf("micro new %s %s: %v", strings.Join(args, " "), name, err)
+		t.Fatalf("micro new helloworld: %v", err)
 	}
 
-	return generatedService{dir: filepath.Join(tmp, name), repoRoot: repoRoot}
-}
+	serviceDir := filepath.Join(tmp, "helloworld")
+	for _, rel := range []string{"go.mod", "main.go", "handler/helloworld.go", "README.md", "Makefile"} {
+		if _, err := os.Stat(filepath.Join(serviceDir, rel)); err != nil {
+			t.Fatalf("generated file %s: %v", rel, err)
+		}
+	}
 
-func (g generatedService) replaceModule(t *testing.T) {
-	t.Helper()
-
-	modPath := filepath.Join(g.dir, "go.mod")
+	modPath := filepath.Join(serviceDir, "go.mod")
 	mod, err := os.ReadFile(modPath)
 	if err != nil {
 		t.Fatal(err)
 	}
 	modText := strings.Replace(string(mod), "go-micro.dev/v6 latest", "go-micro.dev/v6 v6.0.0", 1)
-	modText += "\nreplace go-micro.dev/v6 => " + filepath.ToSlash(g.repoRoot) + "\n"
+	modText += "\nreplace go-micro.dev/v6 => " + filepath.ToSlash(repoRoot) + "\n"
 	if err := os.WriteFile(modPath, []byte(modText), 0644); err != nil {
 		t.Fatal(err)
 	}
-}
-
-func (g generatedService) build(t *testing.T) {
-	t.Helper()
 
 	cmd := exec.Command("go", "build", "./...")
-	cmd.Dir = g.dir
+	cmd.Dir = serviceDir
 	out, err := cmd.CombinedOutput()
 	if err != nil {
 		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/zero_to_hero_cli_test.go

@@ -1,30 +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"} {
-		if !commands[want] {
-			t.Fatalf("missing %q command", want)
-		}
-	}
-	if !subcommands["flow"]["runs"] {
-		t.Fatal("missing inspect boundary: flow runs")
-	}
-}

examples/agent-human-input/README.md

@@ -1,36 +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.

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

flow/flow.go

@@ -182,19 +182,6 @@ func (f *Flow) Register(reg registry.Registry, br broker.Broker, cl client.Clien
 // registry. In-flight and past runs remain in the store; Stop only ends
 // the flow's liveness, mirroring how a service leaves the registry when
 // it shuts down.
-func (f *Flow) withTimeout(ctx context.Context) (context.Context, context.CancelFunc) {
-	if ctx == nil {
-		ctx = context.Background()
-	}
-	if f.opts.Timeout <= 0 {
-		return ctx, func() {}
-	}
-	if _, ok := ctx.Deadline(); ok {
-		return ctx, func() {}
-	}
-	return context.WithTimeout(ctx, f.opts.Timeout)
-}
-
 func (f *Flow) Stop() error {
 	if f.sub != nil {
 		_ = f.sub.Unsubscribe()
@@ -212,9 +199,6 @@ func (f *Flow) Stop() error {
 // called automatically on each broker event, but can also be
 // invoked directly for testing or one-shot use.
 func (f *Flow) Execute(ctx context.Context, data string) error {
-	ctx, cancel := f.withTimeout(ctx)
-	defer cancel()
-
 	// Stepped flows run the ordered, checkpointed step loop.
 	if len(f.opts.Steps) > 0 {
 		_, err := f.startRun(ctx, data)
@@ -222,7 +206,7 @@ func (f *Flow) Execute(ctx context.Context, data string) error {
 	}
 
 	runID := uuid.New().String()
-	ctx = ai.WithRunInfo(ctx, ai.RunInfo{RunID: runID, Flow: f.name})
+	ctx = ai.WithRunInfo(ctx, ai.RunInfo{RunID: runID, Agent: f.name})
 
 	start := time.Now()
 

flow/flow_test.go

@@ -1,11 +1,7 @@
 package flow
 
 import (
-	"context"
 	"testing"
-
-	"go-micro.dev/v6/ai"
-	"go-micro.dev/v6/registry"
 )
 
 func TestNew(t *testing.T) {
@@ -87,45 +83,3 @@ func TestDefaultOptions(t *testing.T) {
 		t.Error("default system prompt is empty")
 	}
 }
-
-func TestSingleStepFlowRunInfoIdentifiesFlow(t *testing.T) {
-	model := &runInfoModel{}
-	f := New("single-observed")
-	f.model = model
-	f.toolSet = ai.NewTools(registry.NewMemoryRegistry())
-
-	if err := f.Execute(context.Background(), "observe me"); err != nil {
-		t.Fatalf("Execute: %v", err)
-	}
-	if model.got.RunID == "" {
-		t.Fatal("RunInfo.RunID is empty")
-	}
-	if model.got.Flow != "single-observed" {
-		t.Fatalf("RunInfo.Flow = %q, want single-observed", model.got.Flow)
-	}
-	if model.got.Agent != "" {
-		t.Fatalf("RunInfo.Agent = %q, want empty for flow-owned LLM run", model.got.Agent)
-	}
-	if model.got.Step != "" {
-		t.Fatalf("RunInfo.Step = %q, want empty for single-step flow", model.got.Step)
-	}
-}
-
-type runInfoModel struct {
-	got ai.RunInfo
-}
-
-func (m *runInfoModel) Init(...ai.Option) error { return nil }
-
-func (m *runInfoModel) Options() ai.Options { return ai.Options{} }
-
-func (m *runInfoModel) Generate(ctx context.Context, _ *ai.Request, _ ...ai.GenerateOption) (*ai.Response, error) {
-	m.got, _ = ai.RunInfoFrom(ctx)
-	return &ai.Response{Reply: "ok"}, nil
-}
-
-func (m *runInfoModel) Stream(context.Context, *ai.Request, ...ai.GenerateOption) (ai.Stream, error) {
-	return nil, ai.ErrStreamingUnsupported
-}
-
-func (m *runInfoModel) String() string { return "run-info-model" }

flow/options.go

@@ -24,9 +24,6 @@ type Options struct {
 	BaseURL string
 	// HistoryLimit is the max messages per flow execution.
 	HistoryLimit int
-	// Timeout bounds one flow execution when the caller did not already
-	// provide a context deadline. Zero means no flow-level timeout.
-	Timeout time.Duration
 	// OnResult is called after each execution with the result.
 	OnResult func(Result)
 	// Agent, if set, names a registered agent the flow hands each event
@@ -97,13 +94,6 @@ func HistoryLimit(n int) Option {
 	return func(o *Options) { o.HistoryLimit = n }
 }
 
-// Timeout bounds one flow execution when the caller did not already
-// provide a context deadline. It applies to broker-triggered runs,
-// Execute calls, resumed stepped runs, and retry backoff waits.
-func Timeout(d time.Duration) Option {
-	return func(o *Options) { o.Timeout = d }
-}
-
 // OnResult sets a callback for each execution result.
 func OnResult(fn func(Result)) Option {
 	return func(o *Options) { o.OnResult = fn }

flow/otel.go

@@ -17,7 +17,6 @@ const (
 	spanNameFlowStep = "flow.step"
 
 	AttrFlowRunID     = "flow.run.id"
-	AttrFlowParentID  = "flow.run.parent_id"
 	AttrFlowName      = "flow.name"
 	AttrFlowStepName  = "flow.step.name"
 	AttrFlowStatus    = "flow.status"
@@ -35,7 +34,6 @@ func (f *Flow) startRunSpan(ctx context.Context, run Run) (context.Context, func
 	}
 	ctx, span := f.tracer().Start(ctx, spanNameFlowRun, trace.WithSpanKind(trace.SpanKindInternal), trace.WithAttributes(
 		attribute.String(AttrFlowRunID, run.ID),
-		attribute.String(AttrFlowParentID, run.ParentID),
 		attribute.String(AttrFlowName, f.name),
 		attribute.String(AttrFlowStatus, run.Status),
 	))
@@ -62,7 +60,6 @@ func (f *Flow) runStepSpan(ctx context.Context, step Step, in State) (State, int
 	info, _ := ai.RunInfoFrom(ctx)
 	ctx, span := f.tracer().Start(ctx, spanNameFlowStep, trace.WithAttributes(
 		attribute.String(AttrFlowRunID, info.RunID),
-		attribute.String(AttrFlowParentID, info.ParentID),
 		attribute.String(AttrFlowName, f.name),
 		attribute.String(AttrFlowStepName, step.Name),
 	))

flow/otel_test.go

@@ -4,7 +4,6 @@ import (
 	"context"
 	"testing"
 
-	"go-micro.dev/v6/ai"
 	"go-micro.dev/v6/store"
 	"go.opentelemetry.io/otel/attribute"
 	"go.opentelemetry.io/otel/sdk/trace"
@@ -20,8 +19,7 @@ func TestFlowOpenTelemetrySpans(t *testing.T) {
 		return in, nil
 	}}
 	f := New("observed", WithCheckpoint(StoreCheckpoint(store.NewMemoryStore(), "observed")), TraceProvider(tp), Steps(step))
-	ctx := withTestRunInfo(context.Background(), "agent-run-otel")
-	if err := f.Execute(ctx, "start"); err != nil {
+	if err := f.Execute(context.Background(), "start"); err != nil {
 		t.Fatal(err)
 	}
 
@@ -34,12 +32,12 @@ func TestFlowOpenTelemetrySpans(t *testing.T) {
 		case spanNameFlowRun:
 			seen[spanNameFlowRun] = true
 			runID = attrs[AttrFlowRunID]
-			if attrs[AttrFlowName] != "observed" || attrs[AttrFlowStatus] != "done" || attrs[AttrFlowParentID] != "agent-run-otel" {
+			if attrs[AttrFlowName] != "observed" || attrs[AttrFlowStatus] != "done" {
 				t.Fatalf("run span attributes = %#v", attrs)
 			}
 		case spanNameFlowStep:
 			seen[spanNameFlowStep] = true
-			if attrs[AttrFlowName] != "observed" || attrs[AttrFlowStepName] != "inspect" || attrs[AttrFlowParentID] != "agent-run-otel" {
+			if attrs[AttrFlowName] != "observed" || attrs[AttrFlowStepName] != "inspect" {
 				t.Fatalf("step span attributes = %#v", attrs)
 			}
 		}
@@ -70,7 +68,3 @@ func flowSpanAttributes(attrs []attribute.KeyValue) map[string]string {
 	}
 	return out
 }
-
-func withTestRunInfo(ctx context.Context, runID string) context.Context {
-	return ai.WithRunInfo(ctx, ai.RunInfo{RunID: runID, Agent: "planner"})
-}

flow/steps.go

@@ -74,14 +74,13 @@ type StepRecord struct {
 // saves and loads. It is retained for success and failure unless the flow
 // opts into cleanup with DeleteOnSuccess.
 type Run struct {
-	ID       string       `json:"id"`
-	ParentID string       `json:"parent_id,omitempty"`
-	Flow     string       `json:"flow"`
-	State    State        `json:"state"`
-	Steps    []StepRecord `json:"steps"`
-	Status   string       `json:"status"` // running | done | failed
-	Started  time.Time    `json:"started"`
-	Updated  time.Time    `json:"updated"`
+	ID      string       `json:"id"`
+	Flow    string       `json:"flow"`
+	State   State        `json:"state"`
+	Steps   []StepRecord `json:"steps"`
+	Status  string       `json:"status"` // running | done | failed
+	Started time.Time    `json:"started"`
+	Updated time.Time    `json:"updated"`
 }
 
 // Checkpoint persists and restores flow runs so a run survives a crash
@@ -310,17 +309,12 @@ func (f *Flow) startRun(ctx context.Context, data string) (Run, error) {
 	if err := validateSteps(f.opts.Steps); err != nil {
 		return Run{}, err
 	}
-	parentID := ""
-	if info, ok := ai.RunInfoFrom(ctx); ok {
-		parentID = info.RunID
-	}
 	run := Run{
-		ID:       uuid.New().String(),
-		ParentID: parentID,
-		Flow:     f.name,
-		State:    State{Stage: f.opts.Steps[0].Name, Data: []byte(data)},
-		Status:   "running",
-		Started:  time.Now(),
+		ID:      uuid.New().String(),
+		Flow:    f.name,
+		State:   State{Stage: f.opts.Steps[0].Name, Data: []byte(data)},
+		Status:  "running",
+		Started: time.Now(),
 	}
 	for _, s := range f.opts.Steps {
 		run.Steps = append(run.Steps, StepRecord{Name: s.Name, Status: "pending"})
@@ -331,9 +325,6 @@ func (f *Flow) startRun(ctx context.Context, data string) (Run, error) {
 // Resume continues a persisted run by id, picking up at the step it
 // stopped on. Completed runs are a no-op.
 func (f *Flow) Resume(ctx context.Context, runID string) error {
-	ctx, cancel := f.withTimeout(ctx)
-	defer cancel()
-
 	if err := validateSteps(f.opts.Steps); err != nil {
 		return err
 	}
@@ -363,9 +354,6 @@ func (f *Flow) Resume(ctx context.Context, runID string) error {
 // stops and returns that run id with the error so callers can log, alert, or
 // retry later without hiding the failing run.
 func (f *Flow) ResumePending(ctx context.Context) (string, error) {
-	ctx, cancel := f.withTimeout(ctx)
-	defer cancel()
-
 	runs, err := f.Pending(ctx)
 	if err != nil {
 		return "", err
@@ -402,7 +390,7 @@ func (f *Flow) Pending(ctx context.Context) ([]Run, error) {
 func (f *Flow) runFrom(ctx context.Context, run Run) (Run, error) {
 	steps := f.opts.Steps
 	ctx = withDeps(ctx, &runDeps{client: f.client, model: f.model, tools: f.toolSet})
-	ctx = ai.WithRunInfo(ctx, ai.RunInfo{RunID: run.ID, ParentID: run.ParentID, Agent: f.name, Flow: f.name})
+	ctx = ai.WithRunInfo(ctx, ai.RunInfo{RunID: run.ID, Agent: f.name})
 	ctx, finishSpan := f.startRunSpan(ctx, run)
 	var spanErr error
 	defer func() { finishSpan(run, spanErr) }()
@@ -490,10 +478,6 @@ func (f *Flow) runStep(ctx context.Context, step Step, in State) (State, int, er
 		if err := ctx.Err(); err != nil {
 			return in, attempt - 1, err
 		}
-		if info, ok := ai.RunInfoFrom(ctx); ok {
-			info.Step = step.Name
-			ctx = ai.WithRunInfo(ctx, info)
-		}
 		out, err := step.Run(ctx, in)
 		if err == nil {
 			return out, attempt, nil

flow/steps_test.go

@@ -99,13 +99,11 @@ func TestFlowStepContextIncludesRunInfo(t *testing.T) {
 		return in, nil
 	}}
 
-	mem := store.NewMemoryStore()
 	f := New("correlated",
-		WithCheckpoint(StoreCheckpoint(mem, "correlated")),
+		WithCheckpoint(StoreCheckpoint(store.NewMemoryStore(), "correlated")),
 		Steps(step),
 	)
-	ctx := ai.WithRunInfo(context.Background(), ai.RunInfo{RunID: "agent-run-1", Agent: "planner"})
-	if err := f.Execute(ctx, "start"); err != nil {
+	if err := f.Execute(context.Background(), "start"); err != nil {
 		t.Fatalf("Execute: %v", err)
 	}
 	if got.Agent != "correlated" {
@@ -114,22 +112,6 @@ func TestFlowStepContextIncludesRunInfo(t *testing.T) {
 	if got.RunID == "" {
 		t.Fatal("RunInfo.RunID is empty")
 	}
-	if got.Flow != "correlated" {
-		t.Fatalf("RunInfo.Flow = %q, want correlated", got.Flow)
-	}
-	if got.ParentID != "agent-run-1" {
-		t.Fatalf("RunInfo.ParentID = %q, want agent-run-1", got.ParentID)
-	}
-	if got.Step != "inspect" {
-		t.Fatalf("RunInfo.Step = %q, want inspect", got.Step)
-	}
-	runs, err := StoreCheckpoint(mem, "correlated").List(context.Background())
-	if err != nil {
-		t.Fatalf("List: %v", err)
-	}
-	if len(runs) != 1 || runs[0].ParentID != "agent-run-1" {
-		t.Fatalf("persisted parent id = %+v, want agent-run-1", runs)
-	}
 }
 
 func TestFlowResumePendingResumesOldestRunsUntilFailure(t *testing.T) {
@@ -279,50 +261,6 @@ func TestFlowStepRetryBackoffWaitsBetweenAttempts(t *testing.T) {
 	}
 }
 
-func TestFlowTimeoutStopsRetryBackoff(t *testing.T) {
-	var attempts int
-	step := Step{Name: "slow", Run: func(_ context.Context, in State) (State, error) {
-		attempts++
-		return in, errors.New("transient")
-	}}
-
-	f := New("timeout-backoff",
-		WithCheckpoint(StoreCheckpoint(store.NewMemoryStore(), "timeout-backoff")),
-		Timeout(20*time.Millisecond),
-		Retry(1),
-		RetryBackoff(time.Hour),
-		Steps(step),
-	)
-	err := f.Execute(context.Background(), "")
-	if err == nil {
-		t.Fatal("expected the timed-out run to fail")
-	}
-	if !errors.Is(err, context.DeadlineExceeded) {
-		t.Errorf("want a context deadline error, got %v", err)
-	}
-	if attempts != 1 {
-		t.Errorf("timeout should stop during backoff before retrying, got %d attempts", attempts)
-	}
-}
-
-func TestFlowTimeoutRespectsExistingDeadline(t *testing.T) {
-	ctx, cancel := context.WithTimeout(context.Background(), time.Hour)
-	defer cancel()
-
-	f := New("existing-deadline", Timeout(time.Millisecond))
-	got, stop := f.withTimeout(ctx)
-	defer stop()
-
-	wantDeadline, _ := ctx.Deadline()
-	gotDeadline, ok := got.Deadline()
-	if !ok {
-		t.Fatal("expected the existing deadline to remain set")
-	}
-	if !gotDeadline.Equal(wantDeadline) {
-		t.Fatalf("deadline = %v, want existing deadline %v", gotDeadline, wantDeadline)
-	}
-}
-
 func TestFlowStepRetryBackoffStopsOnCancel(t *testing.T) {
 	ctx, cancel := context.WithCancel(context.Background())
 	var attempts int

gateway/a2a/a2a.go

@@ -20,16 +20,15 @@
 //
 // 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.
+// Task event), `tasks/get`, and Agent Card discovery. Multi-turn
+// `input-required`, `tasks/resubscribe`, and push notifications are
+// advertised as unsupported and are follow-ups.
 package a2a
 
 import (
 	"context"
 	"encoding/json"
 	"fmt"
-	"io"
 	"log"
 	"net/http"
 	"strings"
@@ -37,7 +36,6 @@ import (
 	"time"
 
 	"github.com/google/uuid"
-	"go-micro.dev/v6/ai"
 	"go-micro.dev/v6/client"
 	codecbytes "go-micro.dev/v6/codec/bytes"
 	"go-micro.dev/v6/registry"
@@ -94,9 +92,6 @@ func New(opts Options) *Gateway {
 // to Agent.Chat (the gateway) or an in-process Ask (an embedded agent).
 type Invoke func(ctx context.Context, text string) (string, error)
 
-// StreamInvoke runs an agent for one message and returns streaming output chunks.
-type StreamInvoke func(ctx context.Context, text string) (ai.Stream, error)
-
 // NewAgentHandler returns an http.Handler that serves the A2A protocol
 // for a single agent: its Agent Card at / and /.well-known/agent.json,
 // and the JSON-RPC endpoint at /. invoke runs the agent. This is what an
@@ -112,19 +107,6 @@ func NewAgentHandler(card AgentCard, invoke Invoke) http.Handler {
 	return mux
 }
 
-// NewAgentStreamHandler is like NewAgentHandler, but serves A2A message/stream
-// by forwarding model chunks as server-sent task updates when stream is non-nil.
-func NewAgentStreamHandler(card AgentCard, invoke Invoke, stream StreamInvoke) http.Handler {
-	d := newDispatcher()
-	mux := http.NewServeMux()
-	card.URL = strings.TrimRight(card.URL, "/")
-	serveCard := func(w http.ResponseWriter, _ *http.Request) { writeJSON(w, http.StatusOK, card) }
-	mux.HandleFunc("GET /{$}", serveCard)
-	mux.HandleFunc("GET /.well-known/agent.json", serveCard)
-	mux.HandleFunc("POST /{$}", func(w http.ResponseWriter, r *http.Request) { d.serveWithStream(w, r, invoke, stream) })
-	return mux
-}
-
 // Serve creates a gateway and serves it on opts.Address (blocking).
 func Serve(opts Options) error {
 	g := New(opts)
@@ -225,21 +207,10 @@ type Task struct {
 	Kind      string     `json:"kind"` // "task"
 }
 
-// PushNotificationConfig tells the gateway where to POST task updates for a
-// task. The gateway stores one config per task and delivers best-effort JSON
-// task snapshots whenever that task changes.
-type PushNotificationConfig struct {
-	URL            string            `json:"url"`
-	Token          string            `json:"token,omitempty"`
-	Authentication map[string]string `json:"authentication,omitempty"`
-}
-
 // Task states (JSON-RPC binding wire values).
 const (
-	stateCompleted     = "completed"
-	stateFailed        = "failed"
-	stateWorking       = "working"
-	stateInputRequired = "input-required"
+	stateCompleted = "completed"
+	stateFailed    = "failed"
 )
 
 // JSON-RPC envelopes.
@@ -344,7 +315,7 @@ func Card(name, url, description string, services []string) AgentCard {
 		URL:             url,
 		Version:         "1.0.0",
 		ProtocolVersion: protocolVersion,
-		Capabilities:    Capabilities{Streaming: true, PushNotifications: true},
+		Capabilities:    Capabilities{Streaming: true, PushNotifications: false},
 		// The agent converses over a single Chat endpoint; advertise that
 		// as one skill, tagged with the services it manages.
 		DefaultInputModes:  []string{"text/plain"},
@@ -425,22 +396,14 @@ func (g *Gateway) handleRPC(w http.ResponseWriter, r *http.Request) {
 // retains recent tasks for tasks/get. It is shared by the gateway (one
 // per registry) and embedded agents (one per agent).
 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
+	mu    sync.Mutex
+	tasks map[string]*Task
+	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{}{}}
-}
+func newDispatcher() *dispatcher { return &dispatcher{tasks: map[string]*Task{}} }
 
 func (d *dispatcher) serve(w http.ResponseWriter, r *http.Request, invoke Invoke) {
-	d.serveWithStream(w, r, invoke, nil)
-}
-
-func (d *dispatcher) serveWithStream(w http.ResponseWriter, r *http.Request, invoke Invoke, streamInvoke StreamInvoke) {
 	var req rpcRequest
 	if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
 		writeRPC(w, nil, nil, &rpcError{Code: errParse, Message: "parse error"})
@@ -455,22 +418,14 @@ func (d *dispatcher) serveWithStream(w http.ResponseWriter, r *http.Request, inv
 	case "message/send":
 		d.send(requestContext(r.Context()), w, req, invoke)
 	case "message/stream":
-		if streamInvoke != nil {
-			d.streamChunks(requestContext(r.Context()), w, req, streamInvoke)
-			return
-		}
 		d.stream(requestContext(r.Context()), w, req, invoke)
 	case "tasks/get":
 		d.get(w, req)
-	case "tasks/pushNotificationConfig/set":
-		d.setPushConfig(w, req)
-	case "tasks/pushNotificationConfig/get":
-		d.getPushConfig(w, req)
 	case "tasks/cancel":
 		// 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})
 	}
@@ -505,66 +460,6 @@ 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) {
-	var p sendParams
-	if err := json.Unmarshal(req.Params, &p); err != nil {
-		writeRPC(w, req.ID, nil, &rpcError{Code: errInvalidParams, Message: "invalid params"})
-		return
-	}
-	text := textOf(p.Message.Parts)
-	if text == "" {
-		writeRPC(w, req.ID, nil, &rpcError{Code: errInvalidParams, Message: "message has no text part"})
-		return
-	}
-	stream, err := invoke(ctx, text)
-	if err != nil {
-		writeRPC(w, req.ID, nil, &rpcError{Code: errInternal, Message: err.Error()})
-		return
-	}
-	defer stream.Close()
-	w.Header().Set("Content-Type", "text/event-stream")
-	w.Header().Set("Cache-Control", "no-cache")
-	w.Header().Set("Connection", "keep-alive")
-	w.WriteHeader(http.StatusOK)
-	enc := json.NewEncoder(sseWriter{w: w})
-	flush := func() {
-		if f, ok := w.(http.Flusher); ok {
-			f.Flush()
-		}
-	}
-	taskID := uuid.New().String()
-	contextID := p.Message.ContextID
-	if contextID == "" {
-		contextID = uuid.New().String()
-	}
-	var reply strings.Builder
-	for {
-		chunk, err := stream.Recv()
-		if err == io.EOF {
-			task := taskFromReplyWithIDs(p.Message, reply.String(), stateCompleted, taskID, contextID)
-			d.store(task)
-			_ = enc.Encode(rpcResponse{JSONRPC: "2.0", ID: req.ID, Result: task})
-			flush()
-			return
-		}
-		if err != nil {
-			task := taskFromReplyWithIDs(p.Message, "error: "+err.Error(), stateFailed, taskID, contextID)
-			d.store(task)
-			_ = enc.Encode(rpcResponse{JSONRPC: "2.0", ID: req.ID, Result: task, Error: &rpcError{Code: errInternal, Message: err.Error()}})
-			flush()
-			return
-		}
-		if chunk == nil || chunk.Reply == "" {
-			continue
-		}
-		reply.WriteString(chunk.Reply)
-		task := taskFromReplyWithIDs(p.Message, reply.String(), stateWorking, taskID, contextID)
-		d.store(task)
-		_ = enc.Encode(rpcResponse{JSONRPC: "2.0", ID: req.ID, Result: task})
-		flush()
-	}
-}
-
 func (d *dispatcher) run(ctx context.Context, params json.RawMessage, invoke Invoke) (*Task, *rpcError) {
 	var p sendParams
 	if err := json.Unmarshal(params, &p); err != nil {
@@ -576,16 +471,24 @@ func (d *dispatcher) run(ctx context.Context, params json.RawMessage, invoke Inv
 	}
 
 	reply, err := invoke(ctx, text)
-	state := stateCompleted
-	if err != nil {
-		reply = "error: " + err.Error()
-		state = stateFailed
-		if isInputRequiredError(err) {
-			reply = err.Error()
-			state = stateInputRequired
-		}
+	contextID := p.Message.ContextID
+	if contextID == "" {
+		contextID = uuid.New().String()
+	}
+	task := &Task{
+		ID:        uuid.New().String(),
+		ContextID: contextID,
+		Kind:      "task",
+		History:   []Message{p.Message},
+		Status:    TaskStatus{Timestamp: time.Now().UTC().Format(time.RFC3339)},
+	}
+	if err != nil {
+		task.Status.State = stateFailed
+		task.Artifacts = []Artifact{textArtifact("error: " + err.Error())}
+	} else {
+		task.Status.State = stateCompleted
+		task.Artifacts = []Artifact{textArtifact(reply)}
 	}
-	task := d.taskFromReply(p.Message, reply, state)
 	d.store(task)
 	return task, nil
 }
@@ -594,49 +497,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 == "" {
@@ -653,47 +513,6 @@ func (d *dispatcher) get(w http.ResponseWriter, req rpcRequest) {
 	writeRPC(w, req.ID, task, nil)
 }
 
-type pushConfigParams struct {
-	ID                     string                 `json:"id"`
-	PushNotificationConfig PushNotificationConfig `json:"pushNotificationConfig"`
-}
-
-func (d *dispatcher) setPushConfig(w http.ResponseWriter, req rpcRequest) {
-	var p pushConfigParams
-	if err := json.Unmarshal(req.Params, &p); err != nil || p.ID == "" || p.PushNotificationConfig.URL == "" {
-		writeRPC(w, req.ID, nil, &rpcError{Code: errInvalidParams, Message: "invalid params"})
-		return
-	}
-	d.mu.Lock()
-	task := d.tasks[p.ID]
-	if task != nil {
-		d.pushConfigs[p.ID] = p.PushNotificationConfig
-	}
-	d.mu.Unlock()
-	if task == nil {
-		writeRPC(w, req.ID, nil, &rpcError{Code: errTaskNotFound, Message: "task not found"})
-		return
-	}
-	writeRPC(w, req.ID, map[string]any{"id": p.ID, "pushNotificationConfig": p.PushNotificationConfig}, nil)
-	go d.deliverPush(p.ID, task)
-}
-
-func (d *dispatcher) getPushConfig(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
-	}
-	d.mu.Lock()
-	cfg, ok := d.pushConfigs[p.ID]
-	d.mu.Unlock()
-	if !ok {
-		writeRPC(w, req.ID, nil, &rpcError{Code: errTaskNotFound, Message: "push notification config not found"})
-		return
-	}
-	writeRPC(w, req.ID, map[string]any{"id": p.ID, "pushNotificationConfig": cfg}, nil)
-}
-
 // ---------------------------------------------------------------------------
 // agent RPC
 // ---------------------------------------------------------------------------
@@ -722,134 +541,13 @@ 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]
+	defer d.mu.Unlock()
 	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
-	var history []Message
-	if taskID != "" {
-		d.mu.Lock()
-		prev := d.tasks[taskID]
-		if prev != nil {
-			contextID = prev.ContextID
-			history = append(history, prev.History...)
-		}
-		d.mu.Unlock()
-	}
-	if taskID == "" {
-		taskID = uuid.New().String()
-	}
-	if contextID == "" {
-		contextID = uuid.New().String()
-	}
-	return taskFromReplyWithIDsAndHistory(input, reply, state, taskID, contextID, history)
-}
-
-func taskFromReplyWithIDs(input Message, reply, state, taskID, contextID string) *Task {
-	return taskFromReplyWithIDsAndHistory(input, reply, state, taskID, contextID, nil)
-}
-
-func taskFromReplyWithIDsAndHistory(input Message, reply, state, taskID, contextID string, history []Message) *Task {
-	input.TaskID = taskID
-	input.ContextID = contextID
-	if input.Kind == "" {
-		input.Kind = "message"
-	}
-	task := &Task{
-		ID:        taskID,
-		ContextID: contextID,
-		Kind:      "task",
-		History:   append(append([]Message{}, history...), input),
-		Status:    TaskStatus{State: state, Timestamp: time.Now().UTC().Format(time.RFC3339)},
-		Artifacts: []Artifact{textArtifact(reply)},
-	}
-	task.History = append(task.History, Message{
-		Role:      "agent",
-		Parts:     []Part{{Kind: "text", Text: reply}},
-		MessageID: uuid.New().String(),
-		TaskID:    task.ID,
-		ContextID: task.ContextID,
-		Kind:      "message",
-	})
-	return task
-}
-
-func (d *dispatcher) deliverPush(taskID string, task *Task) {
-	d.mu.Lock()
-	cfg, ok := d.pushConfigs[taskID]
-	d.mu.Unlock()
-	if !ok || cfg.URL == "" || task == nil {
-		return
-	}
-	body, err := json.Marshal(task)
-	if err != nil {
-		return
-	}
-	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
-	defer cancel()
-	req, err := http.NewRequestWithContext(ctx, http.MethodPost, cfg.URL, strings.NewReader(string(body)))
-	if err != nil {
-		return
-	}
-	req.Header.Set("Content-Type", "application/json")
-	if cfg.Token != "" {
-		req.Header.Set("Authorization", "Bearer "+cfg.Token)
-	}
-	resp, err := http.DefaultClient.Do(req)
-	if err == nil && resp.Body != nil {
-		_ = resp.Body.Close()
 	}
 }
 

gateway/a2a/a2a_test.go

@@ -4,9 +4,7 @@ import (
 	"bytes"
 	"context"
 	"encoding/json"
-	"errors"
 	"fmt"
-	"io"
 	"net/http"
 	"net/http/httptest"
 	"strings"
@@ -14,7 +12,6 @@ import (
 	"time"
 
 	pb "go-micro.dev/v6/agent/proto"
-	"go-micro.dev/v6/ai"
 	"go-micro.dev/v6/client"
 	"go-micro.dev/v6/registry"
 	"go-micro.dev/v6/selector"
@@ -107,12 +104,6 @@ func TestMessageSendAndGet(t *testing.T) {
 	if len(task.Artifacts) != 1 || textOf(task.Artifacts[0].Parts) != "pong" {
 		t.Fatalf("artifact = %+v, want text 'pong'", task.Artifacts)
 	}
-	if len(task.History) != 2 || task.History[1].Role != "agent" || textOf(task.History[1].Parts) != "pong" {
-		t.Fatalf("history = %+v, want user turn followed by agent reply", task.History)
-	}
-	if task.History[1].TaskID != task.ID || task.History[1].ContextID != task.ContextID {
-		t.Fatalf("agent history linkage = task %q/%q context %q/%q", task.History[1].TaskID, task.ID, task.History[1].ContextID, task.ContextID)
-	}
 
 	got := rpcTask(t, ts.URL+"/agents/echo", `{
 		"jsonrpc":"2.0","id":2,"method":"tasks/get","params":{"id":"`+task.ID+`"}}`)
@@ -121,89 +112,6 @@ func TestMessageSendAndGet(t *testing.T) {
 	}
 }
 
-func TestMessageSendContinuesExistingTask(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":"first"}]}}}`, func(_ context.Context, text string) (string, error) {
-		return "reply to " + text, nil
-	})
-
-	secondBody := 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":"second"}]}}}`, first.ID, first.ContextID)
-	second := rpcTaskFromBody(t, d, secondBody, func(_ context.Context, text string) (string, error) {
-		return "reply to " + text, nil
-	})
-
-	if second.ID != first.ID || second.ContextID != first.ContextID {
-		t.Fatalf("continued task identity = %s/%s, want %s/%s", second.ID, second.ContextID, first.ID, first.ContextID)
-	}
-	if len(second.History) != 4 {
-		t.Fatalf("continued history len = %d, want 4: %+v", len(second.History), second.History)
-	}
-	if textOf(second.History[0].Parts) != "first" || textOf(second.History[2].Parts) != "second" {
-		t.Fatalf("continued history did not preserve turns: %+v", second.History)
-	}
-
-	got := rpcTaskFromDispatcher(t, d, first.ID)
-	if got.ID != first.ID || len(got.History) != 4 {
-		t.Fatalf("stored continued task = %+v", got)
-	}
-}
-
-func TestPushNotificationConfigDeliversTaskUpdates(t *testing.T) {
-	d := newDispatcher()
-	updates := make(chan Task, 2)
-	push := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
-		if got := r.Header.Get("Authorization"); got != "Bearer secret" {
-			t.Errorf("authorization = %q, want bearer token", got)
-		}
-		var task Task
-		if err := json.NewDecoder(r.Body).Decode(&task); err != nil {
-			t.Errorf("decode push task: %v", err)
-			return
-		}
-		updates <- task
-		w.WriteHeader(http.StatusAccepted)
-	}))
-	defer push.Close()
-
-	task := rpcTaskFromBody(t, d, `{
-		"jsonrpc":"2.0","id":1,"method":"message/send",
-		"params":{"message":{"role":"user","kind":"message","messageId":"m1",
-			"parts":[{"kind":"text","text":"ping"}]}}}`, func(_ context.Context, text string) (string, error) {
-		return "pong", nil
-	})
-
-	body := fmt.Sprintf(`{"jsonrpc":"2.0","id":2,"method":"tasks/pushNotificationConfig/set","params":{"id":"%s","pushNotificationConfig":{"url":"%s","token":"secret"}}}`, task.ID, push.URL)
-	var setResp struct {
-		Result struct {
-			ID                     string                 `json:"id"`
-			PushNotificationConfig PushNotificationConfig `json:"pushNotificationConfig"`
-		} `json:"result"`
-		Error *rpcError `json:"error"`
-	}
-	rpcDispatcher(t, d, body, nil, &setResp)
-	if setResp.Error != nil {
-		t.Fatalf("set push config error: %+v", setResp.Error)
-	}
-	if setResp.Result.ID != task.ID || setResp.Result.PushNotificationConfig.URL != push.URL {
-		t.Fatalf("set push config result = %+v", setResp.Result)
-	}
-
-	select {
-	case got := <-updates:
-		if got.ID != task.ID || got.Status.State != stateCompleted || textOf(got.Artifacts[0].Parts) != "pong" {
-			t.Fatalf("push update = %+v, want completed task", got)
-		}
-	case <-time.After(2 * time.Second):
-		t.Fatal("timed out waiting for push update")
-	}
-}
-
 func TestMessageSendUsesRequestContext(t *testing.T) {
 	d := newDispatcher()
 	ctx, cancel := context.WithCancel(context.Background())
@@ -238,9 +146,6 @@ func TestMessageSendUsesRequestContext(t *testing.T) {
 	if len(resp.Result.Artifacts) != 1 || textOf(resp.Result.Artifacts[0].Parts) != "error: context canceled" {
 		t.Fatalf("artifact = %+v, want context cancellation", resp.Result.Artifacts)
 	}
-	if len(resp.Result.History) != 2 || resp.Result.History[1].Role != "agent" || textOf(resp.Result.History[1].Parts) != "error: context canceled" {
-		t.Fatalf("history = %+v, want failed agent reply recorded", resp.Result.History)
-	}
 }
 
 func TestMessageStream(t *testing.T) {
@@ -279,316 +184,6 @@ func TestMessageStream(t *testing.T) {
 	}
 }
 
-type sliceStream struct {
-	chunks []string
-	err    error
-}
-
-func (s *sliceStream) Recv() (*ai.Response, error) {
-	if len(s.chunks) == 0 {
-		if s.err != nil {
-			err := s.err
-			s.err = nil
-			return nil, err
-		}
-		return nil, io.EOF
-	}
-	next := s.chunks[0]
-	s.chunks = s.chunks[1:]
-	return &ai.Response{Reply: next}, nil
-}
-
-func (s *sliceStream) Close() error { return nil }
-
-func TestMessageStreamChunksStoreFinalTask(t *testing.T) {
-	d := newDispatcher()
-	body := `{"jsonrpc":"2.0","id":1,"method":"message/stream","params":{"message":{"role":"user","parts":[{"kind":"text","text":"ping"}],"kind":"message"}}}`
-	req := httptest.NewRequest(http.MethodPost, "/", bytes.NewBufferString(body))
-	rr := httptest.NewRecorder()
-
-	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 &sliceStream{chunks: []string{"po", "ng"}}, nil
-	})
-
-	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) != 3 {
-		t.Fatalf("events = %d, want 3; body %s", len(events), rr.Body.String())
-	}
-	for i, event := range events {
-		if event.Error != nil {
-			t.Fatalf("event %d error: %+v", i, event.Error)
-		}
-		if event.Result.ID != events[0].Result.ID || event.Result.ContextID != events[0].Result.ContextID {
-			t.Fatalf("event %d changed task identity: %+v vs %+v", i, event.Result, events[0].Result)
-		}
-	}
-	if events[0].Result.Status.State != stateWorking || textOf(events[0].Result.Artifacts[0].Parts) != "po" {
-		t.Fatalf("first event = %+v, want working po", events[0].Result)
-	}
-	final := events[len(events)-1].Result
-	if final.Status.State != stateCompleted || textOf(final.Artifacts[0].Parts) != "pong" {
-		t.Fatalf("final event = %+v, want completed pong", final)
-	}
-
-	got := rpcTaskFromDispatcher(t, d, final.ID)
-	if got.ID != final.ID || got.Status.State != stateCompleted || textOf(got.Artifacts[0].Parts) != "pong" {
-		t.Fatalf("stored task = %+v, want final", got)
-	}
-}
-
-type contextStream struct {
-	ctx    context.Context
-	closed chan struct{}
-}
-
-func (s *contextStream) Recv() (*ai.Response, error) {
-	<-s.ctx.Done()
-	return nil, s.ctx.Err()
-}
-
-func (s *contextStream) Close() error {
-	close(s.closed)
-	return nil
-}
-
-func TestMessageStreamChunksPropagatesCancellationAndClosesStream(t *testing.T) {
-	d := newDispatcher()
-	ctx, cancel := context.WithCancel(context.Background())
-	closed := make(chan struct{})
-	body := `{"jsonrpc":"2.0","id":1,"method":"message/stream","params":{"message":{"role":"user","parts":[{"kind":"text","text":"ping"}],"kind":"message"}}}`
-	req := httptest.NewRequest(http.MethodPost, "/", bytes.NewBufferString(body)).WithContext(ctx)
-	rr := httptest.NewRecorder()
-	cancel()
-
-	d.serveWithStream(rr, req, nil, func(ctx context.Context, text string) (ai.Stream, error) {
-		if text != "ping" {
-			t.Fatalf("stream text = %q, want ping", text)
-		}
-		return &contextStream{ctx: ctx, closed: closed}, nil
-	})
-
-	select {
-	case <-closed:
-	case <-time.After(time.Second):
-		t.Fatal("stream was not closed")
-	}
-
-	var events []struct {
-		Result Task      `json:"result"`
-		Error  *rpcError `json:"error"`
-	}
-	for _, line := range strings.Split(strings.TrimSpace(rr.Body.String()), "\n") {
-		line = strings.TrimSpace(line)
-		if line == "" {
-			continue
-		}
-		line = strings.TrimPrefix(line, "data: ")
-		var event struct {
-			Result Task      `json:"result"`
-			Error  *rpcError `json:"error"`
-		}
-		if err := json.Unmarshal([]byte(line), &event); err != nil {
-			t.Fatalf("decode event %q: %v", line, err)
-		}
-		events = append(events, event)
-	}
-	if len(events) != 1 {
-		t.Fatalf("events = %d, want 1; body %s", len(events), rr.Body.String())
-	}
-	event := events[0]
-	if event.Error == nil || event.Error.Code != errInternal || event.Error.Message != context.Canceled.Error() {
-		t.Fatalf("error = %+v, want context cancellation", event.Error)
-	}
-	if event.Result.Status.State != stateFailed || textOf(event.Result.Artifacts[0].Parts) != "error: context canceled" {
-		t.Fatalf("failed task = %+v, want context cancellation artifact", event.Result)
-	}
-
-	got := rpcTaskFromDispatcher(t, d, event.Result.ID)
-	if got.Status.State != stateFailed || textOf(got.Artifacts[0].Parts) != "error: context canceled" {
-		t.Fatalf("stored task = %+v, want failed cancellation", got)
-	}
-}
-
-func TestTasksResubscribeStreamsCurrentAndSubsequentEvents(t *testing.T) {
-	d := newDispatcher()
-	initial := &Task{ID: "task-1", ContextID: "ctx-1", Kind: "task", Status: TaskStatus{State: stateWorking, Timestamp: time.Now().UTC().Format(time.RFC3339)}}
-	d.store(initial)
-
-	ctx, cancel := context.WithCancel(context.Background())
-	defer cancel()
-	req := httptest.NewRequest(http.MethodPost, "/", bytes.NewBufferString(`{"jsonrpc":"2.0","id":1,"method":"tasks/resubscribe","params":{"id":"task-1"}}`)).WithContext(ctx)
-	rw := newFlushRecorder()
-	done := make(chan struct{})
-	go func() {
-		d.serve(rw, req, nil)
-		close(done)
-	}()
-
-	first := rw.next(t)
-	if first.Result.ID != initial.ID || first.Result.Status.State != stateWorking {
-		t.Fatalf("first resubscribe event = %+v, want current working task", first.Result)
-	}
-
-	final := &Task{ID: "task-1", ContextID: "ctx-1", Kind: "task", Status: TaskStatus{State: stateCompleted, Timestamp: time.Now().UTC().Format(time.RFC3339)}, Artifacts: []Artifact{textArtifact("done")}}
-	d.store(final)
-	second := rw.next(t)
-	if second.Result.ID != final.ID || second.Result.Status.State != stateCompleted || textOf(second.Result.Artifacts[0].Parts) != "done" {
-		t.Fatalf("second resubscribe event = %+v, want completed update", second.Result)
-	}
-	select {
-	case <-done:
-	case <-time.After(time.Second):
-		t.Fatal("resubscribe did not return after terminal update")
-	}
-}
-
-func TestInputRequiredErrorCreatesContinuableTask(t *testing.T) {
-	d := newDispatcher()
-	first := rpcTaskFromBody(t, d, `{
-		"jsonrpc":"2.0","id":1,"method":"message/send",
-		"params":{"message":{"role":"user","kind":"message","messageId":"m1",
-			"parts":[{"kind":"text","text":"start approval"}]}}}`, func(_ context.Context, text string) (string, error) {
-		return "", errors.New("agent run run-1 paused for approval: waiting for operator")
-	})
-	if first.Status.State != stateInputRequired {
-		t.Fatalf("state = %q, want input-required", first.Status.State)
-	}
-	if textOf(first.Artifacts[0].Parts) != "agent run run-1 paused for approval: waiting for operator" {
-		t.Fatalf("artifact = %+v, want handoff message", first.Artifacts)
-	}
-
-	body := fmt.Sprintf(`{
-		"jsonrpc":"2.0","id":2,"method":"message/send",
-		"params":{"message":{"role":"user","kind":"message","messageId":"m2","taskId":"%s","contextId":"%s",
-			"parts":[{"kind":"text","text":"approved"}]}}}`, first.ID, first.ContextID)
-	continued := rpcTaskFromBody(t, d, body, func(_ context.Context, text string) (string, error) {
-		return "continued after " + text, nil
-	})
-	if continued.ID != first.ID || continued.ContextID != first.ContextID {
-		t.Fatalf("continued identity = %s/%s, want %s/%s", continued.ID, continued.ContextID, first.ID, first.ContextID)
-	}
-	if continued.Status.State != stateCompleted || len(continued.History) != 4 {
-		t.Fatalf("continued task = %+v, want completed task with prior input-required history", continued)
-	}
-	if textOf(continued.History[1].Parts) != "agent run run-1 paused for approval: waiting for operator" || textOf(continued.History[3].Parts) != "continued after approved" {
-		t.Fatalf("continued history = %+v", continued.History)
-	}
-}
-
-type flushRecorder struct {
-	*httptest.ResponseRecorder
-	ch chan string
-}
-
-func newFlushRecorder() *flushRecorder {
-	return &flushRecorder{ResponseRecorder: httptest.NewRecorder(), ch: make(chan string, 16)}
-}
-
-func (r *flushRecorder) Flush() {
-	body := r.Body.String()
-	r.Body.Reset()
-	for _, line := range strings.Split(strings.TrimSpace(body), "\n") {
-		line = strings.TrimSpace(line)
-		if line != "" {
-			r.ch <- line
-		}
-	}
-}
-
-func (r *flushRecorder) next(t *testing.T) struct {
-	Result Task      `json:"result"`
-	Error  *rpcError `json:"error"`
-} {
-	t.Helper()
-	select {
-	case line := <-r.ch:
-		line = strings.TrimPrefix(line, "data: ")
-		var event struct {
-			Result Task      `json:"result"`
-			Error  *rpcError `json:"error"`
-		}
-		if err := json.Unmarshal([]byte(line), &event); err != nil {
-			t.Fatalf("decode event %q: %v", line, err)
-		}
-		if event.Error != nil {
-			t.Fatalf("event error: %+v", event.Error)
-		}
-		return event
-	case <-time.After(time.Second):
-		t.Fatal("timed out waiting for SSE event")
-	}
-	return struct {
-		Result Task      `json:"result"`
-		Error  *rpcError `json:"error"`
-	}{}
-}
-
-func rpcTaskFromDispatcher(t *testing.T, d *dispatcher, id string) Task {
-	t.Helper()
-	body := fmt.Sprintf(`{"jsonrpc":"2.0","id":2,"method":"tasks/get","params":{"id":"%s"}}`, id)
-	req := httptest.NewRequest(http.MethodPost, "/", bytes.NewBufferString(body))
-	rr := httptest.NewRecorder()
-	d.serve(rr, req, nil)
-	var resp struct {
-		Result Task      `json:"result"`
-		Error  *rpcError `json:"error"`
-	}
-	if err := json.NewDecoder(rr.Result().Body).Decode(&resp); err != nil {
-		t.Fatalf("decode tasks/get: %v", err)
-	}
-	if resp.Error != nil {
-		t.Fatalf("tasks/get error: %+v", resp.Error)
-	}
-	return resp.Result
-}
-
-func rpcTaskFromBody(t *testing.T, d *dispatcher, body string, invoke Invoke) Task {
-	t.Helper()
-	var resp struct {
-		Result Task      `json:"result"`
-		Error  *rpcError `json:"error"`
-	}
-	rpcDispatcher(t, d, body, invoke, &resp)
-	if resp.Error != nil {
-		t.Fatalf("rpc error: %+v", resp.Error)
-	}
-	return resp.Result
-}
-
-func rpcDispatcher(t *testing.T, d *dispatcher, body string, invoke Invoke, v any) {
-	t.Helper()
-	req := httptest.NewRequest(http.MethodPost, "/", bytes.NewBufferString(body))
-	rr := httptest.NewRecorder()
-	d.serve(rr, req, invoke)
-	if err := json.NewDecoder(rr.Result().Body).Decode(v); err != nil {
-		t.Fatalf("decode dispatcher response: %v", err)
-	}
-}
-
 func TestUnknownMethod(t *testing.T) {
 	ts, cleanup := newGatewayWithAgent(t)
 	defer cleanup()
@@ -596,9 +191,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)
 	}
 }
 

gateway/a2a/client.go

@@ -60,38 +60,15 @@ func (c *Client) Card(ctx context.Context) (*AgentCard, error) {
 // If the agent returns a task that isn't yet terminal, Send polls
 // tasks/get until it completes or ctx is done.
 func (c *Client) Send(ctx context.Context, text string) (string, error) {
-	task, err := c.SendMessage(ctx, Message{
+	res, err := c.call(ctx, "message/send", sendParams{Message: Message{
 		Role:      "user",
 		Kind:      "message",
 		MessageID: uuid.New().String(),
 		Parts:     []Part{{Kind: "text", Text: text}},
-	})
+	}})
 	if err != nil {
 		return "", err
 	}
-	if task.Status.State != stateCompleted {
-		return "", fmt.Errorf("remote task %s ended in state %q", task.ID, task.Status.State)
-	}
-	return artifactsText(task.Artifacts), nil
-}
-
-// SendMessage sends an A2A message and returns the resulting terminal task.
-// To continue a multi-turn task, pass a Message with TaskID and ContextID set
-// to a prior task's id and context id.
-func (c *Client) SendMessage(ctx context.Context, message Message) (*Task, error) {
-	if message.MessageID == "" {
-		message.MessageID = uuid.New().String()
-	}
-	if message.Kind == "" {
-		message.Kind = "message"
-	}
-	if message.Role == "" {
-		message.Role = "user"
-	}
-	res, err := c.call(ctx, "message/send", sendParams{Message: message})
-	if err != nil {
-		return nil, err
-	}
 
 	// The result is a Message or a Task; the "kind" field disambiguates.
 	var probe struct {
@@ -102,61 +79,33 @@ func (c *Client) SendMessage(ctx context.Context, message Message) (*Task, error
 	if probe.Kind == "message" {
 		var m Message
 		if err := json.Unmarshal(res, &m); err != nil {
-			return nil, err
+			return "", err
 		}
-		return &Task{
-			ID:        m.TaskID,
-			ContextID: m.ContextID,
-			Kind:      "task",
-			Status:    TaskStatus{State: stateCompleted, Timestamp: time.Now().UTC().Format(time.RFC3339)},
-			Artifacts: []Artifact{textArtifact(textOf(m.Parts))},
-			History:   []Message{m},
-		}, nil
+		return textOf(m.Parts), nil
 	}
 
 	var task Task
 	if err := json.Unmarshal(res, &task); err != nil {
-		return nil, err
+		return "", err
 	}
 	for !terminal(task.Status.State) {
 		select {
 		case <-ctx.Done():
-			return nil, ctx.Err()
+			return "", ctx.Err()
 		case <-time.After(300 * time.Millisecond):
 		}
 		got, err := c.call(ctx, "tasks/get", getParams{ID: task.ID})
 		if err != nil {
-			return nil, err
+			return "", err
 		}
 		if err := json.Unmarshal(got, &task); err != nil {
-			return nil, err
+			return "", err
 		}
 	}
-	return &task, nil
-}
-
-// SetPushNotificationConfig asks the remote agent to POST updates for taskID to cfg.URL.
-func (c *Client) SetPushNotificationConfig(ctx context.Context, taskID string, cfg PushNotificationConfig) error {
-	_, err := c.call(ctx, "tasks/pushNotificationConfig/set", pushConfigParams{
-		ID:                     taskID,
-		PushNotificationConfig: cfg,
-	})
-	return err
-}
-
-// PushNotificationConfig returns the remote push notification config for taskID.
-func (c *Client) PushNotificationConfig(ctx context.Context, taskID string) (PushNotificationConfig, error) {
-	res, err := c.call(ctx, "tasks/pushNotificationConfig/get", getParams{ID: taskID})
-	if err != nil {
-		return PushNotificationConfig{}, err
+	if task.Status.State != stateCompleted {
+		return "", fmt.Errorf("remote task %s ended in state %q", task.ID, task.Status.State)
 	}
-	var out struct {
-		PushNotificationConfig PushNotificationConfig `json:"pushNotificationConfig"`
-	}
-	if err := json.Unmarshal(res, &out); err != nil {
-		return PushNotificationConfig{}, err
-	}
-	return out.PushNotificationConfig, nil
+	return artifactsText(task.Artifacts), nil
 }
 
 // call performs one JSON-RPC request and returns the raw result.

gateway/a2a/client_test.go

@@ -2,11 +2,8 @@ package a2a
 
 import (
 	"context"
-	"encoding/json"
-	"net/http"
 	"net/http/httptest"
 	"testing"
-	"time"
 )
 
 // An agent that embeds NewAgentHandler is directly A2A-queryable — no
@@ -51,62 +48,3 @@ func TestClientSendAndCard(t *testing.T) {
 		t.Errorf("Send reply = %q, want pong", reply)
 	}
 }
-
-func TestClientContinuesTaskAndConfiguresPush(t *testing.T) {
-	card := Card("solo", "http://localhost:4000", "", []string{"task"})
-	h := NewAgentHandler(card, func(_ context.Context, text string) (string, error) {
-		return "echo:" + text, nil
-	})
-	ts := httptest.NewServer(h)
-	defer ts.Close()
-
-	updates := make(chan Task, 1)
-	push := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
-		var task Task
-		if err := json.NewDecoder(r.Body).Decode(&task); err != nil {
-			t.Errorf("decode push task: %v", err)
-			return
-		}
-		updates <- task
-		w.WriteHeader(http.StatusAccepted)
-	}))
-	defer push.Close()
-
-	cl := NewClient(ts.URL)
-	first, err := cl.SendMessage(context.Background(), Message{
-		Parts: []Part{{Kind: "text", Text: "one"}},
-	})
-	if err != nil {
-		t.Fatalf("first SendMessage: %v", err)
-	}
-	second, err := cl.SendMessage(context.Background(), Message{
-		TaskID:    first.ID,
-		ContextID: first.ContextID,
-		Parts:     []Part{{Kind: "text", Text: "two"}},
-	})
-	if err != nil {
-		t.Fatalf("second SendMessage: %v", err)
-	}
-	if second.ID != first.ID || second.ContextID != first.ContextID || len(second.History) != 4 {
-		t.Fatalf("continued task = %+v, first %+v", second, first)
-	}
-	cfg := PushNotificationConfig{URL: push.URL}
-	if err := cl.SetPushNotificationConfig(context.Background(), second.ID, cfg); err != nil {
-		t.Fatalf("SetPushNotificationConfig: %v", err)
-	}
-	got, err := cl.PushNotificationConfig(context.Background(), second.ID)
-	if err != nil {
-		t.Fatalf("PushNotificationConfig: %v", err)
-	}
-	if got.URL != push.URL {
-		t.Fatalf("PushNotificationConfig URL = %q, want %q", got.URL, push.URL)
-	}
-	select {
-	case update := <-updates:
-		if update.ID != second.ID {
-			t.Fatalf("push update ID = %q, want %q", update.ID, second.ID)
-		}
-	case <-time.After(2 * time.Second):
-		t.Fatal("timed out waiting for push update")
-	}
-}

health/health.go

@@ -314,8 +314,7 @@ func DNSCheck(host string) CheckFunc {
 //
 // The lookup runs under the check's timeout, so an unreachable registry
 // (for example an etcd that has gone away) is reported as down rather
-// than blocking the probe. Registries that honor ListContext also receive
-// the check context directly.
+// than blocking the probe.
 func RegistryCheck(reg registry.Registry) CheckFunc {
 	return func(ctx context.Context) error {
 		if reg == nil {
@@ -323,7 +322,7 @@ func RegistryCheck(reg registry.Registry) CheckFunc {
 		}
 		errc := make(chan error, 1)
 		go func() {
-			_, err := reg.ListServices(registry.ListContext(ctx))
+			_, err := reg.ListServices()
 			errc <- err
 		}()
 		select {
@@ -338,54 +337,6 @@ func RegistryCheck(reg registry.Registry) CheckFunc {
 	}
 }
 
-// RegistryServiceCheck creates a check that verifies the local service
-// registration is visible in the registry. RegistryCheck only proves the
-// registry is reachable; this check also catches the common failure mode
-// where the process is alive and the registry answers, but this service's
-// node lease/record has disappeared and other services can no longer
-// discover it.
-//
-// Pass the service name and node ID used during registration. The default
-// RPC server node ID is service.Options().Name + "-" + service.Options().Id.
-func RegistryServiceCheck(reg registry.Registry, serviceName, nodeID string) CheckFunc {
-	return func(ctx context.Context) error {
-		if reg == nil {
-			return errors.New("no registry configured")
-		}
-		if serviceName == "" {
-			return errors.New("no service name configured")
-		}
-		if nodeID == "" {
-			return errors.New("no service node configured")
-		}
-
-		errc := make(chan error, 1)
-		go func() {
-			services, err := reg.GetService(serviceName, registry.GetContext(ctx))
-			if err != nil {
-				errc <- fmt.Errorf("registry %s lookup %s failed: %w", reg.String(), serviceName, err)
-				return
-			}
-			for _, service := range services {
-				for _, node := range service.Nodes {
-					if node.Id == nodeID {
-						errc <- nil
-						return
-					}
-				}
-			}
-			errc <- fmt.Errorf("registry %s missing node %s for service %s", reg.String(), nodeID, serviceName)
-		}()
-
-		select {
-		case err := <-errc:
-			return err
-		case <-ctx.Done():
-			return fmt.Errorf("registry %s service check timed out: %w", reg.String(), ctx.Err())
-		}
-	}
-}
-
 // CustomCheck creates a check from any function returning an error
 func CustomCheck(fn func() error) CheckFunc {
 	return func(ctx context.Context) error {

health/registry_check_test.go

@@ -89,60 +89,3 @@ func TestRegistryCheckMarksNotReady(t *testing.T) {
 		t.Error("service should be not-ready when the registry check is down")
 	}
 }
-
-func TestRegistryServiceCheckHealthy(t *testing.T) {
-	reg := registry.NewMemoryRegistry()
-	service := &registry.Service{
-		Name:  "orders",
-		Nodes: []*registry.Node{{Id: "orders-1"}},
-	}
-	if err := reg.Register(service); err != nil {
-		t.Fatalf("register service: %v", err)
-	}
-
-	check := RegistryServiceCheck(reg, "orders", "orders-1")
-	if err := check(context.Background()); err != nil {
-		t.Fatalf("registered service node should pass: %v", err)
-	}
-}
-
-func TestRegistryServiceCheckMissingNode(t *testing.T) {
-	reg := registry.NewMemoryRegistry()
-	service := &registry.Service{
-		Name:  "orders",
-		Nodes: []*registry.Node{{Id: "orders-1"}},
-	}
-	if err := reg.Register(service); err != nil {
-		t.Fatalf("register service: %v", err)
-	}
-
-	check := RegistryServiceCheck(reg, "orders", "orders-2")
-	err := check(context.Background())
-	if err == nil {
-		t.Fatal("missing service node should fail")
-	}
-	if !strings.Contains(err.Error(), "missing node orders-2") {
-		t.Errorf("error should describe the missing node: %v", err)
-	}
-}
-
-func TestRegistryServiceCheckMissingService(t *testing.T) {
-	check := RegistryServiceCheck(registry.NewMemoryRegistry(), "orders", "orders-1")
-	err := check(context.Background())
-	if err == nil {
-		t.Fatal("missing service should fail")
-	}
-	if !strings.Contains(err.Error(), registry.ErrNotFound.Error()) {
-		t.Errorf("error should include registry lookup failure: %v", err)
-	}
-}
-
-func TestRegistryServiceCheckMarksNotReady(t *testing.T) {
-	Reset()
-	defer Reset()
-
-	Register("registry-service", RegistryServiceCheck(registry.NewMemoryRegistry(), "orders", "orders-1"))
-	if IsReady(context.Background()) {
-		t.Error("service should be not-ready when its registry node is missing")
-	}
-}

internal/docs/AGENT_CONFORMANCE.md

@@ -1,39 +0,0 @@
-# Agent provider conformance matrix
-
-`go test ./...` includes `TestAgentProviderConformanceMatrix`, a shared agent
-scenario that runs against every registered chat provider. The scenario asks an
-agent to call a deterministic local tool, verifies the tool receives `ai.RunInfo`,
-and checks the final response carries the conformance marker. A fake provider path
-runs on every machine without network access so CI always exercises the harness.
-
-Live providers are opt-in to avoid flaky unauthenticated PR checks and accidental
-API spend. To run the live matrix, set `GO_MICRO_AGENT_CONFORMANCE_LIVE=1` plus the
-provider API keys you want to exercise:
-
-| Provider | Required API key | Optional model override |
-| --- | --- | --- |
-| OpenAI | `OPENAI_API_KEY` | `GO_MICRO_CONFORMANCE_OPENAI_MODEL` |
-| Anthropic | `ANTHROPIC_API_KEY` | `GO_MICRO_CONFORMANCE_ANTHROPIC_MODEL` |
-| Atlas Cloud | `ATLASCLOUD_API_KEY` | `GO_MICRO_CONFORMANCE_ATLASCLOUD_MODEL` |
-| Gemini | `GEMINI_API_KEY` | `GO_MICRO_CONFORMANCE_GEMINI_MODEL` |
-| Groq | `GROQ_API_KEY` | `GO_MICRO_CONFORMANCE_GROQ_MODEL` |
-| Mistral | `MISTRAL_API_KEY` | `GO_MICRO_CONFORMANCE_MISTRAL_MODEL` |
-| Together | `TOGETHER_API_KEY` | `GO_MICRO_CONFORMANCE_TOGETHER_MODEL` |
-
-When `GO_MICRO_AGENT_CONFORMANCE_LIVE` or a provider key is absent, the live
-provider subtest reports a deterministic skip. When both are present, a provider
-failure is a real test failure because drift in chat, tool calling, run metadata,
-or final-answer behavior means the services → agents lifecycle is no longer
-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/CONTINUOUS_IMPROVEMENT.md

@@ -133,26 +133,16 @@ output) but produce direction and coherence, not just code.
   factual-alignment and crispness fixes auto-merge like any increment;
   brand/positioning copy and blog drafts are *surfaced in a report* for the
   human, never auto-merged.
-- **Architect — continuous (hourly)** (`.github/workflows/architecture-review.yml`).
-  The *founder lens*, running alongside the builders. Each run it **tracks live
-  state** (what just merged, what's in flight), **prioritizes the roadmap**
-  (`ROADMAP.md`, Now → Next → Later) against an internal scan (lifecycle gaps, API
-  coherence and seams, dev-UX friction, missing pieces, drift/realignment), and
-  **maintains the ranked queue** in [`PRIORITIES.md`](PRIORITIES.md) — re-ranking
-  to reflect reality, backing each top item with a scoped issue, and posting an
-  assessment. It runs at `:59`, just before the `:29` increment, so it
-  re-prioritizes and *then* the loop builds the new top. **Its output is the
-  prioritized queue plus the assessment** — it does **not** make breaking or
-  architectural changes itself (those stay with the human). To avoid churn it only
-  opens a PR when the ranking actually changes.
+- **Architect — every few days** (`.github/workflows/architecture-review.yml`).
+  Reviews the framework/harness against the thesis: API coherence, lifecycle
+  gaps, drift/sprawl. **Its output is an assessment plus scoped follow-up
+  issues** that feed the hourly increment loop — it does **not** make breaking or
+  architectural changes itself (those stay with the human).
 
-The two loops are coupled through `PRIORITIES.md`: the **architect decides *what***
-(roadmap + internal priorities, ranked, issue-linked) and the **hourly increment
-loop builds the top open item** — falling back to its own judgment only if the
-queue is empty. DevRel keeps the public story honest alongside. So work is
-roadmap-driven by default, not a fresh guess every hour. Cadence is tunable in each
-workflow's `cron`; the human can reorder `PRIORITIES.md` or its issues at any time
-to redirect. Codex is serial, so these passes queue behind any in-flight increment.
+Together they close the loop: the architect decides *what* should change and files
+issues, the increment loop *builds* them, and DevRel keeps the public story
+honest. Cadence is tunable in each workflow's `cron`. Codex is serial, so these
+passes queue behind any in-flight increment rather than running concurrently.
 
 ## Stop / redirect
 

internal/docs/PRIORITIES.md

@@ -1,29 +0,0 @@
-# Priorities
-
-The ranked work queue for the autonomous improvement loop. The
-**architecture-review** pass (the *architect*) owns this file: each run it turns
-the [roadmap](../../ROADMAP.md) plus an internal scan (gaps in the
-services → agents → workflows lifecycle, API coherence, drift, tech debt, test and
-DX friction) into a single ordered list — highest-value first — and links each
-item to a tracking issue. The hourly **continuous-improvement** pass works the
-**top item whose issue is still open**. So the architect decides *what*, and the
-increment loop *builds* it.
-
-**Reading / editing.** An item is done when its linked issue closes (the increment
-that builds it adds `Closes #<issue>`). Roadmap phase (Now → Next → Later) is the
-primary ordering; internal findings are interleaved by value, not kept in a
-separate list. The human can reorder this list — or the issues — at any time to
-redirect the loop; direction always wins.
-
-**Off-limits to the loop** (the architect proposes these as notes, never as queue
-items the loop can auto-merge): brand/positioning copy, breaking public-API
-changes, architectural rewrites. Those go to the human.
-
-## Developer experience (ranked)
-
-1. **Schedule cross-provider agent conformance** ([#3295](https://github.com/micro/go-micro/issues/3295)) — after the 0→hero reference and provider-focused unit coverage shipped, the highest-value remaining Now-roadmap hardening gap is a scheduled, key-gated provider matrix that proves the same agent/tool workflow keeps working across supported models without blocking contributors who lack secrets.
-2. **Harden agent failure and cancellation semantics** ([#3296](https://github.com/micro/go-micro/issues/3296)) — the harness is increasingly durable, streaming, observable, and human-in-the-loop; the next operability seam is making timeouts, cancellation, rate-limit errors, and retry/backoff behavior predictable across agent, AI provider, service-tool, and flow boundaries.
-3. **Expose run inspection in the CLI inner loop** ([#3297](https://github.com/micro/go-micro/issues/3297)) — the canon promises scaffold → run → chat → inspect → deploy, and recent work has strengthened scaffold/run/chat/deploy; inspection remains the most visible DX gap for turning agent/flow activity into actionable breadcrumbs during local development.
-
-_Seeded by Claude Code from the roadmap + open issues; thereafter maintained by the
-architecture-review pass._

internal/docs/THESIS.md

@@ -4,39 +4,6 @@ This is the North Star for the project and for the autonomous improvement loop
 (see `CONTINUOUS_IMPROVEMENT.md`). Every change should move toward it; work that
 doesn't isn't an improvement, however clean.
 
-## Mission — the problem we solve
-
-Go Micro started in 2015 because building distributed systems in Go was too hard:
-too much boilerplate, too many decisions before a single endpoint runs. The
-mission was to **make building distributed systems simple** — sane defaults,
-pluggable, out of the developer's way.
-
-Agents are distributed systems too. The moment an agent discovers services, calls
-them, holds state, and recovers from failure, it *is* a distributed system — the
-exact problem Go Micro already solved for services. So the mission hasn't
-changed, only extended:
-
-> **Make building agentic, distributed software in Go simple — make building an
-> agent as easy as building a service, on one runtime, because an agent is a
-> distributed system.**
-
-That is the problem we solve, and it is the question every priority is judged
-against: *does this make the services → agents → workflows lifecycle simpler, more
-cohesive, and more operable — or is it scope that doesn't serve that?* It is
-evolution, not a pivot: the decade of services work is the foundation, and the
-agent layer is that foundation leveraged for the AI era.
-
-## The canon
-
-The vision isn't only in this file. The years of focus and context live in the
-**corpus** — the [blog](../website/blog/) (the actual thinking, e.g. `/blog/14`
-"Going All In on AI" and `/blog/27` "Back from the Dead"), the
-[`README`](../../README.md), and the [website](../website/). Those are the canon;
-this North Star is their **distillation** and must stay faithful to them. When the
-two diverge, that's a signal — either the work has drifted from the mission, or the
-North Star has drifted from the lived story and needs re-grounding in the corpus.
-The architect re-derives alignment from the canon, not from this file alone.
-
 ## Thesis
 
 Go Micro is an **agent harness and service framework** — one runtime that, holistically,

internal/harness/plan-delegate/main.go

@@ -225,19 +225,22 @@ func providerKey(provider string) string {
 	return os.Getenv(env)
 }
 
-func runPlanDelegate(provider string) error {
+func main() {
+	provider := flag.String("provider", "mock", "LLM provider: mock (default), anthropic, openai, gemini, groq, mistral, together, atlascloud")
+	flag.Parse()
+
 	apiKey := ""
-	if provider == "mock" {
+	if *provider == "mock" {
 		ai.Register("mock", newMock)
 	} else {
-		apiKey = providerKey(provider)
+		apiKey = providerKey(*provider)
 		if apiKey == "" {
-			fmt.Printf("no API key for provider %q — set MICRO_AI_API_KEY or the provider's key env\n", provider)
-			return nil
+			fmt.Printf("no API key for provider %q — set MICRO_AI_API_KEY or the provider's key env\n", *provider)
+			return
 		}
 	}
 
-	fmt.Printf("\n\033[1mPlan & Delegate — live integration harness (provider: %s)\033[0m\n", provider)
+	fmt.Printf("\n\033[1mPlan & Delegate — live integration harness (provider: %s)\033[0m\n", *provider)
 	fmt.Print("Real services, registry, RPC, agent loop, store, delegation.\n\n")
 
 	reg := registry.NewMemoryRegistry()
@@ -248,14 +251,16 @@ func runPlanDelegate(provider string) error {
 	taskSvc := new(TaskService)
 	task := service.New(service.Name("task"), service.Address("127.0.0.1:0"), service.Registry(reg), service.Client(cl))
 	if err := task.Handle(taskSvc); err != nil {
-		return fmt.Errorf("task handle: %w", err)
+		fmt.Println("task handle:", err)
+		os.Exit(1)
 	}
 	go task.Run()
 
 	notifySvc := new(NotifyService)
 	notify := service.New(service.Name("notify"), service.Address("127.0.0.1:0"), service.Registry(reg), service.Client(cl))
 	if err := notify.Handle(notifySvc); err != nil {
-		return fmt.Errorf("notify handle: %w", err)
+		fmt.Println("notify handle:", err)
+		os.Exit(1)
 	}
 	go notify.Run()
 
@@ -265,7 +270,7 @@ func runPlanDelegate(provider string) error {
 		agent.Address("127.0.0.1:0"),
 		agent.Services("notify"),
 		agent.Prompt("You handle outbound notifications. Use the notify service."),
-		agent.Provider(provider), agent.APIKey(apiKey),
+		agent.Provider(*provider), agent.APIKey(apiKey),
 		agent.WithRegistry(reg), agent.WithClient(cl), agent.WithStore(mem),
 	)
 	go comms.Run()
@@ -277,27 +282,24 @@ func runPlanDelegate(provider string) error {
 		agent.Address("127.0.0.1:0"),
 		agent.Services("task"),
 		agent.Prompt("You coordinate launch work. Plan first, create tasks, and delegate notifications to the \"comms\" agent."),
-		agent.Provider(provider), agent.APIKey(apiKey),
+		agent.Provider(*provider), agent.APIKey(apiKey),
 		agent.WithRegistry(reg), agent.WithClient(cl), agent.WithStore(mem),
 	)
 	go conductor.Run()
 	defer conductor.Stop()
 
 	fmt.Println("waiting for services + agents to register...")
-	waitForService := func(name string) error {
+	waitForService := func(name string) {
 		deadline := time.Now().Add(5 * time.Second)
 		for time.Now().Before(deadline) {
 			if svcs, err := reg.GetService(name); err == nil && len(svcs) > 0 && len(svcs[0].Nodes) > 0 {
-				return nil
+				return
 			}
 			time.Sleep(20 * time.Millisecond)
 		}
-		return fmt.Errorf("service %q never registered", name)
 	}
 	for _, name := range []string{"task", "notify", "comms", "conductor"} {
-		if err := waitForService(name); err != nil {
-			return err
-		}
+		waitForService(name)
 	}
 
 	f := flow.New("zero-to-hero",
@@ -305,12 +307,14 @@ func runPlanDelegate(provider string) error {
 		flow.Prompt("Create three launch tasks (Design, Build, Ship), then make sure owner@acme.com is notified: {{.Data}}"),
 	)
 	if err := f.Register(reg, broker.DefaultBroker, cl); err != nil {
-		return fmt.Errorf("flow register: %w", err)
+		fmt.Println("flow register:", err)
+		os.Exit(1)
 	}
 
 	fmt.Print("\n\033[1m> flow:\033[0m services + agents + workflow + plan/delegate, no API key.\n\n")
 	if err := f.Execute(context.Background(), "launch readiness"); err != nil {
-		return fmt.Errorf("flow execute: %w", err)
+		fmt.Println("\033[31merror:\033[0m", err)
+		os.Exit(1)
 	}
 
 	if rs := f.Results(); len(rs) > 0 {
@@ -321,22 +325,13 @@ func runPlanDelegate(provider string) error {
 	if recs, _ := store.Scope(mem, "agent", "conductor").Read("plan"); len(recs) > 0 {
 		fmt.Printf("\n\033[1mstored plan (agent/conductor/plan):\033[0m %s\n", string(recs[0].Value))
 	} else {
-		return fmt.Errorf("plan was not persisted")
+		fmt.Println("\n\033[31m! plan was not persisted\033[0m")
+		os.Exit(1)
 	}
 	if taskSvc.count() != 3 || notifySvc.count() != 1 {
-		return fmt.Errorf("unexpected side effects: tasks=%d notify=%d", taskSvc.count(), notifySvc.count())
+		fmt.Printf("\n\033[31m! unexpected side effects: tasks=%d notify=%d\033[0m\n", taskSvc.count(), notifySvc.count())
+		os.Exit(1)
 	}
 
 	fmt.Println("\n\033[32m✓ 0→hero flow complete (services → agents → workflow)\033[0m")
-	return nil
-}
-
-func main() {
-	provider := flag.String("provider", "mock", "LLM provider: mock (default), anthropic, openai, gemini, groq, mistral, together, atlascloud")
-	flag.Parse()
-
-	if err := runPlanDelegate(*provider); err != nil {
-		fmt.Println("\033[31merror:\033[0m", err)
-		os.Exit(1)
-	}
 }

internal/harness/plan-delegate/main_test.go

@@ -206,16 +206,3 @@ func TestFlowDispatchesToAgentEndToEnd(t *testing.T) {
 		t.Errorf("notify called %d times, want 1 (flow->agent->delegate->comms chain broken)", n)
 	}
 }
-
-// TestZeroToHeroContract locks the roadmap's second golden path into the
-// ordinary Go test contract. It runs the same executable harness used by
-// `make harness`: services + agents + flow + plan/delegate, with only the
-// LLM replaced by the deterministic mock provider.
-func TestZeroToHeroContract(t *testing.T) {
-	if testing.Short() {
-		t.Skip("0→hero harness boots an end-to-end system; skipped with -short")
-	}
-	if err := runPlanDelegate("mock"); err != nil {
-		t.Fatalf("0→hero harness: %v", err)
-	}
-}

internal/harness/provider-conformance/README.md

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

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

@@ -1,16 +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 three boundaries together:
-
-1. **Run** — `micro run` remains available as the local development entry point.
-2. **Chat** — `micro chat` remains available as the interactive agent entry point.
-3. **Inspect** — `micro flow runs` remains available for durable workflow run
-   history inspection.
-
-After the CLI boundary smoke checks, the script runs the deterministic harnesses
-that boot real services, agents, workflows, store-backed run history, and A2A
-with only the LLM mocked.

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

@@ -1,13 +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
-
-# 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/SECURITY_MIGRATION.md

@@ -2,64 +2,58 @@
 
 ## Overview
 
-Go Micro v6 verifies TLS certificates by default. This guide is for teams
-upgrading from v5, where TLS verification was disabled by default for backward
-compatibility.
+This document provides guidance for migrating to secure TLS certificate verification in go-micro v5.
 
-## Current Status (v6)
+## Current Status (v5)
 
-**Default Behavior**: TLS certificate verification is **enabled** by default
-(`InsecureSkipVerify: false`).
+**Default Behavior**: TLS certificate verification is **disabled** by default (`InsecureSkipVerify: true`)
 
-**What changed from v5**: v5 allowed `MICRO_TLS_SECURE=true` to opt into
-certificate verification. In v6, secure verification is the default and
-`MICRO_TLS_SECURE` is no longer used.
+**Reason**: Backward compatibility with existing deployments to avoid breaking production systems during routine upgrades.
 
-**Development escape hatch**: for local self-signed certificates only, set
-`MICRO_TLS_INSECURE=true` or provide an explicit insecure TLS config.
+**Security Risk**: The default behavior is vulnerable to man-in-the-middle (MITM) attacks.
 
-## Migration Path from v5
+## Migration Path
 
-### 1. Remove the old opt-in flag
+### Option 1: Enable Secure Mode (RECOMMENDED)
 
-Delete any use of the v5-only environment variable:
+Set the environment variable to enable certificate verification:
 
 ```bash
-unset MICRO_TLS_SECURE
+export MICRO_TLS_SECURE=true
 ```
 
-No replacement is required for production: verification is already on in v6.
+This enables proper TLS certificate verification while maintaining compatibility with v5.
 
-### 2. Use the default secure config
+### Option 2: Use SecureConfig Directly
 
-Most services need no TLS-specific code. If you configure TLS explicitly, use a standard `crypto/tls` config with verification enabled:
+In your code, explicitly use the secure configuration:
 
 ```go
 import (
-    "crypto/tls"
     "go-micro.dev/v6/broker"
+    mls "go-micro.dev/v6/util/tls"
 )
 
-// Create broker with certificate verification enabled.
+// Create broker with secure TLS config
 b := broker.NewHttpBroker(
-    broker.TLSConfig(&tls.Config{MinVersion: tls.VersionTLS12}),
+    broker.TLSConfig(mls.SecureConfig()),
 )
 ```
 
-### 3. Provide a custom trust root when needed
+### Option 3: Provide Custom TLS Configuration
 
-For private CAs, provide your own TLS configuration:
+For fine-grained control, provide your own TLS configuration:
 
 ```go
 import (
     "crypto/tls"
     "crypto/x509"
     "go-micro.dev/v6/broker"
-    "os"
+    "io/ioutil"
 )
 
 // Load CA certificates
-caCert, err := os.ReadFile("/path/to/ca-cert.pem")
+caCert, err := ioutil.ReadFile("/path/to/ca-cert.pem")
 if err != nil {
     log.Fatal(err)
 }
@@ -79,61 +73,52 @@ b := broker.NewHttpBroker(
 )
 ```
 
-### 4. Use insecure mode only for local development
-
-If a development environment still uses self-signed certificates that are not in
-your trust store, opt out explicitly:
-
-```bash
-export MICRO_TLS_INSECURE=true
-```
-
-or in code:
-
-```go
-broker.TLSConfig(&tls.Config{InsecureSkipVerify: true, MinVersion: tls.VersionTLS12})
-```
-
-Do not use insecure mode in production.
-
 ## Production Deployment Strategy
 
 ### Rolling Upgrade Considerations
 
-The default changed at the v6 major-version boundary. Before rolling v6 into a
-fleet that uses TLS, verify that:
+The current implementation maintains backward compatibility, allowing safe rolling upgrades:
 
-1. All services present certificates trusted by their peers.
-2. Private or self-signed CAs are installed consistently on every host.
-3. Certificates include the DNS names or IP subject alternative names used by
-   clients.
-4. Any deliberate development-only insecure settings are excluded from
-   production manifests.
+1. **Mixed Version Deployments**: v5 instances can communicate regardless of TLS security settings
+2. **No Immediate Breaking Changes**: Systems continue working with existing behavior
+3. **Gradual Migration**: Enable security incrementally across your infrastructure
 
 ### Recommended Approach
 
-1. **Test in Staging** with the same certificate chain and service names used in
-   production.
-2. **Remove v5 flags** such as `MICRO_TLS_SECURE`; they no longer control v6.
-3. **Monitor for Issues**: watch for TLS handshake failures or certificate
-   validation errors.
-4. **Use explicit insecure mode only in dev** when a short-lived environment
-   cannot yet provide trusted certificates.
+1. **Test in Staging**:
+   ```bash
+   # In staging environment
+   export MICRO_TLS_SECURE=true
+   ```
+   
+2. **Deploy with Feature Flag**: Use environment-based configuration for gradual rollout
+
+3. **Monitor for Issues**: Watch for TLS handshake failures or certificate validation errors
+
+4. **Full Production Rollout**: Once validated, enable across all services
 
 ### Multi-Host/Multi-Process Considerations
 
-**Certificate Trust**: With secure mode as the default, ensure:
+**Certificate Trust**: When enabling secure mode, ensure:
 
-1. All hosts trust the same root CAs.
-2. Self-signed certificates are properly distributed if used.
-3. Certificate validity periods are monitored.
-4. Certificate chains are complete.
+1. All hosts trust the same root CAs
+2. Self-signed certificates are properly distributed if used
+3. Certificate validity periods are monitored
+4. Certificate chains are complete
 
 **Service Mesh Alternative**: Consider using a service mesh (Istio, Linkerd, etc.) for:
 - Automatic mTLS between services
 - Certificate management and rotation
 - No application code changes required
 
+## Future Changes (v6)
+
+In go-micro v6, the default will change to **secure by default**:
+
+- `InsecureSkipVerify: false` (certificate verification enabled)
+- Breaking change requiring major version bump
+- Migration completed before v6 release avoids disruption
+
 ## Testing Your Migration
 
 ### Verify Secure Mode is Active
@@ -142,12 +127,14 @@ fleet that uses TLS, verify that:
 package main
 
 import (
-    "crypto/tls"
     "fmt"
+    mls "go-micro.dev/v6/util/tls"
+    "os"
 )
 
 func main() {
-    config := &tls.Config{MinVersion: tls.VersionTLS12}
+    os.Setenv("MICRO_TLS_SECURE", "true")
+    config := mls.Config()
     fmt.Printf("InsecureSkipVerify: %v (should be false)\n", config.InsecureSkipVerify)
 }
 ```
@@ -168,7 +155,7 @@ Create a test service and verify it:
 **Solution**:
 1. Add the CA certificate to the trusted root CAs
 2. Use a properly signed certificate
-3. For development only: use `MICRO_TLS_INSECURE=true` or an explicit insecure TLS config
+3. For development only: Use `InsecureConfig()` explicitly
 
 ### Issue: "x509: certificate has expired"
 
@@ -179,17 +166,24 @@ Create a test service and verify it:
 2. Implement certificate rotation
 3. Monitor certificate expiry dates
 
-### Issue: Services can't communicate after upgrading to v6
+### Issue: Services can't communicate after enabling secure mode
 
-**Cause**: Certificates that v5 accepted by default are now verified.
+**Cause**: Mixed certificate authorities or missing certificates
 
 **Solution**:
-1. Ensure all services use certificates from a trusted CA
+1. Ensure all services use certificates from the same CA
 2. Distribute CA certificates to all nodes
 3. Verify certificate SANs match service addresses
-4. Use insecure mode only as a temporary local-development workaround
 
 ## Questions?
 
-For issues or questions about TLS security migration, open an issue on GitHub or
-check the documentation at https://go-micro.dev/docs/.
+For issues or questions about TLS security migration, please:
+- Open an issue on GitHub
+- Check the documentation at https://go-micro.dev/docs/
+- Review the security guidelines
+
+## Security Resources
+
+- [OWASP TLS Cheat Sheet](https://cheatsheetseries.owasp.org/cheatsheets/Transport_Layer_Protection_Cheat_Sheet.html)
+- [Go TLS Documentation](https://pkg.go.dev/crypto/tls)
+- [Certificate Best Practices](https://www.ssl.com/guide/ssl-best-practices/)

internal/website/docs/TLS_SECURITY_UPDATE.md

@@ -2,36 +2,66 @@
 
 ## What Changed
 
-Go Micro v6 verifies TLS certificates by default. This completes the v5 security
-migration where verification was opt-in.
+The TLS configuration in go-micro now includes a security deprecation warning.
 
-## Current Behavior (v6.x)
+## Current Behavior (v5.x)
 
-**Default**: TLS certificate verification is **enabled**.
-- `MICRO_TLS_SECURE` was a v5 opt-in flag and is no longer used.
-- For local development with untrusted self-signed certificates, opt out
-  explicitly with `MICRO_TLS_INSECURE=true` or an explicit insecure TLS config.
+**Default**: TLS certificate verification is **disabled** for backward compatibility
+- This maintains existing behavior to avoid breaking production deployments
+- A deprecation warning is logged once per process startup
 
-## Production Recommendation
+**Why**: Changing the default to secure would be a **breaking change** that could disrupt:
+- Production systems during routine upgrades
+- Distributed systems with mixed versions
+- Services using self-signed certificates
 
-For production deployments:
-1. Use CA-signed certificates or distribute your private CA to every host.
-2. Remove old `MICRO_TLS_SECURE` settings from v5-era manifests.
-3. Do not set `MICRO_TLS_INSECURE=true` in production.
-4. Consider service mesh mTLS (Istio, Linkerd) if certificate lifecycle should be
-   managed outside the application.
+## How to Enable Security (Recommended)
+
+### Option 1: Environment Variable
+
+```bash
+export MICRO_TLS_SECURE=true
+```
+
+### Option 2: Use SecureConfig
+
+```go
+import (
+    "go-micro.dev/v6/broker"
+    mls "go-micro.dev/v6/util/tls"
+)
+
+broker := broker.NewHttpBroker(
+    broker.TLSConfig(mls.SecureConfig()),
+)
+```
 
 ## Migration Timeline
 
-- **v5.x**: Insecure by default, opt-in security via `MICRO_TLS_SECURE=true`.
-- **v6.x current**: Secure by default; use `MICRO_TLS_INSECURE=true` only for an
-  explicit development opt-out.
+- **v5.x (Current)**: Insecure by default, opt-in security via `MICRO_TLS_SECURE=true`
+- **v6.x (Future)**: Secure by default (breaking change with major version bump)
+
+## Why This Approach?
+
+This addresses the concerns raised about:
+
+1. **Major version requirements**: No breaking change in v5, deferred to v6
+2. **Cross-host compatibility**: All hosts use same default behavior
+3. **Production safety**: Existing deployments continue working during upgrades
+4. **Migration path**: Clear opt-in path with documentation
 
 ## Documentation
 
-See [SECURITY_MIGRATION.md](SECURITY_MIGRATION.html) for the detailed migration
-guide.
+See [SECURITY_MIGRATION.md](SECURITY_MIGRATION.html) for detailed migration guide.
+
+## Security Recommendation
+
+For production deployments:
+1. Test with `MICRO_TLS_SECURE=true` in staging
+2. Use proper CA-signed certificates
+3. Consider service mesh (Istio, Linkerd) for automatic mTLS
+4. Plan migration before v6 release
 
 ## Questions?
 
-Open an issue on GitHub or check the documentation at https://go-micro.dev/docs/.
+Open an issue on GitHub or check the documentation at https://go-micro.dev/docs/

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

@@ -66,7 +66,7 @@ A card looks like:
   "url": "https://agents.example.com/agents/task-mgr",
   "version": "1.0.0",
   "protocolVersion": "0.3.0",
-  "capabilities": { "streaming": true, "pushNotifications": true },
+  "capabilities": { "streaming": false, "pushNotifications": false },
   "defaultInputModes": ["text/plain"],
   "defaultOutputModes": ["text/plain"],
   "skills": [{ "id": "chat", "name": "Chat", "tags": ["task", "project"] }]
@@ -100,40 +100,7 @@ curl -s https://agents.example.com/agents/task-mgr \
 }
 ```
 
-Retrieve a task later with `tasks/get` (`params: { "id": "…" }`). To continue
-the same piece of work, send another `message/send` with the previous `taskId`
-and `contextId`. The gateway preserves the task id, context id, and prior
-history, then appends the new user turn and agent reply. That makes a remote
-A2A task fit the Go Micro lifecycle: services are still invoked through the
-agent's normal tools, the agent keeps task context across turns, and a workflow
-can poll one task id as the conversation progresses.
-
-## Push notifications
-
-Operators can register a task callback with
-`tasks/pushNotificationConfig/set`:
-
-```bash
-curl -s https://agents.example.com/agents/task-mgr \
-  -H 'content-type: application/json' \
-  -d '{
-    "jsonrpc": "2.0", "id": 2,
-    "method": "tasks/pushNotificationConfig/set",
-    "params": {
-      "id": "task-id",
-      "pushNotificationConfig": {
-        "url": "https://workflow.example.com/a2a/tasks",
-        "token": "optional-bearer-token"
-      }
-    }
-  }'
-```
-
-The gateway stores one callback per retained task and POSTs the latest task
-snapshot to that URL whenever the task changes. Delivery is best effort: failures
-do not fail the agent turn, and there is no retry queue in the in-memory gateway.
-Use `tasks/get` as the source of truth after a missed callback or receiver
-outage. If a token is configured, it is sent as `Authorization: Bearer <token>`.
+Retrieve a task later with `tasks/get` (`params: { "id": "…" }`).
 
 ## Calling out to other agents
 
@@ -165,18 +132,22 @@ yet terminal, it polls `tasks/get` until it completes.
 
 ## Scope
 
-This is the JSON-RPC binding for task execution:
+This is the JSON-RPC binding for completed-task execution:
 
 - **`message/send`** runs the agent and returns a completed `Task`.
 - **`message/stream`** streams the completed `Task` as an SSE `data:` event, giving A2A clients a streaming-compatible path while the underlying agent call remains synchronous.
 - **`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.
+- Push notifications.
+
+These are the natural follow-ups; the completed-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

@@ -37,44 +37,12 @@ your stack — the harness *is* the stack.
 | 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 (flows already do) | In progress |
-| Observability | `RunInfo` → OpenTelemetry spans for runs, model calls, tools, delegation, and failures; persisted run history | Shipped |
+| Observability | `RunInfo` → OpenTelemetry spans; run history on the CLI | In progress |
 | 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.
 
-## Observing agent runs
-
-Pass an OpenTelemetry tracer provider when you construct an agent to turn the
-agent's `RunInfo` into spans:
-
-```go
-agent := micro.NewAgent("conductor",
-    micro.AgentProvider("anthropic"),
-    micro.AgentTraceProvider(otel.GetTracerProvider()),
-)
-```
-
-A traced `Ask` emits a parent `agent.run` span plus child spans for
-`agent.model.call` and `agent.tool.call`. Delegate tool calls are marked with
-`agent.delegate=true`; ephemeral sub-agents start their own `agent.run` span with
-`agent.run.parent_id` set to the delegating run, so a trace shows the hand-off
-from service-like agent to sub-agent. Failure and refusal outcomes set error
-status on the relevant span and are also recorded in the persisted run timeline.
-
-Important span attributes include:
-
-| Attribute | Meaning |
-|---|---|
-| `agent.run.id` | Stable run correlation ID surfaced as `ai.RunInfo.RunID` |
-| `agent.run.parent_id` | Parent run for delegated sub-agent work |
-| `agent.name` | Agent that owns the run or call |
-| `agent.model.provider` / `agent.model.name` | Provider and configured model for model calls |
-| `agent.tool.name` | Tool invoked by the model |
-| `agent.delegate` | Whether the tool call is a delegation boundary |
-| `agent.latency_ms` | Elapsed time for the run/call |
-| `agent.tokens.*` | Token usage when the provider reports it |
-
 ## Why services are the right substrate
 
 An agent that does real work needs typed, discoverable, callable capabilities —

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

@@ -43,7 +43,7 @@ The built-in providers currently register these capability interfaces:
 | `gemini` | Yes | No | No | No |
 | `groq` | Yes | No | No | No |
 | `mistral` | Yes | No | No | No |
-| `openai` | Yes | Yes | No | Yes |
+| `openai` | Yes | Yes | No | No |
 | `together` | Yes | No | No | No |
 
 ## Step 1: Implement the `ai.Model` Interface

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

@@ -100,22 +100,7 @@ c := &x402.Client{
 resp, err := c.Do(req) // a 402 is paid and retried; over-budget calls error instead
 ```
 
-`Payer` is an interface (`Pay(ctx, Requirements) (payment string, error)`) — the consumer counterpart to `Facilitator`. The budget accumulates across calls, so a long-running agent can be handed a fixed allowance for a task. Budget is reserved before payment is created, which means parallel paid calls cannot race past the cap; if payment creation or verification fails, the reservation is released. (The agent-level `AgentMaxSpend` option, wiring this into the agent loop next to `MaxSteps`/`ApproveTool`, is the next step.)
-
-### Live facilitator conformance
-
-The regular test suite uses in-process facilitators and does not need network credentials. To smoke-test a hosted facilitator, run the opt-in live conformance test with a real payment payload and matching requirements:
-
-```sh
-GO_MICRO_X402_LIVE_FACILITATOR_URL=https://facilitator.example \
-GO_MICRO_X402_LIVE_PAYMENT='...' \
-GO_MICRO_X402_LIVE_PAY_TO=0xYourAddress \
-GO_MICRO_X402_LIVE_NETWORK=base \
-GO_MICRO_X402_LIVE_AMOUNT=1 \
-go test ./wrapper/x402 -run TestLiveFacilitatorConformance -count=1
-```
-
-Leave those variables unset in normal CI; the live test skips unless the facilitator URL, payment payload, and pay-to address are all provided.
+`Payer` is an interface (`Pay(ctx, Requirements) (payment string, error)`) — the consumer counterpart to `Facilitator`. The budget accumulates across calls, so a long-running agent can be handed a fixed allowance for a task. (The agent-level `AgentMaxSpend` option, wiring this into the agent loop next to `MaxSteps`/`ApproveTool`, is the next step.)
 
 ## Notes
 

internal/website/docs/roadmap.md

@@ -13,7 +13,7 @@ The foundation is in place:
 - **Services** — register, discover, RPC, events; every endpoint is automatically an MCP tool.
 - **Agents** — a model with memory and tools that manages services, with `plan`, `delegate`, and guardrails (`MaxSteps`, `LoopLimit`, `ApproveTool`) built in, plus tool-execution middleware (`WrapTool`) and run metadata.
 - **Flows** — durable, event-driven workflows: ordered steps that checkpoint and resume after a crash.
-- **Interop** — the MCP gateway (services as tools) and the A2A gateway (agents as agents, both directions, including A2A streaming, push notifications, and multi-turn continuation), both generated from the registry; x402 for paid tools.
+- **Interop** — the MCP gateway (services as tools) and the A2A gateway (agents as agents, both directions), both generated from the registry; x402 for paid tools.
 - **Secure by default** — TLS verification on, state scoped per component.
 
 ## Principles
@@ -37,14 +37,15 @@ 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.
+- **Streaming.** `ai.Stream` is stubbed across providers; real chat and long-task UX need it, end to end through A2A `message/stream`.
 - **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** — pause and resume mid-run (`input-required`), beyond the binary `ApproveTool` gate.
-- **A2A** — richer live-stream reconnection (`tasks/resubscribe`) and `input-required` handoffs.
+- **x402** — conformance against a live facilitator; paid remote tools as agent tools with spend caps.
+- **A2A** — streaming, push notifications, multi-turn tasks.
 
 ## Developer experience (ongoing)
 

micro.go

@@ -110,36 +110,18 @@ func AgentApproveTool(fn ApproveFunc) AgentOption { return agent.ApproveTool(fn)
 // Memory is an agent's pluggable conversation memory.
 type Memory = agent.Memory
 
-// MemoryRecall is implemented by memory backends that retrieve prior context.
-type MemoryRecall = agent.MemoryRecall
-
 // ToolFunc handles a custom agent tool call.
 type ToolFunc = agent.ToolFunc
 
 // NewMemory returns the default store-backed agent memory.
 func NewMemory(s store.Store, key string, limit int) Memory { return agent.NewMemory(s, key, limit) }
 
-// NewCompactingMemory returns store-backed memory with deterministic
-// summarization and retrieval controls.
-func NewCompactingMemory(s store.Store, key string, maxMessages, keepRecent int) Memory {
-	return agent.NewCompactingMemory(s, key, maxMessages, keepRecent)
-}
-
 // NewInMemory returns non-persistent agent memory.
 func NewInMemory(limit int) Memory { return agent.NewInMemory(limit) }
 
 // AgentMemory sets the agent's conversation memory (default: store-backed).
 func AgentMemory(m Memory) AgentOption { return agent.WithMemory(m) }
 
-// AgentCompactMemory enables deterministic default-memory compaction and
-// retrieval for long-running agents.
-func AgentCompactMemory(maxMessages, keepRecent int) AgentOption {
-	return agent.CompactMemory(maxMessages, keepRecent)
-}
-
-// AgentMemoryRecallLimit bounds recalled archived turns injected per Ask.
-func AgentMemoryRecallLimit(n int) AgentOption { return agent.MemoryRecallLimit(n) }
-
 // AgentTool registers a custom tool the agent can call, beyond its services.
 func AgentTool(name, description string, properties map[string]any, handler ToolFunc) AgentOption {
 	return agent.WithTool(name, description, properties, handler)
@@ -160,10 +142,6 @@ func AgentWrapTool(w ...ai.ToolWrapper) AgentOption {
 	return agent.WrapTool(w...)
 }
 
-// AgentTraceProvider enables OpenTelemetry spans for agent runs, model calls,
-// tool calls, delegation, and failures.
-func AgentTraceProvider(tp trace.TracerProvider) AgentOption { return agent.TraceProvider(tp) }
-
 // NewFlow creates an event-driven LLM orchestration unit.
 //
 //	f := micro.NewFlow("onboard-user",

wrapper/x402/client.go

@@ -83,9 +83,7 @@ func (c *Client) Do(req *http.Request) (*http.Response, error) {
 	reqd := ch.Accepts[0]
 	amount, _ := strconv.ParseInt(reqd.MaxAmountRequired, 10, 64)
 
-	// Spend cap: reserve before paying so concurrent calls cannot all pass
-	// the check and overspend the caller's allowance. Roll the reservation
-	// back if payment construction, replay, or verification fails.
+	// Spend cap: refuse before paying if this would exceed the budget.
 	c.mu.Lock()
 	if c.Budget > 0 && c.spent+amount > c.Budget {
 		spent := c.spent
@@ -93,26 +91,13 @@ func (c *Client) Do(req *http.Request) (*http.Response, error) {
 		return nil, fmt.Errorf("x402: paying %d for %s would exceed budget (spent %d of %d)",
 			amount, reqd.Resource, spent, c.Budget)
 	}
-	c.spent += amount
 	c.mu.Unlock()
-	reserved := true
-	rollback := func() {
-		if !reserved {
-			return
-		}
-		c.mu.Lock()
-		c.spent -= amount
-		c.mu.Unlock()
-		reserved = false
-	}
 
 	if c.Payer == nil {
-		rollback()
 		return nil, fmt.Errorf("x402: payment required for %s but no Payer configured", reqd.Resource)
 	}
 	payment, err := c.Payer.Pay(req.Context(), reqd)
 	if err != nil {
-		rollback()
 		return nil, fmt.Errorf("x402: pay: %w", err)
 	}
 
@@ -123,7 +108,6 @@ func (c *Client) Do(req *http.Request) (*http.Response, error) {
 	}
 	retry, err := http.NewRequestWithContext(req.Context(), req.Method, req.URL.String(), rbody)
 	if err != nil {
-		rollback()
 		return nil, err
 	}
 	for k, v := range req.Header {
@@ -133,14 +117,14 @@ func (c *Client) Do(req *http.Request) (*http.Response, error) {
 
 	resp2, err := c.httpClient().Do(retry)
 	if err != nil {
-		rollback()
 		return nil, err
 	}
 	if resp2.StatusCode == http.StatusPaymentRequired {
-		rollback()
 		return resp2, fmt.Errorf("x402: payment for %s was rejected", reqd.Resource)
 	}
 
-	reserved = false
+	c.mu.Lock()
+	c.spent += amount
+	c.mu.Unlock()
 	return resp2, nil
 }

wrapper/x402/client_test.go

@@ -118,62 +118,3 @@ func TestClientFreeEndpoint(t *testing.T) {
 		t.Errorf("free call should spend nothing, got %d", c.Spent())
 	}
 }
-
-// Concurrent callers reserve budget before paying, so a spend cap cannot be
-// overshot by parallel tool calls that all observe the same pre-spend balance.
-func TestClientBudgetReservedConcurrently(t *testing.T) {
-	srv := paidServer("10000")
-	defer srv.Close()
-
-	c := &Client{Payer: &mockPayer{}, Budget: 10000}
-	errCh := make(chan error, 2)
-	for i := 0; i < 2; i++ {
-		go func() {
-			req, _ := http.NewRequest(http.MethodGet, srv.URL, nil)
-			resp, err := c.Do(req)
-			if resp != nil {
-				resp.Body.Close()
-			}
-			errCh <- err
-		}()
-	}
-
-	var paid, refused int
-	for i := 0; i < 2; i++ {
-		if err := <-errCh; err != nil {
-			refused++
-		} else {
-			paid++
-		}
-	}
-	if paid != 1 || refused != 1 {
-		t.Fatalf("paid=%d refused=%d, want one paid and one refused", paid, refused)
-	}
-	if c.Spent() != 10000 {
-		t.Errorf("spent = %d, want 10000", c.Spent())
-	}
-}
-
-// A reserved budget slot is released when payment cannot be produced, so a
-// transient payer failure does not permanently consume an agent's allowance.
-func TestClientBudgetReservationRollsBackOnPayError(t *testing.T) {
-	srv := paidServer("10000")
-	defer srv.Close()
-
-	c := &Client{Payer: payerFunc(func(context.Context, Requirements) (string, error) {
-		return "", context.Canceled
-	}), Budget: 10000}
-	req, _ := http.NewRequest(http.MethodGet, srv.URL, nil)
-	if _, err := c.Do(req); err == nil {
-		t.Fatal("expected pay error")
-	}
-	if c.Spent() != 0 {
-		t.Errorf("failed payment should roll back spend, got %d", c.Spent())
-	}
-}
-
-type payerFunc func(context.Context, Requirements) (string, error)
-
-func (f payerFunc) Pay(ctx context.Context, req Requirements) (string, error) {
-	return f(ctx, req)
-}

wrapper/x402/live_facilitator_test.go

@@ -1,52 +0,0 @@
-package x402
-
-import (
-	"context"
-	"os"
-	"testing"
-	"time"
-)
-
-// TestLiveFacilitatorConformance is an opt-in smoke test for hosted x402
-// facilitators. It is skipped by default so local and CI runs never need live
-// credentials, but operators can run it against Coinbase, Alchemy, or a
-// self-hosted facilitator by providing a real payment payload and matching
-// requirements.
-func TestLiveFacilitatorConformance(t *testing.T) {
-	url := os.Getenv("GO_MICRO_X402_LIVE_FACILITATOR_URL")
-	payment := os.Getenv("GO_MICRO_X402_LIVE_PAYMENT")
-	payTo := os.Getenv("GO_MICRO_X402_LIVE_PAY_TO")
-	if url == "" || payment == "" || payTo == "" {
-		t.Skip("set GO_MICRO_X402_LIVE_FACILITATOR_URL, GO_MICRO_X402_LIVE_PAYMENT, and GO_MICRO_X402_LIVE_PAY_TO to run live x402 facilitator conformance")
-	}
-
-	network := getenv("GO_MICRO_X402_LIVE_NETWORK", "base")
-	amount := getenv("GO_MICRO_X402_LIVE_AMOUNT", "1")
-	asset := os.Getenv("GO_MICRO_X402_LIVE_ASSET")
-	resource := getenv("GO_MICRO_X402_LIVE_RESOURCE", "go-micro-x402-live-conformance")
-
-	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
-	defer cancel()
-	res, err := (&HTTPFacilitator{URL: url}).Verify(ctx, payment, Requirements{
-		Scheme:            "exact",
-		Network:           network,
-		MaxAmountRequired: amount,
-		Resource:          resource,
-		PayTo:             payTo,
-		Asset:             asset,
-		MaxTimeoutSeconds: 60,
-	})
-	if err != nil {
-		t.Fatalf("live facilitator verify: %v", err)
-	}
-	if !res.Valid {
-		t.Fatalf("live facilitator rejected payment: %s", res.Reason)
-	}
-}
-
-func getenv(key, fallback string) string {
-	if v := os.Getenv(key); v != "" {
-		return v
-	}
-	return fallback
-}