Update architect priorities after streaming conformance

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

.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
-# fails if any selected provider secret is missing, so scheduled conformance
-# cannot silently degrade.
+# skips providers whose secrets are absent, so scheduled conformance
+# remains safe in no-key forks while still failing configured providers that drift.
 
 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 plan-delegate workflow harness
-        run: go run ./internal/harness/plan-delegate
+      - name: 0→hero run/chat/inspect reference scenario
+        run: ./internal/harness/zero-to-hero-ci/run.sh
 
   harness-live:
     name: Provider harnesses (live LLM conformance)
@@ -50,6 +50,17 @@ 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 }}
@@ -61,7 +72,6 @@ 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

agent/agent.go

@@ -92,6 +92,11 @@ type agentImpl struct {
 	// Ask. The model provider only sees a refused tool result; the agent
 	// converts it into a durable paused run instead of completing the run.
 	pause *approvalPause
+
+	// currentRun points at the checkpoint record for the Ask currently
+	// holding mu. Tool execution updates it so resumed runs can reuse
+	// completed tool results without replaying side effects.
+	currentRun *flow.Run
 }
 
 // New creates a new Agent.
@@ -253,6 +258,8 @@ func (a *agentImpl) askLocked(ctx context.Context, runID, message, parentRunID s
 		Agent:    a.opts.Name,
 	})
 	run := a.newCheckpointRun(runID, message, parentRunID, existing)
+	a.currentRun = &run
+	defer func() { a.currentRun = nil }()
 	if err := a.saveRun(ctx, run); err != nil {
 		return nil, err
 	}
@@ -283,6 +290,9 @@ func (a *agentImpl) askLocked(ctx context.Context, runID, message, parentRunID s
 		run.Status = "failed"
 		run.Steps[0].Status = "failed"
 		run.Steps[0].Error = err.Error()
+		if a.currentRun != nil {
+			run.Steps = a.currentRun.Steps
+		}
 		_ = a.saveRun(ctx, run)
 		return nil, err
 	}
@@ -326,6 +336,12 @@ func (a *agentImpl) askLocked(ctx context.Context, runID, message, parentRunID s
 	if b, marshalErr := json.Marshal(res); marshalErr == nil {
 		run.State.Data = b
 	}
+	if a.currentRun != nil {
+		run.Steps = a.currentRun.Steps
+	}
+	if len(run.Steps) == 0 {
+		run.Steps = []flow.StepRecord{{Name: agentAskStep}}
+	}
 	run.Steps[0].Status = "done"
 	run.Steps[0].Attempts++
 	run.Steps[0].Result = reply

agent/builtin.go

@@ -102,8 +102,9 @@ func (a *agentImpl) toolHandler() ai.ToolHandler {
 
 	// Innermost first: base, then guardrails (approve → loop → step →
 	// plan), then developer wrappers outermost. Wrapping reverses order,
-	// so the result runs plan → step → loop → approve → base.
+	// so the result runs plan → step → loop → approve → checkpoint → base.
 	h := a.baseHandler()
+	h = a.checkpointToolWrap(h)
 	h = a.approveWrap(h)
 	h = a.loopWrap(h)
 	h = a.stepWrap(h)

agent/checkpoint.go

@@ -6,6 +6,7 @@ import (
 	"fmt"
 	"time"
 
+	"go-micro.dev/v6/ai"
 	"go-micro.dev/v6/flow"
 )
 
@@ -35,6 +36,7 @@ func (a *agentImpl) newCheckpointRun(runID, message, parentRunID string, existin
 		}
 		run.Steps[0].Status = "in_progress"
 		run.Steps[0].Error = ""
+		run.Steps[0].Result = ""
 	}
 	return run
 }
@@ -97,3 +99,60 @@ func (a *agentImpl) pending(ctx context.Context) ([]flow.Run, error) {
 	}
 	return out, nil
 }
