Update architect priority queue

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

agent/agent.go

@@ -43,6 +43,7 @@ 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
@@ -172,6 +173,28 @@ 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(),
+	})
+}
+
 // 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.
@@ -333,13 +356,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.NewAgentHandler(card, func(ctx context.Context, text string) (string, error) {
+		handler := a2a.NewAgentStreamHandler(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/conformance_test.go

@@ -0,0 +1,158 @@
+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)
+	}
+}

ai/capabilities_test.go

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

ai/openai/openai.go

@@ -2,6 +2,7 @@
 package openai
 
 import (
+	"bufio"
 	"bytes"
 	"context"
 	"encoding/json"
@@ -20,6 +21,7 @@ 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
@@ -140,9 +142,88 @@ func (p *Provider) Generate(ctx context.Context, req *ai.Request, opts ...ai.Gen
 	return resp, nil
 }
 
-// Stream generates a streaming response (not yet implemented)
+// Stream generates a streaming response from the OpenAI chat completions API.
 func (p *Provider) Stream(ctx context.Context, req *ai.Request, opts ...ai.GenerateOption) (ai.Stream, error) {
-	return nil, fmt.Errorf("%w: openai provider", ai.ErrStreamingUnsupported)
+	messages := []map[string]any{
+		{"role": "system", "content": req.SystemPrompt},
+		{"role": "user", "content": req.Prompt},
+	}
+	apiReq := map[string]any{
+		"model":    p.opts.Model,
+		"messages": messages,
+		"stream":   true,
+	}
+	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"`
+		}
+		if err := json.Unmarshal([]byte(data), &chunk); err != nil {
+			return nil, fmt.Errorf("failed to parse stream chunk: %w", err)
+		}
+		if len(chunk.Choices) == 0 || chunk.Choices[0].Delta.Content == "" {
+			continue
+		}
+		return &ai.Response{Reply: chunk.Choices[0].Delta.Content}, nil
+	}
+	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()
 }
 
 // callAPI makes an HTTP request to the OpenAI API

ai/openai/openai_test.go

@@ -2,7 +2,11 @@ package openai
 
 import (
 	"context"
+	"encoding/json"
 	"errors"
+	"io"
+	"net/http"
+	"net/http/httptest"
 	"testing"
 
 	"go-micro.dev/v6/ai"
@@ -81,16 +85,44 @@ func TestProvider_Generate_NoAPIKey(t *testing.T) {
 	}
 }
 
-func TestProvider_Stream_NotImplemented(t *testing.T) {
-	p := NewProvider()
+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()
 
-	req := &ai.Request{
-		Prompt: "Hello",
+	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")
 	}
 
-	_, err := p.Stream(context.Background(), req)
-	if !errors.Is(err, ai.ErrStreamingUnsupported) {
-		t.Fatalf("Stream error = %v, want ErrStreamingUnsupported", err)
+	first, err := stream.Recv()
+	if err != nil || first.Reply != "hel" {
+		t.Fatalf("first chunk = %#v, %v; want hel", first, err)
+	}
+	second, err := stream.Recv()
+	if err != nil || second.Reply != "lo" {
+		t.Fatalf("second chunk = %#v, %v; want lo", second, err)
+	}
+	if _, err := stream.Recv(); !errors.Is(err, io.EOF) {
+		t.Fatalf("final error = %v, want EOF", err)
 	}
 }
 

cmd/micro/chat/chat.go

@@ -10,7 +10,9 @@ import (
 	"bufio"
 	"context"
 	"encoding/json"
+	"errors"
 	"fmt"
+	"io"
 	"os"
 	"os/exec"
 	"path/filepath"
@@ -79,6 +81,7 @@ 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,
 	})
@@ -102,6 +105,7 @@ 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
@@ -311,6 +315,7 @@ 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)
@@ -347,6 +352,7 @@ func run(c *cli.Context) error {
 		hist:          ai.NewHistory(50),
 		builtinTools:  builtinTools,
 		builtinHandle: builtinHandle,
+		stream:        streamOutput,
 	}
 	s.refreshTools()
 