+
+func (a *agentImpl) checkpointToolWrap(next ai.ToolHandler) ai.ToolHandler {
+	return func(ctx context.Context, call ai.ToolCall) ai.ToolResult {
+		if a.opts.Checkpoint == nil || a.currentRun == nil {
+			return next(ctx, call)
+		}
+		name := toolCheckpointName(call)
+		if rec, ok := findStep(a.currentRun.Steps, name); ok && rec.Status == "done" {
+			return ai.ToolResult{ID: call.ID, Value: rec.Result, Content: rec.Result}
+		}
+
+		idx := upsertStep(&a.currentRun.Steps, flow.StepRecord{Name: name, Status: "in_progress"})
+		_ = a.saveRun(ctx, *a.currentRun)
+		res := next(ctx, call)
+		a.currentRun.Steps[idx].Attempts++
+		if res.Refused != "" {
+			a.currentRun.Steps[idx].Status = "failed"
+			a.currentRun.Steps[idx].Error = res.Content
+			_ = a.saveRun(ctx, *a.currentRun)
+			return res
+		}
+		a.currentRun.Steps[idx].Status = "done"
+		a.currentRun.Steps[idx].Result = res.Content
+		a.currentRun.Steps[idx].Error = ""
+		_ = a.saveRun(ctx, *a.currentRun)
+		return res
+	}
+}
+
+func toolCheckpointName(call ai.ToolCall) string {
+	b, _ := json.Marshal(call.Input)
+	return "tool:" + call.Name + ":" + string(b)
+}
+
+func findStep(steps []flow.StepRecord, name string) (flow.StepRecord, bool) {
+	for _, step := range steps {
+		if step.Name == name {
+			return step, true
+		}
+	}
+	return flow.StepRecord{}, false
+}
+
+func upsertStep(steps *[]flow.StepRecord, rec flow.StepRecord) int {
+	for i := range *steps {
+		if (*steps)[i].Name == rec.Name {
+			(*steps)[i].Status = rec.Status
+			(*steps)[i].Error = rec.Error
+			return i
+		}
+	}
+	if len(*steps) == 0 || (*steps)[0].Name != agentAskStep {
+		*steps = append([]flow.StepRecord{{Name: agentAskStep, Status: "in_progress"}}, (*steps)...)
+	}
+	*steps = append(*steps, rec)
+	return len(*steps) - 1
+}

agent/checkpoint_test.go

@@ -2,6 +2,7 @@ package agent
 
 import (
 	"context"
+	"errors"
 	"testing"
 
 	"go-micro.dev/v6/ai"
@@ -48,6 +49,58 @@ func TestResumeCompletedCheckpointDoesNotReplayModel(t *testing.T) {
 	}
 }
 
+func TestResumeFailedCheckpointDoesNotReplayCompletedTool(t *testing.T) {
+	ctx := context.Background()
+	cp := flow.StoreCheckpoint(store.NewStore(), "tool-resume-agent")
+	toolRuns := 0
+	first := true
+	fakeGen = func(ctx context.Context, opts ai.Options, req *ai.Request) (*ai.Response, error) {
+		if opts.ToolHandler != nil {
+			res := opts.ToolHandler(ctx, ai.ToolCall{ID: "call-1", Name: "external.charge", Input: map[string]any{"order": "42"}})
+			if res.Content != "charged" {
+				t.Fatalf("tool result = %q, want charged", res.Content)
+			}
+		}
+		if first {
+			first = false
+			return nil, errors.New("model connection dropped after tool")
+		}
+		return &ai.Response{Reply: "finished from checkpoint"}, nil
+	}
+	defer func() { fakeGen = nil }()
+
+	a := newTestAgent(Name("tool-resume-agent"), WithCheckpoint(cp),
+		WithTool("external.charge", "charge once", nil, func(context.Context, map[string]any) (string, error) {
+			toolRuns++
+			return "charged", nil
+		}))
+	_, err := a.Ask(ctx, "charge order 42")
+	if err == nil {
+		t.Fatal("Ask succeeded, want simulated failure")
+	}
+	if toolRuns != 1 {
+		t.Fatalf("tool executions after failed Ask = %d, want 1", toolRuns)
+	}
+
+	runs, err := Pending(ctx, a)
+	if err != nil {
+		t.Fatalf("Pending: %v", err)
+	}
+	if len(runs) != 1 {
+		t.Fatalf("Pending returned %d runs, want 1", len(runs))
+	}
+	resp, err := Resume(ctx, a, runs[0].ID)
+	if err != nil {
+		t.Fatalf("Resume: %v", err)
+	}
+	if resp.Reply != "finished from checkpoint" {
+		t.Fatalf("Resume reply = %q", resp.Reply)
+	}
+	if toolRuns != 1 {
+		t.Fatalf("tool executions after Resume = %d, want completed tool was not replayed", toolRuns)
+	}
+}
+
 func TestPendingReturnsUnfinishedAgentRuns(t *testing.T) {
 	ctx := context.Background()
 	cp := flow.StoreCheckpoint(store.NewStore(), "pending-agent")

ai/openai/openai_test.go

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

ai/retry.go

@@ -13,9 +13,28 @@ 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
 }
 
@@ -23,7 +42,7 @@ func (e *RetryError) Error() string {
 	if e == nil {
 		return ""
 	}
-	return fmt.Sprintf("ai generate failed after %d attempt(s): %v", e.Attempts, e.Err)
+	return fmt.Sprintf("ai generate failed after %d attempt(s) (%s): %v", e.Attempts, e.ErrorKind(), e.Err)
 }
 
 func (e *RetryError) Unwrap() error {
@@ -33,6 +52,13 @@ 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
@@ -76,9 +102,10 @@ func GenerateWithRetry(ctx context.Context, m Model, req *Request, policy Genera
 		if ctx.Err() != nil {
 			return nil, ctx.Err()
 		}
-		if attempt == policy.MaxAttempts || !IsTransientError(err) {
-			if attempt > 1 || IsTransientError(err) {
-				return nil, &RetryError{Attempts: attempt, Err: err}
+		transient := IsTransientError(err)
+		if attempt == policy.MaxAttempts || !transient {
+			if attempt > 1 || transient {
+				return nil, &RetryError{Attempts: attempt, Kind: ClassifyError(err), Err: err}
 			}
 			return nil, err
 		}
@@ -106,25 +133,57 @@ func GenerateWithRetry(ctx context.Context, m Model, req *Request, policy Genera
 		case <-t.C:
 		}
 	}
-	return nil, &RetryError{Attempts: policy.MaxAttempts, Err: last}
+	return nil, &RetryError{Attempts: policy.MaxAttempts, Kind: ClassifyError(last), Err: last}
 }
 
-// IsTransientError reports whether err is worth retrying at the provider boundary.
-func IsTransientError(err error) bool {
+// ClassifyError maps provider and context failures to stable operational kinds.
+func ClassifyError(err error) ErrorKind {
 	if err == nil {
-		return false
+		return ""
+	}
+	var classified ClassifiedError
+	if errors.As(err, &classified) {
+		if kind := classified.ErrorKind(); kind != "" {
+			return kind
+		}
 	}
 	if errors.Is(err, context.Canceled) {
-		return false
+		return ErrorKindCanceled
 	}
 	if errors.Is(err, context.DeadlineExceeded) {
-		return true
+		return ErrorKindTimeout
 	}
 	var sc StatusCoder
 	if errors.As(err, &sc) {
 		code := sc.StatusCode()
-		return code == 429 || code >= 500
+		switch {
+		case code == 429:
+			return ErrorKindRateLimited
+		case code >= 500:
+			return ErrorKindUnavailable
+		case code > 0:
+			return ErrorKindProvider
+		}
 	}
 	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")
+	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
+	}
+}
+
+// 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:
+		return false
+	}
 }

ai/retry_test.go

@@ -133,3 +133,51 @@ func TestGenerateWithRetryAddsAttemptMetadataToRunInfo(t *testing.T) {
 		}
 	}
 }
+
+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/cli/new/contract_test.go

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

cmd/micro/zero_to_hero_cli_test.go

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

gateway/a2a/a2a_test.go

@@ -359,6 +359,79 @@ func TestMessageStreamChunksStoreFinalTask(t *testing.T) {
 	}
 }
 