@@ -449,12 +455,21 @@ func (s *session) ask(ctx context.Context, prompt string) error {
 	// Fallback: direct service access (no agents)
 	s.hist.Add("user", prompt)
 
-	resp, err := s.model.Generate(ctx, &ai.Request{
+	req := &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
 	}
@@ -489,6 +504,34 @@ 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.

gateway/a2a/a2a.go

@@ -29,6 +29,7 @@ import (
 	"context"
 	"encoding/json"
 	"fmt"
+	"io"
 	"log"
 	"net/http"
 	"strings"
@@ -36,6 +37,7 @@ 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"
@@ -92,6 +94,9 @@ 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
@@ -107,6 +112,19 @@ 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)
@@ -404,6 +422,10 @@ type dispatcher 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"})
@@ -418,6 +440,10 @@ func (d *dispatcher) serve(w http.ResponseWriter, r *http.Request, invoke Invoke
 	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)
@@ -460,6 +486,50 @@ 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})
+	var reply strings.Builder
+	for {
+		chunk, err := stream.Recv()
+		if err == io.EOF {
+			break
+		}
+		if err != nil {
+			_ = enc.Encode(rpcResponse{JSONRPC: "2.0", ID: req.ID, Error: &rpcError{Code: errInternal, Message: err.Error()}})
+			return
+		}
+		if chunk == nil || chunk.Reply == "" {
+			continue
+		}
+		reply.WriteString(chunk.Reply)
+		task := taskFromReply(p.Message, reply.String(), stateCompleted)
+		_ = enc.Encode(rpcResponse{JSONRPC: "2.0", ID: req.ID, Result: task})
+		if f, ok := w.(http.Flusher); ok {
+			f.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 {
@@ -471,32 +541,12 @@ func (d *dispatcher) run(ctx context.Context, params json.RawMessage, invoke Inv
 	}
 
 	reply, err := invoke(ctx, text)
-	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)},
-	}
+	state := stateCompleted
 	if err != nil {
 		reply = "error: " + err.Error()
-		task.Status.State = stateFailed
-	} else {
-		task.Status.State = stateCompleted
+		state = stateFailed
 	}
-	task.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",
-	})
+	task := taskFromReply(p.Message, reply, state)
 	d.store(task)
 	return task, nil
 }
@@ -559,6 +609,30 @@ func (d *dispatcher) store(t *Task) {
 	}
 }
 
+func taskFromReply(input Message, reply, state string) *Task {
+	contextID := input.ContextID
+	if contextID == "" {
+		contextID = uuid.New().String()
+	}
+	task := &Task{
+		ID:        uuid.New().String(),
+		ContextID: contextID,
+		Kind:      "task",
+		History:   []Message{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 textOf(parts []Part) string {
 	var b strings.Builder
 	for _, p := range parts {

health/health.go

@@ -314,7 +314,8 @@ 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.
+// than blocking the probe. Registries that honor ListContext also receive
+// the check context directly.
 func RegistryCheck(reg registry.Registry) CheckFunc {
 	return func(ctx context.Context) error {
 		if reg == nil {
@@ -322,7 +323,7 @@ func RegistryCheck(reg registry.Registry) CheckFunc {
 		}
 		errc := make(chan error, 1)
 		go func() {
-			_, err := reg.ListServices()
+			_, err := reg.ListServices(registry.ListContext(ctx))
 			errc <- err
 		}()
 		select {
@@ -337,6 +338,54 @@ 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,3 +89,60 @@ 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

@@ -0,0 +1,30 @@
+# 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.

internal/docs/PRIORITIES.md

@@ -21,20 +21,17 @@ changes, architectural rewrites. Those go to the human.
 
 ## Now (ranked)
 
-1. **Streaming end to end** (#3012) — `ai.Stream` through `micro chat`, the agent
-   RPC, and A2A `message/stream`; scope to chat + one provider first. Roadmap →
-   *Next* and now the highest-value lifecycle seam after durable agent resume
-   shipped.
-2. **Registry disconnection detection** (#2956) — readiness/health when a service
-   silently loses its registry connection. Community-requested production
-   reliability; keeps the service substrate operable because agents depend on
-   discovery.
-3. **Agent observability spans** (#3182) — export `RunInfo` as OpenTelemetry spans
+1. **Agent observability spans** (#3182) — export `RunInfo` as OpenTelemetry spans
    for agent runs, model calls, tool calls, delegation, and failures. Roadmap →
-   *Next*; makes the now-durable harness inspectable in production.
-4. **Execution lifecycle hooks & metadata** (#2980) — before/after-tool, retry,
+   *Next* agent observability; now the registry readiness gap (#2956), durable
+   resume, streaming, and provider conformance work have shipped, the biggest
+   remaining seam is production inspectability across the services → agents →
+   workflows runtime.
+2. **Execution lifecycle hooks & metadata** (#2980) — before/after-tool, retry,
    and failure hooks; first check overlap with the shipped run-timeline /
-   OpenTelemetry work and scope to what's not already covered.
+   OpenTelemetry work and scope to what's not already covered. Roadmap → *Next*
+   resilience/operability, but ranked after observability so new hooks attach to
+   the same run story instead of creating another seam.
 
 _Seeded by Claude Code from the roadmap + open issues; thereafter maintained by the
 architecture-review pass._

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 | No |
+| `openai` | Yes | Yes | No | Yes |
 | `together` | Yes | No | No | No |
 
 ## Step 1: Implement the `ai.Model` Interface