+type contextStream struct {
+	ctx    context.Context
+	closed chan struct{}
+}
+
+func (s *contextStream) Recv() (*ai.Response, error) {
+	<-s.ctx.Done()
+	return nil, s.ctx.Err()
+}
+
+func (s *contextStream) Close() error {
+	close(s.closed)
+	return nil
+}
+
+func TestMessageStreamChunksPropagatesCancellationAndClosesStream(t *testing.T) {
+	d := newDispatcher()
+	ctx, cancel := context.WithCancel(context.Background())
+	closed := make(chan struct{})
+	body := `{"jsonrpc":"2.0","id":1,"method":"message/stream","params":{"message":{"role":"user","parts":[{"kind":"text","text":"ping"}],"kind":"message"}}}`
+	req := httptest.NewRequest(http.MethodPost, "/", bytes.NewBufferString(body)).WithContext(ctx)
+	rr := httptest.NewRecorder()
+	cancel()
+
+	d.serveWithStream(rr, req, nil, func(ctx context.Context, text string) (ai.Stream, error) {
+		if text != "ping" {
+			t.Fatalf("stream text = %q, want ping", text)
+		}
+		return &contextStream{ctx: ctx, closed: closed}, nil
+	})
+
+	select {
+	case <-closed:
+	case <-time.After(time.Second):
+		t.Fatal("stream was not closed")
+	}
+
+	var events []struct {
+		Result Task      `json:"result"`
+		Error  *rpcError `json:"error"`
+	}
+	for _, line := range strings.Split(strings.TrimSpace(rr.Body.String()), "\n") {
+		line = strings.TrimSpace(line)
+		if line == "" {
+			continue
+		}
+		line = strings.TrimPrefix(line, "data: ")
+		var event struct {
+			Result Task      `json:"result"`
+			Error  *rpcError `json:"error"`
+		}
+		if err := json.Unmarshal([]byte(line), &event); err != nil {
+			t.Fatalf("decode event %q: %v", line, err)
+		}
+		events = append(events, event)
+	}
+	if len(events) != 1 {
+		t.Fatalf("events = %d, want 1; body %s", len(events), rr.Body.String())
+	}
+	event := events[0]
+	if event.Error == nil || event.Error.Code != errInternal || event.Error.Message != context.Canceled.Error() {
+		t.Fatalf("error = %+v, want context cancellation", event.Error)
+	}
+	if event.Result.Status.State != stateFailed || textOf(event.Result.Artifacts[0].Parts) != "error: context canceled" {
+		t.Fatalf("failed task = %+v, want context cancellation artifact", event.Result)
+	}
+
+	got := rpcTaskFromDispatcher(t, d, event.Result.ID)
+	if got.Status.State != stateFailed || textOf(got.Artifacts[0].Parts) != "error: context canceled" {
+		t.Fatalf("stored task = %+v, want failed cancellation", got)
+	}
+}
+
 func 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)

internal/docs/AGENT_CONFORMANCE.md

@@ -28,3 +28,12 @@ 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/PRIORITIES.md

@@ -19,11 +19,13 @@ redirect the loop; direction always wins.
 items the loop can auto-merge): brand/positioning copy, breaking public-API
 changes, architectural rewrites. Those go to the human.
 
+## Next (ranked)
+
+1. **Export agent RunInfo as OpenTelemetry spans** ([#3256](https://github.com/micro/go-micro/issues/3256)) — connect the existing run/model/tool metadata to standard tracing so developers can inspect agent behavior with the same operational tools they use for services.
+
 ## Later (ranked)
 
-No open queued items. The previous top item, **A2A push notifications and
-multi-turn task support** (#3212), is closed; the next architecture-review pass
-should seed a fresh issue-backed item from the roadmap and improvement radar.
+2. **Add A2A resubscribe and input-required handoff support** ([#3235](https://github.com/micro/go-micro/issues/3235)) — after push notifications and multi-turn continuation shipped, finish the remaining long-running A2A interoperability gap: reconnecting to live task streams and carrying human-input-required handoffs through the gateway.
 
 _Seeded by Claude Code from the roadmap + open issues; thereafter maintained by the
 architecture-review pass._

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

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

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