Compare commits

..

1 Commits

Author SHA1 Message Date
Codex a3c9c40881 docs(priorities): refresh architect queue
Harness (E2E) / Harnesses (mock LLM) (push) Waiting to run
Harness (E2E) / Provider harnesses (live LLM conformance) (push) Waiting to run
Lint / golangci-lint (push) Waiting to run
Run Tests / Unit Tests (push) Waiting to run
Run Tests / Etcd Integration Tests (push) Waiting to run
2026-06-29 18:06:30 +00:00
43 changed files with 160 additions and 2085 deletions
+22 -31
View File
@@ -14,20 +14,6 @@ on:
schedule:
- cron: "17 6 * * *" # daily, so the world is exercised even without changes
workflow_dispatch:
inputs:
providers:
description: "Comma-separated providers for live conformance (default: all supported)"
required: false
default: "anthropic,openai,gemini,groq,mistral,together,atlascloud"
harnesses:
description: "Comma-separated harnesses for live conformance"
required: false
default: "agent,universe,agent-flow,plan-delegate,a2a-stream-fallback"
require_configured:
description: "Fail selected live providers that do not have repository secrets"
required: false
type: boolean
default: false
jobs:
harness:
@@ -41,8 +27,14 @@ jobs:
cache: true
- name: Build
run: go build ./...
- name: 0→1 and 0→hero developer-flow harness
run: make harness
- name: 0→1 scaffold contract
run: go test ./cmd/micro/cli/new -run TestZeroToOneContract -count=1
- name: Universe end-to-end (asserts; exits non-zero on failure)
run: go run ./internal/harness/universe
- name: Agent-flow harness
run: go run ./internal/harness/agent-flow
- name: 0→hero run/chat/inspect reference scenario
run: ./internal/harness/zero-to-hero-ci/run.sh
harness-live:
name: Provider harnesses (live LLM conformance)
@@ -58,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 }}
@@ -68,22 +71,10 @@ jobs:
TOGETHER_API_KEY: ${{ secrets.TOGETHER_API_KEY }}
ATLASCLOUD_API_KEY: ${{ secrets.ATLASCLOUD_API_KEY }}
run: |
PROVIDERS="${{ github.event.inputs.providers || 'anthropic,openai,gemini,groq,mistral,together,atlascloud' }}"
HARNESSES="${{ github.event.inputs.harnesses || 'agent,universe,agent-flow,plan-delegate,a2a-stream-fallback' }}"
REQUIRE_CONFIGURED="${{ github.event.inputs.require_configured || 'false' }}"
args=(
-providers "$PROVIDERS"
-harnesses "$HARNESSES"
-summary-json provider-conformance-summary.json
-summary-markdown provider-conformance-summary.md
go run ./internal/harness/provider-conformance \
-summary-json provider-conformance-summary.json \
-summary-markdown provider-conformance-summary.md \
-capabilities-markdown provider-capabilities.md
)
if [ "$REQUIRE_CONFIGURED" = "true" ]; then
args+=( -require-configured )
fi
go run ./internal/harness/provider-conformance "${args[@]}"
- name: Publish provider conformance summary
if: always()
run: |
+1 -2
View File
@@ -47,10 +47,9 @@ test-coverage:
# This mirrors the default CI path so local dogfooding catches scaffold,
# run/chat/inspect, and 0→hero regressions before a PR is opened.
harness:
go test ./cmd/micro/cli/new -run TestZeroToOne -count=1
go test ./cmd/micro/cli/new -run TestZeroToOneContract -count=1
./internal/harness/zero-to-hero-ci/run.sh
go run ./internal/harness/agent-flow
go run ./internal/harness/provider-conformance -providers mock
# Run the same harnesses against every configured live provider. Providers
# without API keys are skipped; configured providers must pass.
-8
View File
@@ -64,14 +64,6 @@ curl -X POST http://localhost:8080/api/helloworld/Helloworld.Call \
-H 'Content-Type: application/json' -d '{"name":"World"}'
```
This scaffold → run → call path is covered by the no-secret CI harness. To run
the same local contract (including the 0→hero services → agents → workflows path,
chat/inspect CLI boundaries, and deploy dry-run), use:
```bash
make harness
```
### Generate from a prompt — with an LLM key
Set a provider key, describe what you want, and the AI designs services, writes handlers, compiles, and starts them:
+1 -20
View File
@@ -113,14 +113,13 @@ func Builtins(opts ...Option) (tools []ai.Tool, handle func(name string, input m
// prevents runaway recursion).
func (a *agentImpl) toolHandler() ai.ToolHandler {
if a.ephemeral {
return a.toolTimeoutWrap(a.tools.Handler())
return a.tools.Handler()
}
// Innermost first: base, then guardrails (approve → loop → step →
// plan), then developer wrappers outermost. Wrapping reverses order,
// so the result runs plan → step → loop → approve → checkpoint → base.
h := a.baseHandler()
h = a.toolTimeoutWrap(h)
h = a.checkpointToolWrap(h)
h = a.approveWrap(h)
h = a.loopWrap(h)
@@ -149,21 +148,6 @@ func contextWrap(next ai.ToolHandler) ai.ToolHandler {
}
}
// toolTimeoutWrap gives each tool execution its own deadline while preserving
// caller cancellation. Handlers still execute synchronously; tools that honor
// context (custom tools, delegate RPC/A2A, and go-micro RPC clients) return
// promptly with a bounded error result when the deadline expires.
func (a *agentImpl) toolTimeoutWrap(next ai.ToolHandler) ai.ToolHandler {
return func(ctx context.Context, call ai.ToolCall) ai.ToolResult {
if a.opts.ToolTimeout <= 0 {
return next(ctx, call)
}
toolCtx, cancel := context.WithTimeout(ctx, a.opts.ToolTimeout)
defer cancel()
return next(toolCtx, call)
}
}
// baseHandler executes a tool call: a developer custom tool, the built-in
// delegate, or an RPC to the service. It is the innermost handler.
func (a *agentImpl) baseHandler() ai.ToolHandler {
@@ -335,9 +319,6 @@ func (a *agentImpl) handleDelegate(ctx context.Context, call ai.ToolCall) ai.Too
WithRegistry(a.opts.Registry),
WithClient(a.opts.Client),
WithStore(a.opts.Store),
ModelCallTimeout(a.opts.ModelTimeout),
ModelRetry(a.opts.ModelMaxAttempts, a.opts.ModelRetryBackoff),
ToolCallTimeout(a.opts.ToolTimeout),
TraceProvider(a.opts.TraceProvider),
)
// Record lineage so the sub-agent's tool calls carry this run as parent.
+1 -13
View File
@@ -88,9 +88,6 @@ func (a *agentImpl) resume(ctx context.Context, runID string) (*Response, error)
}
return &resp, nil
}
if terminalAgentRunStatus(run.Status) {
return nil, fmt.Errorf("agent run %s is terminal with status %q", runID, run.Status)
}
message := string(run.State.Data)
parentID := run.ParentID
a.mu.Lock()
@@ -156,22 +153,13 @@ func (a *agentImpl) pending(ctx context.Context) ([]flow.Run, error) {
}
out := runs[:0]
for _, run := range runs {
if run.Flow == a.opts.Name && !terminalAgentRunStatus(run.Status) {
if run.Flow == a.opts.Name && run.Status != "done" {
out = append(out, run)
}
}
return out, nil
}
func terminalAgentRunStatus(status string) bool {
switch status {
case "done", "canceled", "expired":
return true
default:
return false
}
}
func (a *agentImpl) checkpointToolWrap(next ai.ToolHandler) ai.ToolHandler {
return func(ctx context.Context, call ai.ToolCall) ai.ToolResult {
if a.opts.Checkpoint == nil || a.currentRun == nil {
+5 -79
View File
@@ -13,7 +13,7 @@ import (
func TestResumeCompletedCheckpointDoesNotReplayModel(t *testing.T) {
ctx := context.Background()
cp := flow.StoreCheckpoint(store.NewMemoryStore(), "durable-agent")
cp := flow.StoreCheckpoint(store.NewStore(), "durable-agent")
calls := 0
fakeGen = func(ctx context.Context, opts ai.Options, req *ai.Request) (*ai.Response, error) {
calls++
@@ -52,7 +52,7 @@ func TestResumeCompletedCheckpointDoesNotReplayModel(t *testing.T) {
func TestResumeFailedCheckpointDoesNotReplayCompletedTool(t *testing.T) {
ctx := context.Background()
cp := flow.StoreCheckpoint(store.NewMemoryStore(), "tool-resume-agent")
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) {
@@ -170,7 +170,7 @@ func countMemoryContent(messages []ai.Message, needle string) int {
func TestPendingReturnsUnfinishedAgentRuns(t *testing.T) {
ctx := context.Background()
cp := flow.StoreCheckpoint(store.NewMemoryStore(), "pending-agent")
cp := flow.StoreCheckpoint(store.NewStore(), "pending-agent")
run := flow.Run{ID: "run-1", Flow: "pending-agent", Status: "failed", State: flow.State{Stage: agentAskStep, Data: []byte("retry me")}}
if err := cp.Save(ctx, run); err != nil {
t.Fatalf("Save: %v", err)
@@ -185,38 +185,9 @@ func TestPendingReturnsUnfinishedAgentRuns(t *testing.T) {
}
}
func TestPendingSkipsTerminalCanceledAndExpiredAgentRuns(t *testing.T) {
ctx := context.Background()
cp := flow.StoreCheckpoint(store.NewMemoryStore(), "terminal-agent")
for _, run := range []flow.Run{
{ID: "active", Flow: "terminal-agent", Status: "failed", State: flow.State{Stage: agentAskStep, Data: []byte("retry me")}},
{ID: "done", Flow: "terminal-agent", Status: "done", State: flow.State{Stage: agentAskStep, Data: []byte("done")}},
{ID: "canceled", Flow: "terminal-agent", Status: "canceled", State: flow.State{Stage: agentAskStep, Data: []byte("canceled")}},
{ID: "expired", Flow: "terminal-agent", Status: "expired", State: flow.State{Stage: agentAskStep, Data: []byte("expired")}},
} {
if err := cp.Save(ctx, run); err != nil {
t.Fatalf("Save(%s): %v", run.ID, err)
}
}
a := newTestAgent(Name("terminal-agent"), WithCheckpoint(cp))
runs, err := Pending(ctx, a)
if err != nil {
t.Fatalf("Pending: %v", err)
}
if len(runs) != 1 || runs[0].ID != "active" {
t.Fatalf("Pending = %#v, want only active failed run", runs)
}
for _, id := range []string{"canceled", "expired"} {
if _, err := Resume(ctx, a, id); err == nil || !strings.Contains(err.Error(), "terminal") {
t.Fatalf("Resume(%s) err = %v, want terminal status error", id, err)
}
}
}
func TestHumanInputPauseResumesSameRunWithInput(t *testing.T) {
ctx := context.Background()
cp := flow.StoreCheckpoint(store.NewMemoryStore(), "input-agent")
cp := flow.StoreCheckpoint(store.NewStore(), "input-agent")
calls := 0
fakeGen = func(ctx context.Context, opts ai.Options, req *ai.Request) (*ai.Response, error) {
calls++
@@ -272,54 +243,9 @@ func TestHumanInputPauseResumesSameRunWithInput(t *testing.T) {
}
}
func TestHumanInputResumeHonorsCanceledContextAndLeavesRunPending(t *testing.T) {
ctx := context.Background()
cp := flow.StoreCheckpoint(store.NewMemoryStore(), "input-cancel-agent")
fakeGen = func(ctx context.Context, opts ai.Options, req *ai.Request) (*ai.Response, error) {
if opts.ToolHandler != nil {
opts.ToolHandler(ctx, ai.ToolCall{ID: "input-1", Name: toolHumanInput, Input: map[string]any{"prompt": "Approve deploy?"}})
}
return &ai.Response{Reply: "waiting"}, nil
}
defer func() { fakeGen = nil }()
a := newTestAgent(Name("input-cancel-agent"), WithCheckpoint(cp))
if _, err := a.Ask(ctx, "deploy the service"); err == nil {
t.Fatal("Ask succeeded, want input-required pause")
}
runs, err := Pending(ctx, a)
if err != nil {
t.Fatalf("Pending: %v", err)
}
if len(runs) != 1 {
t.Fatalf("Pending returned %d runs, want 1: %#v", len(runs), runs)
}
canceled, cancel := context.WithCancel(ctx)
cancel()
if _, err := ResumeInput(canceled, a, runs[0].ID, "yes"); !errors.Is(err, context.Canceled) {
t.Fatalf("ResumeInput canceled err = %v, want context.Canceled", err)
}
loaded, ok, err := cp.Load(ctx, runs[0].ID)
if err != nil || !ok {
t.Fatalf("Load paused run ok=%v err=%v", ok, err)
}
if loaded.Status != "paused" || loaded.State.Stage != agentInputStep {
t.Fatalf("run status/stage after canceled resume = %s/%s, want paused/%s", loaded.Status, loaded.State.Stage, agentInputStep)
}
var pause inputPause
if err := loaded.State.Scan(&pause); err != nil {
t.Fatalf("Scan pause after canceled resume: %v", err)
}
if pause.OriginalMessage != "deploy the service" || pause.Prompt != "Approve deploy?" {
t.Fatalf("pause after canceled resume = %#v", pause)
}
}
func TestApprovalDenialPausesCheckpointedRunAndResumeContinues(t *testing.T) {
ctx := context.Background()
cp := flow.StoreCheckpoint(store.NewMemoryStore(), "approval-agent")
cp := flow.StoreCheckpoint(store.NewStore(), "approval-agent")
calls := 0
fakeGen = func(ctx context.Context, opts ai.Options, req *ai.Request) (*ai.Response, error) {
calls++
-15
View File
@@ -33,29 +33,14 @@ func TestAgentProviderConformanceMatrix(t *testing.T) {
{name: "together", key: "TOGETHER_API_KEY", model: "GO_MICRO_CONFORMANCE_TOGETHER_MODEL", live: true},
}
selected := selectedConformanceProviders(os.Getenv("GO_MICRO_AGENT_CONFORMANCE_PROVIDERS"))
for _, provider := range providers {
provider := provider
if len(selected) > 0 && !selected[provider.name] {
continue
}
t.Run(provider.name, func(t *testing.T) {
runAgentConformanceScenario(t, provider)
})
}
}
func selectedConformanceProviders(csv string) map[string]bool {
out := map[string]bool{}
for _, part := range strings.Split(csv, ",") {
part = strings.TrimSpace(part)
if part != "" {
out[part] = true
}
}
return out
}
func runAgentConformanceScenario(t *testing.T, provider conformanceProvider) {
t.Helper()
if provider.live {
-13
View File
@@ -56,10 +56,6 @@ type Options struct {
// ModelRetryBackoff is the base delay between transient provider failures
// (grows exponentially per attempt when retries are enabled).
ModelRetryBackoff time.Duration
// ToolTimeout bounds each tool execution (0 disables). The timeout is
// applied before custom tools, delegate, and service RPC calls so context
// deadlines propagate consistently through the agent loop.
ToolTimeout time.Duration
// Memory is the agent's conversation memory. Nil = the default
// store-backed memory (durable across restarts).
@@ -116,7 +112,6 @@ func newOptions(opts ...Option) Options {
ModelTimeout: 30 * time.Second,
ModelMaxAttempts: 1, // retries opt-in via ModelRetry (see field doc)
ModelRetryBackoff: 100 * time.Millisecond,
ToolTimeout: 30 * time.Second,
// On by default and lenient: identical repeated calls are a
// no-progress loop, never useful. Set LoopLimit(0) to disable.
LoopLimit: 3,
@@ -209,14 +204,6 @@ func ModelCallTimeout(d time.Duration) Option {
return func(o *Options) { o.ModelTimeout = d }
}
// ToolCallTimeout sets the timeout for each tool execution. It bounds custom
// tools, built-in delegate calls, and service RPC tools with the same context
// deadline so mid-run cancellation and slow tools produce safe error results
// instead of unbounded agent runs. Set 0 to disable.
func ToolCallTimeout(d time.Duration) Option {
return func(o *Options) { o.ToolTimeout = d }
}
// ModelRetry sets the provider retry budget and backoff for transient failures.
func ModelRetry(maxAttempts int, backoff time.Duration) Option {
return func(o *Options) {
+29 -68
View File
@@ -38,7 +38,6 @@ const (
AttrGuardrailBlock = "agent.guardrail.block"
AttrRefusal = "agent.refusal"
AttrInputChars = "agent.input.chars"
AttrErrorKind = "agent.error.kind"
)
type RunEvent struct {
@@ -58,7 +57,6 @@ type RunEvent struct {
Tokens Usage `json:"tokens,omitempty"`
Refused string `json:"refused,omitempty"`
Error string `json:"error,omitempty"`
ErrorKind string `json:"error_kind,omitempty"`
InputChars int `json:"input_chars,omitempty"`
}
@@ -68,8 +66,7 @@ type Usage = ai.Usage
// Zero values preserve the full deterministic run list.
type RunListOptions struct {
// Status, when set, keeps only runs with the matching status
// (for example "running", "done", "canceled", "timeout",
// "rate_limited", "error", or "refused").
// (for example "running", "done", "error", or "refused").
Status string
// TraceID, when set, keeps only runs correlated with this trace id.
// A prefix is accepted so operators can paste the shortened trace id
@@ -82,19 +79,18 @@ type RunListOptions struct {
// RunSummary is a compact index entry for a recorded agent run.
type RunSummary struct {
RunID string `json:"run_id"`
Agent string `json:"agent"`
ParentID string `json:"parent_id,omitempty"`
TraceID string `json:"trace_id,omitempty"`
SpanID string `json:"span_id,omitempty"`
StartedAt time.Time `json:"started_at"`
UpdatedAt time.Time `json:"updated_at"`
DurationMS int64 `json:"duration_ms,omitempty"`
Events int `json:"events"`
Status string `json:"status,omitempty"`
LastKind string `json:"last_kind,omitempty"`
LastError string `json:"last_error,omitempty"`
LastErrorKind string `json:"last_error_kind,omitempty"`
RunID string `json:"run_id"`
Agent string `json:"agent"`
ParentID string `json:"parent_id,omitempty"`
TraceID string `json:"trace_id,omitempty"`
SpanID string `json:"span_id,omitempty"`
StartedAt time.Time `json:"started_at"`
UpdatedAt time.Time `json:"updated_at"`
DurationMS int64 `json:"duration_ms,omitempty"`
Events int `json:"events"`
Status string `json:"status,omitempty"`
LastKind string `json:"last_kind,omitempty"`
LastError string `json:"last_error,omitempty"`
}
func (a *agentImpl) tracer() trace.Tracer {
@@ -114,7 +110,7 @@ func (a *agentImpl) startRun(ctx context.Context, message string) (context.Conte
return ctx, func(err error) {
latency := time.Since(start).Milliseconds()
if err != nil {
a.recordRunEvent(RunEvent{Time: time.Now(), RunID: info.RunID, ParentID: info.ParentID, Agent: info.Agent, Kind: "error", LatencyMS: latency, Error: err.Error(), ErrorKind: string(ai.ClassifyError(err))})
a.recordRunEvent(RunEvent{Time: time.Now(), RunID: info.RunID, ParentID: info.ParentID, Agent: info.Agent, Kind: "error", LatencyMS: latency, Error: err.Error()})
return
}
a.recordRunEvent(RunEvent{Time: time.Now(), RunID: info.RunID, ParentID: info.ParentID, Agent: info.Agent, Kind: "done", LatencyMS: latency})
@@ -128,10 +124,9 @@ func (a *agentImpl) startRun(ctx context.Context, message string) (context.Conte
latency := time.Since(start).Milliseconds()
span.SetAttributes(attribute.Int64(AttrLatencyMS, latency))
if err != nil {
span.SetAttributes(attribute.String(AttrErrorKind, string(ai.ClassifyError(err))))
span.RecordError(err)
span.SetStatus(codes.Error, err.Error())
a.recordSpanEvent(span, RunEvent{Time: time.Now(), RunID: info.RunID, ParentID: info.ParentID, Agent: info.Agent, Kind: "error", LatencyMS: latency, Error: err.Error(), ErrorKind: string(ai.ClassifyError(err))})
a.recordSpanEvent(span, RunEvent{Time: time.Now(), RunID: info.RunID, ParentID: info.ParentID, Agent: info.Agent, Kind: "error", LatencyMS: latency, Error: err.Error()})
} else {
span.SetStatus(codes.Ok, "")
a.recordSpanEvent(span, RunEvent{Time: time.Now(), RunID: info.RunID, ParentID: info.ParentID, Agent: info.Agent, Kind: "done", LatencyMS: latency})
@@ -162,7 +157,6 @@ func (m *tracedModel) Generate(ctx context.Context, req *ai.Request, opts ...ai.
e := RunEvent{Time: time.Now(), RunID: info.RunID, ParentID: info.ParentID, Agent: info.Agent, Kind: "model", Provider: provider, Model: model, Attempt: info.Attempt, MaxAttempts: info.MaxAttempts, LatencyMS: dur, Tokens: usage}
if err != nil {
e.Error = err.Error()
e.ErrorKind = string(ai.ClassifyError(err))
}
m.a.recordRunEvent(e)
return resp, err
@@ -191,7 +185,6 @@ func (m *tracedModel) Generate(ctx context.Context, req *ai.Request, opts ...ai.
}
span.SetAttributes(attrs...)
if err != nil {
span.SetAttributes(attribute.String(AttrErrorKind, string(ai.ClassifyError(err))))
span.RecordError(err)
span.SetStatus(codes.Error, err.Error())
} else {
@@ -201,7 +194,6 @@ func (m *tracedModel) Generate(ctx context.Context, req *ai.Request, opts ...ai.
e := RunEvent{Time: time.Now(), RunID: info.RunID, ParentID: info.ParentID, Agent: info.Agent, Kind: "model", Provider: provider, Model: model, Attempt: info.Attempt, MaxAttempts: info.MaxAttempts, LatencyMS: dur, Tokens: usage}
if err != nil {
e.Error = err.Error()
e.ErrorKind = string(ai.ClassifyError(err))
}
m.a.recordSpanEvent(span, e)
return resp, err
@@ -228,8 +220,7 @@ func (a *agentImpl) traceTool(next ai.ToolHandler) ai.ToolHandler {
if a.opts.TraceProvider == nil {
res := next(ctx, call)
dur := time.Since(start).Milliseconds()
resErr := resultError(res)
a.recordRunEvent(RunEvent{Time: time.Now(), RunID: info.RunID, ParentID: info.ParentID, Agent: info.Agent, Kind: "tool", Name: call.Name, LatencyMS: dur, Refused: res.Refused, Error: resErr, ErrorKind: classifyToolError(resErr)})
a.recordRunEvent(RunEvent{Time: time.Now(), RunID: info.RunID, ParentID: info.ParentID, Agent: info.Agent, Kind: "tool", Name: call.Name, LatencyMS: dur, Refused: res.Refused, Error: resultError(res)})
return res
}
@@ -246,11 +237,8 @@ func (a *agentImpl) traceTool(next ai.ToolHandler) ai.ToolHandler {
if res.Refused != "" {
attrs = append(attrs, attribute.Bool(AttrGuardrailBlock, true), attribute.String(AttrRefusal, res.Refused))
}
resErr := resultError(res)
if kind := classifyToolError(resErr); kind != "" {
attrs = append(attrs, attribute.String(AttrErrorKind, kind))
}
span.SetAttributes(attrs...)
resErr := resultError(res)
if res.Refused != "" {
span.SetStatus(codes.Error, res.Refused)
} else if resErr != "" {
@@ -259,7 +247,7 @@ func (a *agentImpl) traceTool(next ai.ToolHandler) ai.ToolHandler {
span.SetStatus(codes.Ok, "")
}
span.End()
a.recordSpanEvent(span, RunEvent{Time: time.Now(), RunID: info.RunID, ParentID: info.ParentID, Agent: info.Agent, Kind: "tool", Name: call.Name, LatencyMS: dur, Refused: res.Refused, Error: resErr, ErrorKind: classifyToolError(resErr)})
a.recordSpanEvent(span, RunEvent{Time: time.Now(), RunID: info.RunID, ParentID: info.ParentID, Agent: info.Agent, Kind: "tool", Name: call.Name, LatencyMS: dur, Refused: res.Refused, Error: resErr})
return res
}
}
@@ -276,19 +264,6 @@ func resultError(res ai.ToolResult) string {
return ""
}
func classifyToolError(err string) string {
switch {
case err == "":
return ""
case strings.Contains(strings.ToLower(err), "context canceled"):
return string(ai.ErrorKindCanceled)
case strings.Contains(strings.ToLower(err), "deadline exceeded"):
return string(ai.ErrorKindTimeout)
default:
return string(ai.ErrorKindProvider)
}
}
func (a *agentImpl) recordSpanEvent(span trace.Span, e RunEvent) {
if sc := span.SpanContext(); sc.IsValid() {
e.TraceID = sc.TraceID().String()
@@ -334,9 +309,6 @@ func runEventAttributes(e RunEvent) []attribute.KeyValue {
if e.Error != "" {
attrs = append(attrs, attribute.String("agent.error", e.Error))
}
if e.ErrorKind != "" {
attrs = append(attrs, attribute.String(AttrErrorKind, e.ErrorKind))
}
return attrs
}
@@ -416,9 +388,6 @@ func ListRunSummariesWithOptions(s store.Store, agentName string, opts RunListOp
if e.Error != "" {
summary.LastError = e.Error
}
if e.ErrorKind != "" {
summary.LastErrorKind = e.ErrorKind
}
}
if opts.Status != "" && summary.Status != opts.Status {
continue
@@ -445,32 +414,24 @@ func runStatus(events []RunEvent) string {
}
status := "running"
for _, e := range events {
if e.Refused != "" && status == "running" {
if e.Error != "" {
status = "error"
}
if e.Refused != "" && status != "error" {
status = "refused"
}
if e.Error != "" || e.Kind == "error" {
status = runErrorStatus(e.ErrorKind)
}
if e.Kind == "done" && status == "running" {
status = "done"
switch e.Kind {
case "error":
status = "error"
case "done":
if status == "running" {
status = "done"
}
}
}
return status
}
func runErrorStatus(kind string) string {
switch ai.ErrorKind(kind) {
case ai.ErrorKindCanceled:
return "canceled"
case ai.ErrorKindTimeout:
return "timeout"
case ai.ErrorKindRateLimited:
return "rate_limited"
default:
return "error"
}
}
func LoadRunEvents(s store.Store, agentName, runID string) ([]RunEvent, error) {
st := store.Scope(s, "agent", agentName)
keys, err := st.List(store.ListPrefix("runs/" + runID + "/"))
+7 -31
View File
@@ -241,7 +241,7 @@ func TestAgentOpenTelemetrySpansModelFailure(t *testing.T) {
sawRunError = true
}
case spanNameModelCall:
if attrs[AttrAgentName] == "failing-runner" && attrs[AttrAttempt] == "1" && attrs[AttrErrorKind] == string(ai.ErrorKindUnknown) && s.Status().Code == codesError {
if attrs[AttrAgentName] == "failing-runner" && attrs[AttrAttempt] == "1" && s.Status().Code == codesError {
sawModelError = true
}
}
@@ -263,7 +263,7 @@ func TestAgentOpenTelemetrySpansModelFailure(t *testing.T) {
}
var sawModelEvent bool
for _, event := range events {
if event.Kind == "model" && event.Attempt == 1 && event.MaxAttempts == 1 && event.Error != "" && event.ErrorKind == string(ai.ErrorKindUnknown) {
if event.Kind == "model" && event.Attempt == 1 && event.MaxAttempts == 1 && event.Error != "" {
sawModelEvent = true
}
}
@@ -429,7 +429,7 @@ func TestListRunSummaries(t *testing.T) {
{Time: time.Unix(0, 1), RunID: "run-a", Agent: "runner", TraceID: "trace-a", SpanID: "span-a", Kind: "run", Name: "first"},
{Time: time.Unix(0, 2), RunID: "run-a", Agent: "runner", Kind: "tool", Name: "probe"},
{Time: time.Unix(0, 3), RunID: "run-b", Agent: "runner", ParentID: "parent", Kind: "run", Name: "second"},
{Time: time.Unix(0, 4), RunID: "run-b", Agent: "runner", ParentID: "parent", Kind: "error", Error: "context deadline exceeded", ErrorKind: string(ai.ErrorKindTimeout)},
{Time: time.Unix(0, 4), RunID: "run-b", Agent: "runner", ParentID: "parent", Kind: "error", Error: "boom"},
}
for _, e := range events {
b, err := json.Marshal(e)
@@ -452,35 +452,11 @@ func TestListRunSummaries(t *testing.T) {
if got[0].RunID != "run-a" || got[0].TraceID != "trace-a" || got[0].SpanID != "span-a" || got[0].Events != 2 || got[0].Status != "running" || got[0].DurationMS != 0 || got[0].LastKind != "tool" || !got[0].UpdatedAt.Equal(time.Unix(0, 2)) {
t.Fatalf("unexpected run-a summary: %#v", got[0])
}
if got[1].RunID != "run-b" || got[1].ParentID != "parent" || got[1].Events != 2 || got[1].Status != "timeout" || got[1].DurationMS != 0 || got[1].LastKind != "error" || got[1].LastError != "context deadline exceeded" || got[1].LastErrorKind != string(ai.ErrorKindTimeout) {
if got[1].RunID != "run-b" || got[1].ParentID != "parent" || got[1].Events != 2 || got[1].Status != "error" || got[1].DurationMS != 0 || got[1].LastKind != "error" || got[1].LastError != "boom" {
t.Fatalf("unexpected run-b summary: %#v", got[1])
}
}
func TestRunStatusClassifiesOperationalErrorKinds(t *testing.T) {
tests := []struct {
name string
kind ai.ErrorKind
want string
}{
{name: "canceled", kind: ai.ErrorKindCanceled, want: "canceled"},
{name: "timeout", kind: ai.ErrorKindTimeout, want: "timeout"},
{name: "rate limited", kind: ai.ErrorKindRateLimited, want: "rate_limited"},
{name: "provider", kind: ai.ErrorKindProvider, want: "error"},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
got := runStatus([]RunEvent{
{Kind: "run"},
{Kind: "error", Error: "failed", ErrorKind: string(tt.kind)},
})
if got != tt.want {
t.Fatalf("runStatus() = %q, want %q", got, tt.want)
}
})
}
}
func TestListRunSummariesWithOptionsFiltersAndLimits(t *testing.T) {
st := store.NewMemoryStore()
scoped := store.Scope(st, "agent", "runner")
@@ -488,7 +464,7 @@ func TestListRunSummariesWithOptionsFiltersAndLimits(t *testing.T) {
{Time: time.Unix(0, 1), RunID: "run-old", Agent: "runner", Kind: "run"},
{Time: time.Unix(0, 2), RunID: "run-old", Agent: "runner", Kind: "done"},
{Time: time.Unix(0, 3), RunID: "run-new", Agent: "runner", TraceID: "abcdef1234567890", Kind: "run"},
{Time: time.Unix(0, 4), RunID: "run-new", Agent: "runner", Kind: "error", Error: "rate limit exceeded", ErrorKind: string(ai.ErrorKindRateLimited)},
{Time: time.Unix(0, 4), RunID: "run-new", Agent: "runner", Kind: "error", Error: "boom"},
}
for _, e := range events {
b, err := json.Marshal(e)
@@ -500,11 +476,11 @@ func TestListRunSummariesWithOptionsFiltersAndLimits(t *testing.T) {
}
}
got, err := ListRunSummariesWithOptions(st, "runner", RunListOptions{Status: "rate_limited", TraceID: "abcdef", Limit: 1})
got, err := ListRunSummariesWithOptions(st, "runner", RunListOptions{Status: "error", TraceID: "abcdef", Limit: 1})
if err != nil {
t.Fatal(err)
}
if len(got) != 1 || got[0].RunID != "run-new" || got[0].Status != "rate_limited" {
if len(got) != 1 || got[0].RunID != "run-new" || got[0].Status != "error" {
t.Fatalf("filtered summaries = %#v", got)
}
}
-27
View File
@@ -102,30 +102,3 @@ func TestCanceledAskContextSkipsToolExecution(t *testing.T) {
t.Fatalf("plan persisted after canceled tool context: %q", plan)
}
}
func TestToolCallTimeoutPropagatesDeadlineToCustomTool(t *testing.T) {
var sawDeadline bool
a := newTestAgent(
Name("tool-timeout"),
ToolCallTimeout(10*time.Millisecond),
WithTool("slow", "slow tool", nil, func(ctx context.Context, input map[string]any) (string, error) {
if _, ok := ctx.Deadline(); ok {
sawDeadline = true
}
<-ctx.Done()
return "", ctx.Err()
}),
)
start := time.Now()
content := toolContent(a.toolHandler(), "slow", nil)
if !sawDeadline {
t.Fatal("custom tool did not receive a deadline")
}
if !strings.Contains(content, context.DeadlineExceeded.Error()) {
t.Fatalf("tool result = %q, want deadline exceeded", content)
}
if elapsed := time.Since(start); elapsed > 200*time.Millisecond {
t.Fatalf("tool call took %s, want bounded timeout", elapsed)
}
}
-3
View File
@@ -153,9 +153,6 @@ func (a *agentImpl) resumeWithStreamEvents(ctx context.Context, runID string, ev
}
return &resp, nil
}
if terminalAgentRunStatus(run.Status) {
return nil, errors.New("agent: checkpointed run is terminal with status " + run.Status)
}
a.mu.Lock()
defer a.mu.Unlock()
-203
View File
@@ -1,203 +0,0 @@
package ai_test
import (
"context"
"encoding/json"
"errors"
"io"
"net/http"
"net/http/httptest"
"reflect"
"strings"
"testing"
"time"
"go-micro.dev/v6/ai"
_ "go-micro.dev/v6/ai/anthropic"
_ "go-micro.dev/v6/ai/atlascloud"
_ "go-micro.dev/v6/ai/gemini"
_ "go-micro.dev/v6/ai/groq"
_ "go-micro.dev/v6/ai/mistral"
_ "go-micro.dev/v6/ai/openai"
_ "go-micro.dev/v6/ai/together"
)
func TestStreamProvidersConformToOpenAICompatibleSSE(t *testing.T) {
providers := conformingStreamProviders(t)
for _, provider := range providers {
provider := provider
t.Run(provider, func(t *testing.T) {
var sawRequest bool
ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
sawRequest = true
if r.URL.Path != "/v1/chat/completions" {
t.Fatalf("path = %s, want /v1/chat/completions", r.URL.Path)
}
if got := r.Header.Get("Accept"); got != "text/event-stream" {
t.Fatalf("Accept = %q, want text/event-stream", got)
}
if got := r.Header.Get("Authorization"); got != "Bearer test-key" {
t.Fatalf("Authorization = %q, want bearer API key", got)
}
var body map[string]any
if err := json.NewDecoder(r.Body).Decode(&body); err != nil {
t.Fatalf("decode request: %v", err)
}
if body["model"] == "" {
t.Fatal("request omitted model")
}
if body["stream"] != true {
t.Fatalf("stream = %#v, want true", body["stream"])
}
streamOptions, ok := body["stream_options"].(map[string]any)
if !ok || streamOptions["include_usage"] != true {
t.Fatalf("stream_options = %#v, want include_usage=true", body["stream_options"])
}
messages, ok := body["messages"].([]any)
if !ok || len(messages) != 4 {
t.Fatalf("messages = %#v, want system + history + prompt", body["messages"])
}
wantRoles := []string{"system", "user", "assistant", "user"}
for i, wantRole := range wantRoles {
message, ok := messages[i].(map[string]any)
if !ok || message["role"] != wantRole {
t.Fatalf("message[%d] = %#v, want role %q", i, messages[i], wantRole)
}
}
w.Header().Set("Content-Type", "text/event-stream")
_, _ = w.Write([]byte(": keepalive\n\n"))
_, _ = w.Write([]byte("event: ignored\n\n"))
_, _ = w.Write([]byte("data: {\"choices\":[{\"delta\":{\"content\":\"hel\"}}]}\n\n"))
_, _ = w.Write([]byte("data: {\"choices\":[{\"delta\":{\"content\":\"lo\"}}]}\n\n"))
_, _ = w.Write([]byte("data: {\"choices\":[],\"usage\":{\"prompt_tokens\":3,\"completion_tokens\":2,\"total_tokens\":5}}\n\n"))
_, _ = w.Write([]byte("data: [DONE]\n\n"))
}))
defer ts.Close()
model := ai.New(provider, ai.WithAPIKey("test-key"), ai.WithBaseURL(ts.URL))
if model == nil {
t.Fatalf("ai.New(%q) returned nil", provider)
}
stream, err := model.Stream(context.Background(), &ai.Request{
SystemPrompt: "system",
Messages: []ai.Message{
{Role: "user", Content: "previous question"},
{Role: "assistant", Content: "previous answer"},
},
Prompt: "current question",
})
if err != nil {
t.Fatalf("Stream returned error: %v", err)
}
defer stream.Close()
if !sawRequest {
t.Fatal("server did not receive stream request")
}
assertStreamReply(t, stream, "hel")
assertStreamReply(t, stream, "lo")
usage, err := stream.Recv()
if err != nil {
t.Fatalf("usage chunk error: %v", err)
}
if usage.Reply != "" || usage.Usage != (ai.Usage{InputTokens: 3, OutputTokens: 2, TotalTokens: 5}) {
t.Fatalf("usage chunk = %#v", usage)
}
if _, err := stream.Recv(); !errors.Is(err, io.EOF) {
t.Fatalf("final error = %v, want EOF", err)
}
})
}
}
func TestStreamProvidersCloseCancelsInFlightRequest(t *testing.T) {
for _, provider := range conformingStreamProviders(t) {
provider := provider
t.Run(provider, func(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()
stream, err := ai.New(provider, ai.WithAPIKey("test-key"), ai.WithBaseURL(ts.URL)).Stream(context.Background(), &ai.Request{Prompt: "Hello"})
if err != nil {
t.Fatalf("Stream returned error: %v", err)
}
assertStreamReply(t, stream, "hel")
if err := stream.Close(); err != nil {
t.Fatalf("Close returned error: %v", err)
}
if err := stream.Close(); err != nil {
t.Fatalf("second Close returned error: %v", err)
}
select {
case <-released:
case <-time.After(time.Second):
t.Fatal("server did not observe canceled stream request")
}
})
}
}
func TestUnsupportedProvidersReturnStreamingUnsupportedAndStayUnregistered(t *testing.T) {
for _, provider := range []string{"anthropic", "gemini"} {
provider := provider
t.Run(provider, func(t *testing.T) {
if caps := ai.ProviderCapabilities(provider); caps.Stream {
t.Fatalf("ProviderCapabilities(%q).Stream = true, want false", provider)
}
_, err := ai.New(provider, ai.WithAPIKey("test-key")).Stream(context.Background(), &ai.Request{Prompt: "Hello"})
if !errors.Is(err, ai.ErrStreamingUnsupported) {
t.Fatalf("Stream error = %v, want ErrStreamingUnsupported", err)
}
if err != nil && strings.Contains(err.Error(), "test-key") {
t.Fatal("streaming unsupported error leaked API key")
}
})
}
}
func conformingStreamProviders(t *testing.T) []string {
t.Helper()
providers := ai.RegisteredProviders("stream")
allowed := map[string]struct{}{
"atlascloud": {},
"groq": {},
"mistral": {},
"openai": {},
"together": {},
}
var out []string
for _, provider := range providers {
if _, ok := allowed[provider]; ok {
out = append(out, provider)
}
}
want := []string{"atlascloud", "groq", "mistral", "openai", "together"}
if !reflect.DeepEqual(out, want) {
t.Fatalf("conforming stream providers = %#v, want %#v (registered stream providers: %#v)", out, want, providers)
}
return out
}
func assertStreamReply(t *testing.T, stream ai.Stream, want string) {
t.Helper()
chunk, err := stream.Recv()
if err != nil {
t.Fatalf("Recv error = %v, want reply %q", err, want)
}
if chunk.Reply != want {
t.Fatalf("Reply = %q, want %q", chunk.Reply, want)
}
}
+35 -86
View File
@@ -43,9 +43,6 @@ func Deploy(c *cli.Context) error {
}
target, remotePath := resolveDeployTarget(c, target, cfg)
if c.Bool("dry-run") {
return printDeployPlan(c, target, cfg, remotePath)
}
return deploySSH(c, target, cfg, remotePath)
}
@@ -99,81 +96,6 @@ func showDeployTargets(cfg *config.Config) error {
return fmt.Errorf("%s", sb.String())
}
func printDeployPlan(c *cli.Context, target string, cfg *config.Config, remotePath string) error {
dir := c.Args().Get(1)
if dir == "" {
dir = "."
}
absDir, err := filepath.Abs(dir)
if err != nil {
return fmt.Errorf("failed to get absolute path: %w", err)
}
if cfg == nil {
cfg, _ = config.Load(absDir)
}
if remotePath == "" {
remotePath = defaultRemotePath
}
services, err := deployServices(absDir, cfg, c.String("service"))
if err != nil {
return err
}
fmt.Println()
fmt.Println(" \033[1mmicro deploy --dry-run\033[0m")
fmt.Println()
fmt.Printf(" Target \033[36m%s\033[0m\n", target)
fmt.Printf(" Remote path %s\n", remotePath)
fmt.Printf(" Services %s\n", strings.Join(services, ", "))
fmt.Println()
fmt.Println(" Plan:")
fmt.Println(" 1. Build linux/amd64 service binaries")
fmt.Printf(" 2. Copy binaries to %s/bin/\n", remotePath)
fmt.Println(" 3. Enable and restart micro@<service> systemd units")
fmt.Println(" 4. Check service health")
fmt.Println()
fmt.Println(" No SSH, rsync, systemd, or remote deployment was performed.")
return nil
}
func deployServices(absDir string, cfg *config.Config, filterService string) ([]string, error) {
if filterService != "" && cfg != nil {
found := false
for _, svc := range cfg.Services {
if svc.Name == filterService {
found = true
break
}
}
if !found && len(cfg.Services) > 0 {
return nil, fmt.Errorf("service '%s' not found in configuration", filterService)
}
}
if cfg != nil && len(cfg.Services) > 0 {
sorted, err := cfg.TopologicalSort()
if err != nil {
return nil, err
}
services := make([]string, 0, len(sorted))
for _, svc := range sorted {
if filterService == "" || svc.Name == filterService {
services = append(services, svc.Name)
}
}
return services, nil
}
services := []string{filepath.Base(absDir)}
if filterService != "" && filterService != services[0] {
return nil, fmt.Errorf("service '%s' not found (only '%s' available)", filterService, services[0])
}
return services, nil
}
func deploySSH(c *cli.Context, target string, cfg *config.Config, remotePath string) error {
dir := c.Args().Get(1)
if dir == "" {
@@ -199,10 +121,19 @@ func deploySSH(c *cli.Context, target string, cfg *config.Config, remotePath str
fmt.Println()
fmt.Printf(" Target \033[36m%s\033[0m\n\n", target)
// Early validation: resolve services before SSH checks.
services, err := deployServices(absDir, cfg, c.String("service"))
if err != nil {
return err
// Early validation: Check if the requested service exists before SSH checks
filterService := c.String("service")
if filterService != "" && cfg != nil {
found := false
for _, svc := range cfg.Services {
if svc.Name == filterService {
found = true
break
}
}
if !found && len(cfg.Services) > 0 {
return fmt.Errorf("service '%s' not found in configuration", filterService)
}
}
// Step 1: Check SSH connectivity
@@ -222,6 +153,28 @@ func deploySSH(c *cli.Context, target string, cfg *config.Config, remotePath str
fmt.Println("\u2713")
// Step 3: Build binaries
var services []string
if cfg != nil && len(cfg.Services) > 0 {
sorted, err := cfg.TopologicalSort()
if err != nil {
return err
}
for _, svc := range sorted {
// If --service flag is provided, only include that service
if filterService == "" || svc.Name == filterService {
services = append(services, svc.Name)
}
}
} else {
// Single service project
services = []string{filepath.Base(absDir)}
// If --service flag was provided for a single-service project, validate it matches
if filterService != "" && filterService != services[0] {
return fmt.Errorf("service '%s' not found (only '%s' available)", filterService, services[0])
}
}
fmt.Printf(" Building binaries... ")
if err := buildBinaries(absDir, cfg, c.Bool("build"), services); err != nil {
fmt.Println("\u2717")
@@ -540,10 +493,6 @@ The deploy process:
Name: "service",
Usage: "Deploy only a specific service (for multi-service projects)",
},
&cli.BoolFlag{
Name: "dry-run",
Usage: "Print the deployment plan without building, connecting, copying, or restarting services",
},
},
})
}
-47
View File
@@ -17,7 +17,6 @@ func newDeployTestContext(t *testing.T, args ...string) *cli.Context {
set.String("ssh", "", "")
set.String("service", "", "")
set.Bool("build", false, "")
set.Bool("dry-run", false, "")
if err := set.Parse(args); err != nil {
t.Fatalf("parse flags: %v", err)
}
@@ -119,49 +118,3 @@ deploy prod
t.Fatalf("deploy target = %#v", prod)
}
}
func TestDeployDryRunPlansConfiguredTargetWithoutRemoteSideEffects(t *testing.T) {
dir := t.TempDir()
if err := os.WriteFile(dir+"/micro.mu", []byte(`service api
path ./api
deploy prod
ssh deploy@prod.example.com
path /srv/micro
`), 0644); err != nil {
t.Fatalf("write config: %v", err)
}
oldwd, err := os.Getwd()
if err != nil {
t.Fatalf("getwd: %v", err)
}
if err := os.Chdir(dir); err != nil {
t.Fatalf("chdir: %v", err)
}
t.Cleanup(func() {
if err := os.Chdir(oldwd); err != nil {
t.Errorf("restore cwd: %v", err)
}
})
ctx := newDeployTestContext(t, "--dry-run", "prod")
if err := Deploy(ctx); err != nil {
t.Fatalf("dry-run deploy: %v", err)
}
}
func TestDeployDryRunValidatesRequestedService(t *testing.T) {
ctx := newDeployTestContext(t, "--dry-run", "--service", "missing", "prod")
cfg := &config.Config{Services: map[string]*config.Service{
"api": {Name: "api", Path: "./api"},
}}
err := printDeployPlan(ctx, "deploy@prod.example.com", cfg, defaultRemotePath)
if err == nil {
t.Fatal("expected dry-run to validate service names")
}
if !strings.Contains(err.Error(), "service 'missing' not found in configuration") {
t.Fatalf("unexpected error: %v", err)
}
}
+2 -5
View File
@@ -498,13 +498,10 @@ func printBanner(services []*serviceProcess, gw *server.Gateway, watching bool,
fmt.Printf(" Dashboard \033[36mhttp://localhost%s\033[0m\n", gw.Addr())
fmt.Printf(" API \033[36mhttp://localhost%s/api/{service}/{method}\033[0m\n", gw.Addr())
fmt.Printf(" Agent \033[36mhttp://localhost%s/agent\033[0m\n", gw.Addr())
// MCP tools are served on the gateway by default — every endpoint is an
// AI-callable tool, so surface it rather than hiding it behind a flag.
fmt.Printf(" MCP Tools \033[36mhttp://localhost%s/mcp/tools\033[0m\n", gw.Addr())
fmt.Printf(" Health \033[36mhttp://localhost%s/health\033[0m\n", gw.Addr())
if mcpAddr != "" {
// Optional standalone MCP protocol server (e.g. for MCP clients).
fmt.Printf(" MCP Server \033[36mhttp://localhost%s\033[0m (full MCP protocol)\n", mcpAddr)
fmt.Printf(" MCP \033[36mhttp://localhost%s\033[0m\n", mcpAddr)
fmt.Printf(" MCP Tools \033[36mhttp://localhost%s/mcp/tools\033[0m\n", mcpAddr)
fmt.Printf(" WebSocket \033[36mws://localhost%s/mcp/ws\033[0m\n", mcpAddr)
}
}
+1 -18
View File
@@ -19,7 +19,7 @@ func TestZeroToHeroCLIBoundaries(t *testing.T) {
}
}
for _, want := range []string{"run", "chat", "flow", "inspect", "deploy"} {
for _, want := range []string{"run", "chat", "flow", "inspect"} {
if !commands[want] {
t.Fatalf("missing %q command", want)
}
@@ -30,21 +30,4 @@ func TestZeroToHeroCLIBoundaries(t *testing.T) {
if !subcommands["inspect"]["agent"] || !subcommands["inspect"]["flow"] {
t.Fatal("missing inspect boundary: inspect agent/flow")
}
var hasDeployDryRun bool
for _, command := range microcmd.DefaultCmd.App().Commands {
if command.Name != "deploy" {
continue
}
for _, flag := range command.Flags {
for _, name := range flag.Names() {
if name == "dry-run" {
hasDeployDryRun = true
}
}
}
}
if !hasDeployDryRun {
t.Fatal("missing deploy boundary: deploy --dry-run")
}
}
+4 -5
View File
@@ -80,11 +80,10 @@ A workflow as ordered, checkpointed steps that survives a crash and resumes wher
- **Checkpoint** — each step is persisted; on `Resume`, completed steps are not re-run (no duplicate side effects)
### [support](./support/)
A maintained 0-to-hero reference path in one runnable file:
- **scaffold** typed `customers`, `tickets`, and `notify` services
- **run/chat** with a support agent that uses those services as tools
- **inspect** the event-driven `intake` flow and approval gate
- **CI** keeps the deterministic mock-model journey runnable with `go test ./examples/support`
A real-world support desk — the "zero to hero" shape in one runnable file:
- **services** (`customers`, `tickets`, `notify`) become the agent's tools automatically
- **flow** turns a `ticket.created` event into work for the agent (the event is the prompt)
- **guardrail** — the agent triages freely but can't email a customer without passing the approval gate
## Coming Soon
-45
View File
@@ -1,45 +0,0 @@
# Durable agent run resume
This example shows the agent-side counterpart to `examples/flow-durable`: an
agent run is checkpointed with the same `Checkpoint` interface used by flows,
then resumed after an interruption without repeating a completed side effect.
The sample uses an in-memory store to keep repeated local runs deterministic;
use your service store for process-restart recovery.
Run it with:
```sh
go run ./examples/agent-durable
```
The demo model calls `inventory.reserve`, then fails to mimic a process dying
after the tool call was checkpointed. `micro.AgentPending` finds the unfinished
run and `micro.AgentResume` continues it from the saved checkpoint. The final
`tool executions: 1` line is the important bit: the reservation tool was not
called a second time during resume.
## When to use this instead of a durable flow
Use a durable flow when the path is known ahead of time: ordered service calls,
retries, timers, compensation, and a precise resume stage such as `reserve` or
`charge`. Use a checkpointed agent run when the path is open-ended and the model
may choose tools dynamically, but completed tool side effects still must not be
replayed after a crash or provider failure.
They compose: keep deterministic business process in `flow-durable`, then hand
off the judgment-heavy step to a checkpointed agent when the workflow needs
model-directed tool use. Both use the same `Checkpoint` backend, so inspection
and recovery can share one run-history store.
In a service, use the same pattern at startup:
```go
pending, _ := micro.AgentPending(ctx, agent)
for _, run := range pending {
_, _ = micro.AgentResume(ctx, agent, run.ID)
}
```
`context.Context` cancellation and deadlines are still honored by checkpoint
loads/saves, model calls, and tool calls. Runs with terminal statuses such as
`done`, `canceled`, and `expired` are not returned by `AgentPending`.
-88
View File
@@ -1,88 +0,0 @@
// Package main demonstrates durable agent runs: a checkpointed agent can
// resume after a crash without re-executing completed tool calls.
package main
import (
"context"
"errors"
"fmt"
"sync/atomic"
micro "go-micro.dev/v6"
"go-micro.dev/v6/ai"
"go-micro.dev/v6/store"
)
func main() {
ctx := context.Background()
checkpoint := micro.StoreCheckpoint(store.NewMemoryStore(), "durable-agent-demo")
model := &demoModel{failFirst: true}
ai.Register("durable-demo", func(opts ...ai.Option) ai.Model {
_ = model.Init(opts...)
return model
})
var reservations atomic.Int32
ag := micro.NewAgent("durable-agent-demo",
micro.AgentWithCheckpoint(checkpoint),
micro.AgentProvider("durable-demo"),
micro.AgentTool("inventory.reserve", "reserve inventory exactly once", map[string]any{
"sku": map[string]any{"type": "string"},
}, func(ctx context.Context, input map[string]any) (string, error) {
count := reservations.Add(1)
return fmt.Sprintf("reserved %s (execution %d)", input["sku"], count), nil
}),
)
_, err := ag.Ask(ctx, "reserve sku-123 and confirm")
fmt.Println("initial run:", err)
pending, err := micro.AgentPending(ctx, ag)
if err != nil {
panic(err)
}
if len(pending) == 0 {
panic("expected a checkpointed run to resume")
}
resp, err := micro.AgentResume(ctx, ag, pending[0].ID)
if err != nil {
panic(err)
}
fmt.Println("resumed reply:", resp.Reply)
fmt.Println("tool executions:", reservations.Load())
}
type demoModel struct {
failFirst bool
opts ai.Options
}
func (m *demoModel) Init(opts ...ai.Option) error {
m.opts = ai.NewOptions(opts...)
return nil
}
func (m *demoModel) Options() ai.Options { return m.opts }
func (m *demoModel) String() string { return "durable-demo" }
func (m *demoModel) Generate(ctx context.Context, req *ai.Request, opts ...ai.GenerateOption) (*ai.Response, error) {
if m.opts.ToolHandler != nil {
res := m.opts.ToolHandler(ctx, ai.ToolCall{
ID: "reserve-1",
Name: "inventory.reserve",
Input: map[string]any{"sku": "sku-123"},
})
if res.Content == "" {
return nil, errors.New("reservation tool returned no content")
}
}
if m.failFirst {
m.failFirst = false
return nil, errors.New("simulated process interruption after checkpointed tool call")
}
return &ai.Response{Reply: "sku-123 is reserved; no duplicate reservation was made"}, nil
}
func (m *demoModel) Stream(context.Context, *ai.Request, ...ai.GenerateOption) (ai.Stream, error) {
return nil, ai.ErrStreamingUnsupported
}
-48
View File
@@ -1,48 +0,0 @@
package main
import (
"bytes"
"io"
"os"
"strings"
"testing"
)
func TestDurableAgentExampleResumesWithoutReplayingTool(t *testing.T) {
out := captureStdout(t, main)
if !strings.Contains(out, "simulated process interruption after checkpointed tool call") {
t.Fatalf("example output %q did not show the initial interrupted run", out)
}
if !strings.Contains(out, "resumed reply: sku-123 is reserved; no duplicate reservation was made") {
t.Fatalf("example output %q did not show the resumed response", out)
}
if !strings.Contains(out, "tool executions: 1") {
t.Fatalf("example output %q did not prove the tool was not replayed", out)
}
}
func captureStdout(t *testing.T, fn func()) string {
t.Helper()
old := os.Stdout
r, w, err := os.Pipe()
if err != nil {
t.Fatalf("pipe stdout: %v", err)
}
os.Stdout = w
var buf bytes.Buffer
done := make(chan struct{})
go func() {
_, _ = io.Copy(&buf, r)
close(done)
}()
fn()
_ = w.Close()
os.Stdout = old
<-done
_ = r.Close()
return buf.String()
}
-8
View File
@@ -34,11 +34,3 @@ calls that tool, Go Micro persists the run with status `paused` and stage
`input-required`. Plain `agent.Resume` continues to support completed, failed,
and approval-paused runs; input-required runs are resumed with
`agent.ResumeInput` so the human response is explicit.
## Cancellation and deadlines
`ResumeInput` uses the caller's `context.Context` for checkpoint reads, writes,
and the resumed model/tool turn. If the context is canceled or its deadline
expires before the resume is committed, the call returns the context error and
the checkpointed run remains `paused` at `input-required`; list it with
`agent.Pending` and retry with a fresh context once the operator is ready.
+5 -27
View File
@@ -1,25 +1,9 @@
# Zero-to-hero support desk
# Support desk
A maintained 0-to-hero reference for the Go Micro lifecycle: scaffold a few
typed services, run them in one process, let an agent chat with those services
as tools, then inspect the durable flow that triggered the work. It is one
runnable file and one CI smoke test, so the reference path stays honest as the
framework evolves.
## The path
1. **Scaffold services**`customers`, `tickets`, and `notify` are ordinary
typed Go Micro services. Their request/response structs and method comments
become the tool contract the agent sees.
2. **Run the harness** — the example starts an in-memory registry, broker,
client, store, services, agent, and flow in one process; no external
dependencies or API key are required for the default run.
3. **Chat through an agent** — the `support` agent receives the ticket event as
a prompt and calls service tools to look up the customer, triage the ticket,
and draft a reply.
4. **Inspect the workflow** — the `intake` flow records the event-driven run and
prints the agent result, showing the service → agent → workflow lifecycle as
one runtime.
A real-world agent built the Go Micro way: a few services, an agent that
manages them, an event that triggers it, and a human-in-the-loop gate on the
one action that touches a customer. It's the "zero to hero" shape in one
runnable file.
## The scenario
@@ -60,12 +44,6 @@ agent, which:
go run main.go # mock model — deterministic, no API key
```
The maintained check is the same deterministic path:
```bash
go test ./examples/support
```
Against a live model, the agent reasons about the ticket itself instead of
following the script:
+1 -24
View File
@@ -1,32 +1,9 @@
package main
import (
"os"
"strings"
"testing"
)
import "testing"
func TestRunSupportMockSmoke(t *testing.T) {
if err := runSupport("mock"); err != nil {
t.Fatalf("support example failed: %v", err)
}
}
func TestZeroToHeroReadmeDocumentsLifecycle(t *testing.T) {
b, err := os.ReadFile("README.md")
if err != nil {
t.Fatalf("read README.md: %v", err)
}
doc := string(b)
for _, want := range []string{
"Scaffold services",
"Run the harness",
"Chat through an agent",
"Inspect the workflow",
"go test ./examples/support",
} {
if !strings.Contains(doc, want) {
t.Fatalf("README.md missing zero-to-hero step %q", want)
}
}
}
-212
View File
@@ -1,212 +0,0 @@
package flow
import (
"context"
"encoding/json"
"fmt"
"sort"
"strings"
"time"
"go-micro.dev/v6/ai"
)
// AnalyzeOptions configures Analyze.
type AnalyzeOptions struct {
// MaxFeedbackSamples bounds the number of representative grader feedback
// strings retained per candidate. Values <= 0 use a small default.
MaxFeedbackSamples int
}
// AnalyzeOption configures Analyze.
type AnalyzeOption func(*AnalyzeOptions)
// AnalyzeMaxFeedbackSamples sets how many grader feedback examples are kept for
// each candidate in the report.
func AnalyzeMaxFeedbackSamples(n int) AnalyzeOption {
return func(o *AnalyzeOptions) { o.MaxFeedbackSamples = n }
}
// Report is the machine-readable output of Analyze. Candidates are ordered from
// worst to best so an agent, CLI, or human can pick the first improvement to try.
type Report struct {
Candidates []Candidate `json:"candidates"`
}
// Candidate identifies one underperforming flow step and the trace evidence that
// made it worth improving.
type Candidate struct {
Step string `json:"step"`
Metric string `json:"metric"`
Score float64 `json:"score"`
Runs int `json:"runs"`
Failures int `json:"failures"`
PassRate float64 `json:"pass_rate"`
ErrorRate float64 `json:"error_rate"`
AverageRetries float64 `json:"average_retries"`
P50Latency time.Duration `json:"p50_latency"`
P95Latency time.Duration `json:"p95_latency"`
SampleFeedback []string `json:"sample_feedback,omitempty"`
RunIDs []string `json:"run_ids,omitempty"`
}
// Analyze aggregates a bounded window of persisted flow runs and returns ranked
// hill-climbing candidates. It uses the same Run records read by Checkpoint.List:
// failed verification fields in step results drive pass-rate and feedback, step
// status drives error rate, and retry attempts contribute retry pressure. An
// empty window returns an empty report.
func Analyze(runs []Run, opts ...AnalyzeOption) Report {
o := AnalyzeOptions{MaxFeedbackSamples: 3}
for _, opt := range opts {
opt(&o)
}
if o.MaxFeedbackSamples <= 0 {
o.MaxFeedbackSamples = 3
}
stats := map[string]*stepStats{}
for _, run := range runs {
for _, step := range run.Steps {
if step.Name == "" {
continue
}
s := stats[step.Name]
if s == nil {
s = &stepStats{}
stats[step.Name] = s
}
s.runs++
s.runIDs = appendUnique(s.runIDs, run.ID)
if step.Attempts > 1 {
s.retries += step.Attempts - 1
}
if step.Status == "failed" || step.Error != "" {
s.errors++
}
if len(run.Steps) > 0 && !run.Started.IsZero() && !run.Updated.IsZero() {
s.latencies = append(s.latencies, run.Updated.Sub(run.Started)/time.Duration(len(run.Steps)))
}
passed, feedback, ok := verificationFields(step.Result)
if ok {
s.graded++
if !passed {
s.gradeFailures++
if feedback != "" && len(s.feedback) < o.MaxFeedbackSamples {
s.feedback = append(s.feedback, feedback)
}
}
}
}
}
report := Report{}
for step, s := range stats {
if s.runs == 0 {
continue
}
failures := s.errors + s.gradeFailures
passRate := 1.0
if s.graded > 0 {
passRate = float64(s.graded-s.gradeFailures) / float64(s.graded)
} else if s.errors > 0 {
passRate = float64(s.runs-s.errors) / float64(s.runs)
}
errorRate := float64(s.errors) / float64(s.runs)
avgRetries := float64(s.retries) / float64(s.runs)
score := float64(s.gradeFailures)*3 + float64(s.errors)*2 + avgRetries
metric := "pass_rate"
if s.gradeFailures == 0 && s.errors > 0 {
metric = "error_rate"
} else if s.gradeFailures == 0 && s.errors == 0 && s.retries > 0 {
metric = "retry_count"
}
report.Candidates = append(report.Candidates, Candidate{
Step: step, Metric: metric, Score: score, Runs: s.runs, Failures: failures,
PassRate: passRate, ErrorRate: errorRate, AverageRetries: avgRetries,
P50Latency: percentile(s.latencies, 0.50), P95Latency: percentile(s.latencies, 0.95),
SampleFeedback: append([]string(nil), s.feedback...), RunIDs: append([]string(nil), s.runIDs...),
})
}
sort.SliceStable(report.Candidates, func(i, j int) bool {
a, b := report.Candidates[i], report.Candidates[j]
if a.Score == b.Score {
return a.Step < b.Step
}
return a.Score > b.Score
})
return report
}
type stepStats struct {
runs, graded, gradeFailures, errors, retries int
feedback, runIDs []string
latencies []time.Duration
}
// PromptOptimizer proposes prompt improvements for a candidate without mutating
// the source flow. Applying the returned prompt stays explicitly gated by the caller.
type PromptOptimizer struct{ model ai.Model }
// LLMOptimizer returns an optimizer that asks model to revise prompts for
// Analyze candidates. The model is injected so tests and callers can use mocks.
func LLMOptimizer(model ai.Model) *PromptOptimizer { return &PromptOptimizer{model: model} }
// OptimizePrompt asks the model for a revised prompt for candidate using the
// current prompt and trace feedback. It returns only the proposal; it never
// modifies a Flow, Step, or Checkpoint.
func (o *PromptOptimizer) OptimizePrompt(ctx context.Context, candidate Candidate, currentPrompt string) (string, error) {
if o == nil || o.model == nil {
return "", fmt.Errorf("flow: LLMOptimizer requires a model")
}
prompt := fmt.Sprintf("Revise this workflow step prompt to improve the failing step.\nStep: %s\nMetric: %s\nScore: %.2f\nFeedback:\n- %s\n\nCurrent prompt:\n%s\n\nReturn only the revised prompt.", candidate.Step, candidate.Metric, candidate.Score, strings.Join(candidate.SampleFeedback, "\n- "), currentPrompt)
resp, err := o.model.Generate(ctx, &ai.Request{Prompt: prompt})
if err != nil {
return "", err
}
proposal := strings.TrimSpace(resp.Answer)
if proposal == "" {
proposal = strings.TrimSpace(resp.Reply)
}
if proposal == "" {
return "", fmt.Errorf("flow: LLMOptimizer returned an empty prompt")
}
return proposal, nil
}
func verificationFields(result string) (bool, string, bool) {
if result == "" {
return false, "", false
}
var obj map[string]any
if err := json.Unmarshal([]byte(result), &obj); err != nil {
return false, "", false
}
v, ok := obj["verification_passed"].(bool)
if !ok {
return false, "", false
}
fb, _ := obj["verification_feedback"].(string)
return v, fb, true
}
func appendUnique(values []string, value string) []string {
if value == "" {
return values
}
for _, v := range values {
if v == value {
return values
}
}
return append(values, value)
}
func percentile(values []time.Duration, p float64) time.Duration {
if len(values) == 0 {
return 0
}
sorted := append([]time.Duration(nil), values...)
sort.Slice(sorted, func(i, j int) bool { return sorted[i] < sorted[j] })
idx := int(float64(len(sorted)-1) * p)
return sorted[idx]
}
-89
View File
@@ -1,89 +0,0 @@
package flow
import (
"context"
"strings"
"testing"
"time"
"go-micro.dev/v6/ai"
)
func TestAnalyzeRanksFailedGraderStepAbovePassingStep(t *testing.T) {
now := time.Now()
runs := []Run{
{ID: "run-1", Started: now, Updated: now.Add(time.Second), Steps: []StepRecord{
{Name: "draft", Status: "done", Attempts: 2, Result: `{"verification_passed":false,"verification_feedback":"cite sources"}`},
{Name: "publish", Status: "done", Attempts: 1, Result: `{"verification_passed":true,"verification_feedback":"ok"}`},
}},
{ID: "run-2", Started: now, Updated: now.Add(2 * time.Second), Steps: []StepRecord{
{Name: "draft", Status: "done", Attempts: 1, Result: `{"verification_passed":false,"verification_feedback":"too vague"}`},
{Name: "publish", Status: "done", Attempts: 1, Result: `{"verification_passed":true,"verification_feedback":"ok"}`},
}},
}
report := Analyze(runs)
if len(report.Candidates) != 2 {
t.Fatalf("Analyze returned %d candidates, want 2", len(report.Candidates))
}
if got := report.Candidates[0].Step; got != "draft" {
t.Fatalf("top candidate = %q, want draft", got)
}
if report.Candidates[0].PassRate != 0 {
t.Fatalf("draft pass rate = %v, want 0", report.Candidates[0].PassRate)
}
}
func TestAnalyzeCarriesFeedbackSamplesAndRunIDs(t *testing.T) {
report := Analyze([]Run{{ID: "run-9", Steps: []StepRecord{{
Name: "grade", Status: "done", Attempts: 3,
Result: `{"verification_passed":false,"verification_feedback":"include totals"}`,
}}}})
if len(report.Candidates) != 1 {
t.Fatalf("candidates = %d, want 1", len(report.Candidates))
}
c := report.Candidates[0]
if len(c.SampleFeedback) != 1 || c.SampleFeedback[0] != "include totals" {
t.Fatalf("feedback = %#v, want include totals", c.SampleFeedback)
}
if len(c.RunIDs) != 1 || c.RunIDs[0] != "run-9" {
t.Fatalf("run ids = %#v, want run-9", c.RunIDs)
}
if c.AverageRetries != 2 {
t.Fatalf("average retries = %v, want 2", c.AverageRetries)
}
}
func TestAnalyzeEmptyWindowReturnsEmptyReport(t *testing.T) {
if got := Analyze(nil); len(got.Candidates) != 0 {
t.Fatalf("empty Analyze candidates = %d, want 0", len(got.Candidates))
}
}
func TestLLMOptimizerReturnsProposalWithoutMutatingFlow(t *testing.T) {
f := New("optimize", Prompt("original prompt"))
before := f.opts.Prompt
optimizer := LLMOptimizer(&optimizerModel{reply: "revised prompt"})
proposal, err := optimizer.OptimizePrompt(context.Background(), Candidate{Step: "draft", Metric: "pass_rate", SampleFeedback: []string{"cite sources"}}, before)
if err != nil {
t.Fatalf("OptimizePrompt returned error: %v", err)
}
if !strings.Contains(proposal, "revised") {
t.Fatalf("proposal = %q, want revised prompt", proposal)
}
if f.opts.Prompt != before {
t.Fatalf("flow prompt mutated to %q, want %q", f.opts.Prompt, before)
}
}
type optimizerModel struct{ reply string }
func (m *optimizerModel) Init(...ai.Option) error { return nil }
func (m *optimizerModel) Options() ai.Options { return ai.Options{} }
func (m *optimizerModel) Generate(context.Context, *ai.Request, ...ai.GenerateOption) (*ai.Response, error) {
return &ai.Response{Reply: m.reply}, nil
}
func (m *optimizerModel) Stream(context.Context, *ai.Request, ...ai.GenerateOption) (ai.Stream, error) {
return nil, ai.ErrStreamingUnsupported
}
func (m *optimizerModel) String() string { return "optimizer" }
-45
View File
@@ -1,45 +0,0 @@
package flow_test
import (
"context"
"fmt"
"strings"
"go-micro.dev/v6/flow"
)
func ExampleVerify() {
generate := func(_ context.Context, in flow.State) (flow.State, error) {
if strings.Contains(in.String(), "feedback") {
in.Data = []byte(`{"answer":"include a source"}`)
return in, nil
}
in.Data = []byte(`{"answer":"draft"}`)
return in, nil
}
grader := func(_ context.Context, out flow.State) (bool, string, error) {
return strings.Contains(out.String(), "source"), "add a source", nil
}
out, _ := flow.Verify(generate, grader, flow.VerifyMaxAttempts(2))(context.Background(), flow.State{})
fmt.Println(strings.Contains(out.String(), `"verification_passed":true`))
// Output: true
}
func ExampleAnalyze() {
runs := []flow.Run{{
ID: "run-1",
Steps: []flow.StepRecord{{
Name: "draft",
Status: "done",
Result: `{"verification_passed":false,"verification_feedback":"add a source"}`,
}},
}}
report := flow.Analyze(runs)
fmt.Println(report.Candidates[0].Step)
fmt.Println(report.Candidates[0].SampleFeedback[0])
// Output:
// draft
// add a source
}
-191
View File
@@ -1,191 +0,0 @@
package flow
import (
"context"
"encoding/json"
"fmt"
"strings"
"time"
"go-micro.dev/v6/ai"
)
// Grader checks a step output against a rubric. It returns pass=true when the
// output is acceptable; otherwise feedback should explain what the next attempt
// should fix.
type Grader func(ctx context.Context, out State) (pass bool, feedback string, err error)
// VerifyOptions configure Verify.
type VerifyOptions struct {
// MaxAttempts bounds how many times the body can run. Default 2.
MaxAttempts int
// Backoff waits between failed grades. Zero means retry immediately.
Backoff time.Duration
// FeedbackField is the JSON field used to thread grader feedback into the
// next attempt's input. Default "feedback".
FeedbackField string
}
// VerifyOption configures Verify.
type VerifyOption func(*VerifyOptions)
// VerifyMaxAttempts sets the total attempt budget for Verify. Values <= 0 use
// the default of 2.
func VerifyMaxAttempts(n int) VerifyOption { return func(o *VerifyOptions) { o.MaxAttempts = n } }
// VerifyBackoff sets the delay between failed verification attempts.
func VerifyBackoff(d time.Duration) VerifyOption { return func(o *VerifyOptions) { o.Backoff = d } }
// VerifyFeedbackField sets the JSON field used to pass grader feedback to the
// next body attempt. Empty values use "feedback".
func VerifyFeedbackField(field string) VerifyOption {
return func(o *VerifyOptions) { o.FeedbackField = field }
}
// Verify runs body, grades its output, and retries with grader feedback threaded
// into the next input until the grader passes or MaxAttempts is exhausted. It is
// a StepFunc, so it composes directly as Step.Run with Loop, LLM, Call, Agent, or
// any code-defined step.
//
// On a failed grade, Verify adds the feedback to the next attempt's input as a
// JSON field named "feedback" (or VerifyFeedbackField). When all attempts fail,
// it returns the last output without error, annotated with verification fields so
// the run can keep the bounded failure outcome in its state:
// "verification_passed": false, "verification_feedback", and
// "verification_attempts".
func Verify(body StepFunc, grader Grader, opts ...VerifyOption) StepFunc {
o := VerifyOptions{MaxAttempts: 2, FeedbackField: "feedback"}
for _, op := range opts {
op(&o)
}
if o.MaxAttempts <= 0 {
o.MaxAttempts = 2
}
if o.FeedbackField == "" {
o.FeedbackField = "feedback"
}
return func(ctx context.Context, in State) (State, error) {
if body == nil {
return in, fmt.Errorf("flow: Verify requires a body step")
}
if grader == nil {
return in, fmt.Errorf("flow: Verify requires a grader")
}
cur := in
last := in
feedback := ""
for attempt := 1; attempt <= o.MaxAttempts; attempt++ {
if err := ctx.Err(); err != nil {
return last, err
}
if feedback != "" {
var err error
cur, err = stateWithField(cur, o.FeedbackField, feedback)
if err != nil {
return last, err
}
}
out, err := body(ctx, cur)
if err != nil {
return last, fmt.Errorf("verify attempt %d: %w", attempt, err)
}
last = out
pass, fb, err := grader(ctx, out)
if err != nil {
return last, fmt.Errorf("verify grade attempt %d: %w", attempt, err)
}
if pass {
return stateWithVerification(out, true, fb, attempt)
}
feedback = fb
cur = in
if attempt < o.MaxAttempts && o.Backoff > 0 {
select {
case <-time.After(o.Backoff):
case <-ctx.Done():
return last, ctx.Err()
}
}
}
return stateWithVerification(last, false, feedback, o.MaxAttempts)
}
}
// LLMGrader returns a grader that asks the flow model to judge the latest output
// against rubric. The model should answer with pass/fail plus short feedback.
// It reuses the flow's configured model, so it must run inside a flow.
func LLMGrader(rubric string) Grader {
return func(ctx context.Context, out State) (bool, string, error) {
d := depsFrom(ctx)
if d == nil || d.model == nil {
return false, "", fmt.Errorf("flow: LLMGrader requires a flow model (set Provider/APIKey)")
}
prompt := fmt.Sprintf("Grade the latest result against this rubric:\n%s\n\nLatest result:\n%s\n\nAnswer with PASS or FAIL on the first line, followed by one short feedback sentence.", rubric, out.String())
resp, err := d.model.Generate(ctx, &ai.Request{Prompt: prompt})
if err != nil {
return false, "", err
}
reply := resp.Answer
if reply == "" {
reply = resp.Reply
}
return parseGrade(reply)
}
}
func parseGrade(reply string) (bool, string, error) {
text := strings.TrimSpace(reply)
if text == "" {
return false, "", fmt.Errorf("flow: LLMGrader returned an empty grade")
}
lines := strings.SplitN(text, "\n", 2)
first := strings.ToLower(strings.TrimSpace(lines[0]))
feedback := ""
if len(lines) > 1 {
feedback = strings.TrimSpace(lines[1])
}
pass := strings.HasPrefix(first, "pass") || isAffirmative(first)
if !pass && feedback == "" {
feedback = text
}
return pass, feedback, nil
}
func stateWithField(s State, field, value string) (State, error) {
var obj map[string]any
if len(s.Data) > 0 && json.Unmarshal(s.Data, &obj) == nil && obj != nil {
obj[field] = value
return stateWithObject(s, obj)
}
obj = map[string]any{field: value}
if len(s.Data) > 0 {
obj["data"] = s.String()
}
return stateWithObject(s, obj)
}
func stateWithVerification(s State, passed bool, feedback string, attempts int) (State, error) {
var obj map[string]any
if len(s.Data) > 0 && json.Unmarshal(s.Data, &obj) == nil && obj != nil {
obj["verification_passed"] = passed
obj["verification_feedback"] = feedback
obj["verification_attempts"] = attempts
return stateWithObject(s, obj)
}
obj = map[string]any{
"data": s.String(),
"verification_passed": passed,
"verification_feedback": feedback,
"verification_attempts": attempts,
}
return stateWithObject(s, obj)
}
func stateWithObject(s State, obj map[string]any) (State, error) {
b, err := json.Marshal(obj)
if err != nil {
return s, err
}
s.Data = b
return s, nil
}
-96
View File
@@ -1,96 +0,0 @@
package flow
import (
"context"
"strings"
"testing"
)
func TestVerifyPassesFirstTry(t *testing.T) {
attempts := 0
step := Verify(func(_ context.Context, in State) (State, error) {
attempts++
in.Data = []byte(`{"answer":"ok"}`)
return in, nil
}, func(context.Context, State) (bool, string, error) {
return true, "looks good", nil
}, VerifyMaxAttempts(3))
out, err := step(context.Background(), State{})
if err != nil {
t.Fatalf("Verify returned error: %v", err)
}
if attempts != 1 {
t.Fatalf("body attempts = %d, want 1", attempts)
}
var got map[string]any
if err := out.Scan(&got); err != nil {
t.Fatalf("scan output: %v", err)
}
if got["verification_passed"] != true {
t.Fatalf("verification_passed = %v, want true", got["verification_passed"])
}
}
func TestVerifyRetriesWithFeedback(t *testing.T) {
attempts := 0
var secondInput map[string]string
step := Verify(func(_ context.Context, in State) (State, error) {
attempts++
if attempts == 2 {
if err := in.Scan(&secondInput); err != nil {
t.Fatalf("scan second input: %v", err)
}
}
in.Data = []byte(`{"answer":"draft"}`)
return in, nil
}, func(_ context.Context, _ State) (bool, string, error) {
return attempts >= 2, "include citations", nil
}, VerifyMaxAttempts(3))
out, err := step(context.Background(), State{Data: []byte(`{"topic":"agents"}`)})
if err != nil {
t.Fatalf("Verify returned error: %v", err)
}
if attempts != 2 {
t.Fatalf("body attempts = %d, want 2", attempts)
}
if secondInput["feedback"] != "include citations" {
t.Fatalf("feedback = %q, want include citations", secondInput["feedback"])
}
if !strings.Contains(out.String(), `"verification_passed":true`) {
t.Fatalf("output missing successful verification annotation: %s", out.String())
}
}
func TestVerifyExhaustsAttemptsReturnsLastOutput(t *testing.T) {
attempts := 0
step := Verify(func(_ context.Context, in State) (State, error) {
attempts++
in.Data = []byte(`{"answer":"still wrong"}`)
return in, nil
}, func(context.Context, State) (bool, string, error) {
return false, "try again", nil
}, VerifyMaxAttempts(2))
out, err := step(context.Background(), State{})
if err != nil {
t.Fatalf("Verify returned error: %v", err)
}
if attempts != 2 {
t.Fatalf("body attempts = %d, want 2", attempts)
}
var got map[string]any
if err := out.Scan(&got); err != nil {
t.Fatalf("scan output: %v", err)
}
if got["verification_passed"] != false {
t.Fatalf("verification_passed = %v, want false", got["verification_passed"])
}
if got["verification_feedback"] != "try again" {
t.Fatalf("verification_feedback = %v, want try again", got["verification_feedback"])
}
if got["verification_attempts"] != float64(2) {
t.Fatalf("verification_attempts = %v, want 2", got["verification_attempts"])
}
}
+16 -116
View File
@@ -28,7 +28,6 @@ package a2a
import (
"context"
"encoding/json"
"errors"
"fmt"
"io"
"log"
@@ -141,11 +140,8 @@ func (g *Gateway) Handler() http.Handler {
// Per-agent card (served at the agent's url and at its well-known path).
mux.HandleFunc("GET /agents/{name}", g.handleCard)
mux.HandleFunc("GET /agents/{name}/.well-known/agent.json", g.handleCard)
mux.HandleFunc("GET /agents/{name}/skills/{skill}", g.handleSkillCard)
mux.HandleFunc("GET /agents/{name}/skills/{skill}/.well-known/agent.json", g.handleSkillCard)
// Per-agent JSON-RPC endpoint.
mux.HandleFunc("POST /agents/{name}", g.handleRPC)
mux.HandleFunc("POST /agents/{name}/skills/{skill}", g.handleSkillRPC)
// Top-level well-known: serve the single agent's card if there's
// exactly one, otherwise point to the directory.
mux.HandleFunc("GET /.well-known/agent.json", g.handleWellKnown)
@@ -342,17 +338,23 @@ func Card(name, url, description string, services []string) AgentCard {
description = "Go Micro agent"
}
}
skills := skillsFromServices(services)
return AgentCard{
Name: name,
Description: description,
URL: url,
Version: "1.0.0",
ProtocolVersion: protocolVersion,
Capabilities: Capabilities{Streaming: true, PushNotifications: true},
Name: name,
Description: description,
URL: url,
Version: "1.0.0",
ProtocolVersion: protocolVersion,
Capabilities: Capabilities{Streaming: true, PushNotifications: true},
// The agent converses over a single Chat endpoint; advertise that
// as one skill, tagged with the services it manages.
DefaultInputModes: []string{"text/plain"},
DefaultOutputModes: []string{"text/plain"},
Skills: skills,
Skills: []Skill{{
ID: "chat",
Name: "Chat",
Description: "Converse with the agent to operate its services.",
Tags: services,
}},
}
}
@@ -369,19 +371,6 @@ func (g *Gateway) lookupCard(name string) (AgentCard, bool) {
return g.card(name, meta), true
}
func (g *Gateway) lookupSkillCard(name, skillID string) (AgentCard, Skill, bool) {
card, ok := g.lookupCard(name)
if !ok {
return AgentCard{}, Skill{}, false
}
for _, skill := range card.Skills {
if skill.ID == skillID {
return card, skill, true
}
}
return AgentCard{}, Skill{}, false
}
// ---------------------------------------------------------------------------
// HTTP handlers
// ---------------------------------------------------------------------------
@@ -404,17 +393,6 @@ func (g *Gateway) handleCard(w http.ResponseWriter, r *http.Request) {
writeJSON(w, http.StatusOK, card)
}
func (g *Gateway) handleSkillCard(w http.ResponseWriter, r *http.Request) {
card, skill, ok := g.lookupSkillCard(r.PathValue("name"), r.PathValue("skill"))
if !ok {
http.NotFound(w, r)
return
}
card.URL = g.opts.BaseURL + "/agents/" + r.PathValue("name") + "/skills/" + skill.ID
card.Skills = []Skill{skill}
writeJSON(w, http.StatusOK, card)
}
func (g *Gateway) handleWellKnown(w http.ResponseWriter, r *http.Request) {
cards, err := g.cards()
if err != nil {
@@ -443,18 +421,6 @@ func (g *Gateway) handleRPC(w http.ResponseWriter, r *http.Request) {
})
}
func (g *Gateway) handleSkillRPC(w http.ResponseWriter, r *http.Request) {
name := r.PathValue("name")
_, skill, ok := g.lookupSkillCard(name, r.PathValue("skill"))
if !ok {
writeRPC(w, nil, nil, &rpcError{Code: errInvalidParams, Message: "unknown agent skill: " + name + "/" + r.PathValue("skill")})
return
}
g.disp.serve(w, r, func(ctx context.Context, text string) (string, error) {
return g.callAgent(ctx, name, skillPrompt(skill, text))
})
}
// dispatcher handles A2A JSON-RPC requests against an Invoke function and
// retains recent tasks for tasks/get. It is shared by the gateway (one
// per registry) and embedded agents (one per agent).
@@ -490,7 +456,7 @@ func (d *dispatcher) serveWithStream(w http.ResponseWriter, r *http.Request, inv
d.send(requestContext(r.Context()), w, req, invoke)
case "message/stream":
if streamInvoke != nil {
d.streamChunks(requestContext(r.Context()), w, req, streamInvoke, invoke)
d.streamChunks(requestContext(r.Context()), w, req, streamInvoke)
return
}
d.stream(requestContext(r.Context()), w, req, invoke)
@@ -539,7 +505,7 @@ func (d *dispatcher) stream(ctx context.Context, w http.ResponseWriter, req rpcR
}
}
func (d *dispatcher) streamChunks(ctx context.Context, w http.ResponseWriter, req rpcRequest, invoke StreamInvoke, fallback Invoke) {
func (d *dispatcher) streamChunks(ctx context.Context, w http.ResponseWriter, req rpcRequest, invoke StreamInvoke) {
var p sendParams
if err := json.Unmarshal(req.Params, &p); err != nil {
writeRPC(w, req.ID, nil, &rpcError{Code: errInvalidParams, Message: "invalid params"})
@@ -552,10 +518,6 @@ func (d *dispatcher) streamChunks(ctx context.Context, w http.ResponseWriter, re
}
stream, err := invoke(ctx, text)
if err != nil {
if errors.Is(err, ai.ErrStreamingUnsupported) && fallback != nil {
d.stream(ctx, w, req, fallback)
return
}
writeRPC(w, req.ID, nil, &rpcError{Code: errInternal, Message: err.Error()})
return
}
@@ -891,68 +853,6 @@ func (d *dispatcher) deliverPush(taskID string, task *Task) {
}
}
func skillsFromServices(services []string) []Skill {
if len(services) == 0 {
return []Skill{{ID: "chat", Name: "Chat", Description: "Converse with the agent to operate its services."}}
}
seen := map[string]bool{}
var skills []Skill
for _, service := range services {
service = strings.TrimSpace(service)
if service == "" {
continue
}
id := skillID(service)
if id == "" || seen[id] {
continue
}
seen[id] = true
skills = append(skills, Skill{
ID: id,
Name: skillName(service),
Description: fmt.Sprintf("Operate the %s service through this agent.", service),
Tags: []string{service},
})
}
if len(skills) == 0 {
return []Skill{{ID: "chat", Name: "Chat", Description: "Converse with the agent to operate its services."}}
}
return skills
}
func skillID(service string) string {
service = strings.ToLower(strings.TrimSpace(service))
var b strings.Builder
dash := false
for _, r := range service {
if (r >= 'a' && r <= 'z') || (r >= '0' && r <= '9') {
b.WriteRune(r)
dash = false
continue
}
if !dash && b.Len() > 0 {
b.WriteByte('-')
dash = true
}
}
return strings.Trim(b.String(), "-")
}
func skillName(service string) string {
parts := strings.FieldsFunc(service, func(r rune) bool { return r == '-' || r == '_' || r == '.' || r == '/' || r == ' ' })
for i, part := range parts {
if part == "" {
continue
}
parts[i] = strings.ToUpper(part[:1]) + part[1:]
}
return strings.Join(parts, " ")
}
func skillPrompt(skill Skill, text string) string {
return fmt.Sprintf("Use the %q skill (%s) for this request.\n\n%s", skill.Name, skill.ID, text)
}
func textOf(parts []Part) string {
var b strings.Builder
for _, p := range parts {
+3 -98
View File
@@ -51,7 +51,7 @@ func newGatewayWithAgent(t *testing.T) (*httptest.Server, func()) {
server.Name("echo"),
server.Address("127.0.0.1:0"),
server.Registry(reg),
server.Metadata(map[string]string{"type": "agent", "services": "task,project"}),
server.Metadata(map[string]string{"type": "agent", "services": ""}),
)
if err := pb.RegisterAgentHandler(srv, echoAgent{}); err != nil {
t.Fatalf("register agent handler: %v", err)
@@ -88,40 +88,8 @@ func TestAgentCardFromRegistry(t *testing.T) {
if card.URL != "http://gw/agents/echo" {
t.Errorf("card url = %q", card.URL)
}
if card.ProtocolVersion == "" {
t.Errorf("card missing protocolVersion: %+v", card)
}
if got := skillIDs(card.Skills); strings.Join(got, ",") != "task,project" {
t.Errorf("skill IDs = %v, want [task project]", got)
}
}
func TestSkillEndpointServesFocusedCardAndRoutesRPC(t *testing.T) {
ts, cleanup := newGatewayWithAgent(t)
defer cleanup()
resp, err := http.Get(ts.URL + "/agents/echo/skills/task/.well-known/agent.json")
if err != nil {
t.Fatalf("get skill card: %v", err)
}
defer resp.Body.Close()
if resp.StatusCode != http.StatusOK {
t.Fatalf("skill card status = %d", resp.StatusCode)
}
var card AgentCard
if err := json.NewDecoder(resp.Body).Decode(&card); err != nil {
t.Fatalf("decode skill card: %v", err)
}
if card.URL != "http://gw/agents/echo/skills/task" || len(card.Skills) != 1 || card.Skills[0].ID != "task" {
t.Fatalf("skill card = %+v, want task-only card at skill URL", card)
}
task := rpcTask(t, ts.URL+"/agents/echo/skills/task", `{
"jsonrpc":"2.0","id":1,"method":"message/send",
"params":{"message":{"role":"user","kind":"message","messageId":"m1",
"parts":[{"kind":"text","text":"ping"}]}}}`)
if task.Status.State != stateCompleted || textOf(task.Artifacts[0].Parts) != "pong" {
t.Fatalf("skill task = %+v, want completed pong", task)
if card.ProtocolVersion == "" || len(card.Skills) == 0 {
t.Errorf("card missing protocolVersion or skills: %+v", card)
}
}
@@ -465,61 +433,6 @@ func TestMessageStreamChunksPropagatesCancellationAndClosesStream(t *testing.T)
}
}
func TestMessageStreamChunksFallsBackWhenUnsupported(t *testing.T) {
d := newDispatcher()
body := `{"jsonrpc":"2.0","id":1,"method":"message/stream","params":{"message":{"role":"user","parts":[{"kind":"text","text":"ping"}],"kind":"message"}}}`
req := httptest.NewRequest(http.MethodPost, "/", bytes.NewBufferString(body))
rr := httptest.NewRecorder()
var streamed bool
var fallbackText string
d.serveWithStream(rr, req, func(ctx context.Context, text string) (string, error) {
fallbackText = text
return "pong", nil
}, func(ctx context.Context, text string) (ai.Stream, error) {
streamed = true
return nil, fmt.Errorf("%w: test provider", ai.ErrStreamingUnsupported)
})
if !streamed {
t.Fatal("stream invoke was not attempted")
}
if fallbackText != "ping" {
t.Fatalf("fallback text = %q, want ping", fallbackText)
}
if ct := rr.Result().Header.Get("Content-Type"); !strings.HasPrefix(ct, "text/event-stream") {
t.Fatalf("content-type = %q, want text/event-stream", ct)
}
var events []struct {
Result Task `json:"result"`
Error *rpcError `json:"error"`
}
for _, line := range strings.Split(strings.TrimSpace(rr.Body.String()), "\n") {
line = strings.TrimSpace(line)
if line == "" {
continue
}
line = strings.TrimPrefix(line, "data: ")
var event struct {
Result Task `json:"result"`
Error *rpcError `json:"error"`
}
if err := json.Unmarshal([]byte(line), &event); err != nil {
t.Fatalf("decode event %q: %v", line, err)
}
events = append(events, event)
}
if len(events) != 1 {
t.Fatalf("events = %d, want 1; body %s", len(events), rr.Body.String())
}
if events[0].Error != nil {
t.Fatalf("fallback event error: %+v", events[0].Error)
}
if events[0].Result.Status.State != stateCompleted || textOf(events[0].Result.Artifacts[0].Parts) != "pong" {
t.Fatalf("fallback task = %+v, want completed pong", events[0].Result)
}
}
func TestTasksResubscribeStreamsCurrentAndSubsequentEvents(t *testing.T) {
d := newDispatcher()
initial := &Task{ID: "task-1", ContextID: "ctx-1", Kind: "task", Status: TaskStatus{State: stateWorking, Timestamp: time.Now().UTC().Format(time.RFC3339)}}
@@ -731,11 +644,3 @@ func rpcTask(t *testing.T, url, body string) Task {
}
return resp.Result
}
func skillIDs(skills []Skill) []string {
ids := make([]string, 0, len(skills))
for _, skill := range skills {
ids = append(ids, skill.ID)
}
return ids
}
+3 -5
View File
@@ -21,11 +21,9 @@ changes, architectural rewrites. Those go to the human.
## Work queue (ranked)
1. **Add scheduled cross-provider 0→hero conformance** ([#3454](https://github.com/micro/go-micro/issues/3454)) — the previous top Now/Next items shipped: verification/grader loops landed in #3443, run-trace optimization analysis landed in #3447, and durable agent resume was completed by #3452, closing #3449. With the core loop primitives now in place, the highest-value Now-phase gap is trust: the same scaffold → run → chat/tool → workflow path must stay true across supported providers on a schedule, with keyed runs gated and no-secret CI still useful. This keeps the services → agents → workflows lifecycle cohesive instead of letting provider behavior drift behind green unit tests.
2. **Emit OpenTelemetry spans for agent RunInfo timelines** ([#3455](https://github.com/micro/go-micro/issues/3455)) — flows now have trace-oriented optimization feedback, and durable agent runs can resume, but the agent side still needs first-class operability in production traces. Translating `RunInfo` / run timeline events into spans closes a Next-phase observability seam across agent runs, tool calls, model calls, retries, failures, and checkpoint/resume events without changing public APIs.
3. **Complete end-to-end chat and A2A streaming coverage** ([#3456](https://github.com/micro/go-micro/issues/3456)) — provider streaming conformance and A2A fallback work recently shipped, but the mission is one runtime where agents can operate as services, which means streaming must be dependable through the whole path: provider tokens → chat / `Agent.Chat` → A2A. This remains behind conformance and observability because it is a Next-phase depth item, but it is the next user-visible seam in the developer inner loop and interop story.
1. **Add human-in-the-loop pause and resume for agent workflows** ([#3329](https://github.com/micro/go-micro/issues/3329)) — the highest-value remaining Later-phase operability gap now that provider conformance, failure hardening, the getting-started contract, streaming, durable resume/checkpoints, A2A task lifecycle, and `RunInfo` tracing have shipped. Scheduled and looping work still needs durable pending-input states that let humans approve, correct, or add context and then resume the same service/agent/workflow runtime.
2. **gateway/a2a: multiple typed skills per agent card** ([#3342](https://github.com/micro/go-micro/issues/3342)) — the A2A gateway now has the full task lifecycle (send/stream/get/cancel/resubscribe, push config, multi-turn, input-required); the remaining interop gap is skill granularity. `Card` only ever advertises one synthetic "chat" skill with services flattened into tags. Let an agent advertise N typed skills and route per skill, so domain-routing agents expose their real capabilities over A2A through the gateway instead of a hand-rolled handler.
3. **Add a maintained 0-to-hero reference example** ([#3368](https://github.com/micro/go-micro/issues/3368)) — keep the mission legible in code, not only docs: one CI-verifiable example should walk scaffold → run → chat → inspect across typed services, an agent, and a durable flow. This closes the remaining DX/coherence seam between the README promise, website roadmap, and the lived developer inner loop.
_Seeded by Claude Code from the roadmap + open issues; thereafter maintained by the
architecture-review pass._
@@ -1,184 +0,0 @@
// A2A stream fallback harness.
//
// It exercises the gateway boundary that fronts an agent over A2A. The agent is
// configured with tools and memory, but its model streaming path deliberately
// reports ai.ErrStreamingUnsupported; the A2A gateway must fall back to the
// normal Ask path and still complete the same tool-calling run.
package main
import (
"bufio"
"bytes"
"context"
"encoding/json"
"errors"
"flag"
"fmt"
"io"
"net/http"
"net/http/httptest"
"os"
"strings"
"time"
"go-micro.dev/v6/agent"
"go-micro.dev/v6/ai"
"go-micro.dev/v6/gateway/a2a"
"go-micro.dev/v6/registry"
"go-micro.dev/v6/store"
)
type mockModel struct{ opts ai.Options }
func newMock(opts ...ai.Option) ai.Model {
m := &mockModel{}
_ = m.Init(opts...)
return m
}
func (m *mockModel) Init(opts ...ai.Option) error {
for _, o := range opts {
o(&m.opts)
}
return nil
}
func (m *mockModel) Options() ai.Options { return m.opts }
func (m *mockModel) String() string { return "mock" }
func (m *mockModel) Stream(context.Context, *ai.Request, ...ai.GenerateOption) (ai.Stream, error) {
return nil, ai.ErrStreamingUnsupported
}
func (m *mockModel) Generate(ctx context.Context, req *ai.Request, _ ...ai.GenerateOption) (*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 || m.opts.ToolHandler == nil {
return nil, errors.New("missing tools or tool handler")
}
res := m.opts.ToolHandler(ctx, ai.ToolCall{ID: "a2a-fallback-call", Name: "fallback_echo", Input: map[string]any{"value": "a2a-fallback"}})
if res.Content == "" {
return nil, errors.New("empty tool result")
}
return &ai.Response{Reply: "fallback completed", Answer: res.Content, ToolCalls: []ai.ToolCall{{ID: "a2a-fallback-call", Name: "fallback_echo", Input: map[string]any{"value": "a2a-fallback"}, Result: res.Content}}}, nil
}
func providerKey(provider string) string {
if v := os.Getenv("MICRO_AI_API_KEY"); v != "" {
return v
}
env := map[string]string{
"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",
}[provider]
return os.Getenv(env)
}
func main() {
provider := flag.String("provider", "mock", "LLM provider: mock (default), anthropic, openai, ...")
flag.Parse()
apiKey := ""
if *provider == "mock" {
ai.Register("mock", newMock)
} else {
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
}
}
fmt.Printf("\n\033[1mA2A streaming fallback conformance (provider: %s)\033[0m\n", *provider)
reg := registry.NewMemoryRegistry()
st := store.NewMemoryStore()
var sawTool, sawRunInfo bool
ag := agent.New(
agent.Name("a2a-fallback"),
agent.Provider(*provider),
agent.APIKey(apiKey),
agent.Prompt("Use fallback_echo exactly once with value a2a-fallback, then answer with the tool result."),
agent.WithRegistry(reg),
agent.WithStore(st),
agent.WithMemory(agent.NewInMemory(8)),
agent.ModelCallTimeout(45*time.Second),
agent.WithTool("fallback_echo", "Echo the A2A fallback 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 || info.RunID == "" || info.Agent != "a2a-fallback" {
return "", fmt.Errorf("unexpected run info: %+v", info)
}
sawRunInfo = true
if input["value"] != "a2a-fallback" {
return "", fmt.Errorf("unexpected value %v", input["value"])
}
return `{"marker":"a2a-fallback-ok"}`, nil
}),
)
card := a2a.Card("a2a-fallback", "http://example.invalid/a2a-fallback", "", nil)
handler := a2a.NewAgentStreamHandler(card, func(ctx context.Context, text string) (string, error) {
resp, err := ag.Ask(ctx, text)
if err != nil {
return "", err
}
return resp.Reply, nil
}, ag.Stream)
body := []byte(`{"jsonrpc":"2.0","id":1,"method":"message/stream","params":{"message":{"role":"user","parts":[{"kind":"text","text":"Run the A2A fallback conformance check."}],"kind":"message"}}}`)
req := httptest.NewRequest(http.MethodPost, "/", bytes.NewReader(body))
req.Header.Set("Content-Type", "application/json")
rr := httptest.NewRecorder()
handler.ServeHTTP(rr, req)
res := rr.Result()
defer res.Body.Close()
if res.StatusCode != http.StatusOK {
b, _ := io.ReadAll(res.Body)
fmt.Fprintf(os.Stderr, "unexpected status %d: %s\n", res.StatusCode, b)
os.Exit(1)
}
if ct := res.Header.Get("Content-Type"); !strings.HasPrefix(ct, "text/event-stream") {
fmt.Fprintf(os.Stderr, "content-type = %q, want text/event-stream\n", ct)
os.Exit(1)
}
payload, err := readSSEData(res.Body)
if err != nil {
fmt.Fprintln(os.Stderr, err)
os.Exit(1)
}
if !strings.Contains(payload, "a2a-fallback-ok") {
fmt.Fprintf(os.Stderr, "stream payload missing marker: %s\n", payload)
os.Exit(1)
}
if !sawTool || !sawRunInfo {
fmt.Fprintf(os.Stderr, "tool=%v runInfo=%v\n", sawTool, sawRunInfo)
os.Exit(1)
}
fmt.Println("\n\033[32m✓ A2A message/stream fell back to Ask and preserved tool/run metadata\033[0m")
}
func readSSEData(r io.Reader) (string, error) {
scanner := bufio.NewScanner(r)
var payload strings.Builder
for scanner.Scan() {
line := scanner.Text()
if data, ok := strings.CutPrefix(line, "data: "); ok {
payload.WriteString(data)
payload.WriteByte('\n')
}
}
if err := scanner.Err(); err != nil {
return "", err
}
if payload.Len() == 0 {
return "", errors.New("no SSE data received")
}
if !json.Valid([]byte(strings.TrimSpace(payload.String()))) {
return "", fmt.Errorf("SSE data is not JSON: %s", payload.String())
}
return payload.String(), nil
}
@@ -7,17 +7,13 @@ suite is safe for local development, forks, and scheduled CI.
## What it exercises
`go run ./internal/harness/provider-conformance` fans out over the provider-facing
agent test and the harnesses in `internal/harness`:
`go run ./internal/harness/provider-conformance` fans out over the harnesses in
`internal/harness`:
- `agent` — provider tool-call conformance through `agent.Ask`, including run metadata propagation.
- `universe` — service discovery plus agent tool calls over the real runtime.
- `agent-flow` — a workflow event that drives an agent to call services.
- `plan-delegate` — plan persistence plus agent-to-agent delegation and service
calls.
- `a2a-stream-fallback` — A2A `message/stream` through the gateway, including
fallback from unsupported provider streaming to the tool-calling `Ask` path while
preserving run metadata.
The command also emits the registered provider capability matrix so the run shows
which providers advertise model, image, video, and streaming support.
@@ -52,8 +48,7 @@ Provider keys are read from `MICRO_AI_API_KEY` or the provider-specific variable
| AtlasCloud | `ATLASCLOUD_API_KEY` |
Use `-require-configured` when you want a selected provider without a key to fail
instead of skip. This is useful for manually checking that a required provider
secret is actually wired into CI before relying on that provider as covered:
instead of skip:
```sh
go run ./internal/harness/provider-conformance \
@@ -64,18 +59,13 @@ go run ./internal/harness/provider-conformance \
## Scheduled CI behavior
The `Harness (E2E)` workflow runs on pushes and pull requests with deterministic
mock LLMs, including `provider-conformance -providers mock`. On the daily
schedule and manual dispatch it also runs the live provider conformance job. A
manual dispatch can narrow `providers` or `harnesses`, and can set
`require_configured=true` to fail fast when an expected repository secret is
missing; scheduled runs keep the safe default and report missing keys as skips.
That job:
mock LLMs. On the daily schedule and manual dispatch it also runs the live
provider conformance job. That job:
1. runs the same `agent`, `universe`, `agent-flow`, `plan-delegate`, and `a2a-stream-fallback` harness list,
2. reads the provider keys from repository secrets,
3. skips providers whose secrets are absent,
4. fails when any configured provider fails a harness, and
5. uploads JSON and Markdown coverage artifacts for the run.
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
+2 -58
View File
@@ -34,8 +34,6 @@ import (
_ "go-micro.dev/v6/ai/together"
)
const defaultHarnesses = "agent,universe,agent-flow,plan-delegate,a2a-stream-fallback"
var providerEnv = map[string]string{
"anthropic": "ANTHROPIC_API_KEY",
"openai": "OPENAI_API_KEY",
@@ -47,8 +45,8 @@ var providerEnv = map[string]string{
}
func main() {
providersFlag := flag.String("providers", defaultProviders(), "comma-separated providers to check; use mock for deterministic local checks")
harnessesFlag := flag.String("harnesses", defaultHarnesses, "comma-separated harness names under internal/harness; agent runs the provider tool-call conformance test")
providersFlag := flag.String("providers", "anthropic,openai,gemini,groq,mistral,together,atlascloud", "comma-separated providers to check; use mock for deterministic local checks")
harnessesFlag := flag.String("harnesses", "universe,agent-flow,plan-delegate", "comma-separated harness names under internal/harness")
timeoutFlag := flag.Duration("timeout", 10*time.Minute, "timeout per provider/harness run")
requireConfiguredFlag := flag.Bool("require-configured", false, "fail when a selected live provider is missing an API key")
capabilitiesFlag := flag.Bool("capabilities", true, "print the registered provider capability matrix before running conformance")
@@ -171,8 +169,6 @@ func writeSummaryMarkdown(path string, summary conformanceSummary) error {
var b strings.Builder
b.WriteString("# Provider conformance summary\n\n")
fmt.Fprintf(&b, "Passed: %d. Skipped providers: %d. Failed: %d.\n\n", summary.Passed, summary.Skipped, summary.Failed)
fmt.Fprintf(&b, "Providers: %s.\n\n", markdownList(summary.Providers))
fmt.Fprintf(&b, "Harnesses: %s.\n\n", markdownList(summary.Harnesses))
b.WriteString("## Capability matrix\n\n")
b.WriteString(capabilityMarkdown(summary.Capabilities))
b.WriteString("\n## Harness results\n\n")
@@ -198,17 +194,6 @@ func capabilityMarkdown(rows []ai.CapabilityRow) string {
return b.String()
}
func markdownList(values []string) string {
if len(values) == 0 {
return "—"
}
escaped := make([]string, 0, len(values))
for _, value := range values {
escaped = append(escaped, "`"+strings.ReplaceAll(value, "`", "\\`")+"`")
}
return strings.Join(escaped, ", ")
}
func markdownCell(s string) string {
if s == "" {
return "—"
@@ -259,9 +244,6 @@ func validateSelection(providers, harnesses []string) error {
}
for _, harness := range harnesses {
if harness == "agent" {
continue
}
if strings.Contains(harness, string(os.PathSeparator)) || harness == "." || harness == ".." {
return fmt.Errorf("invalid harness name %q", harness)
}
@@ -294,15 +276,6 @@ func repoRoot() string {
}
}
func defaultProviders() string {
providers := make([]string, 0, len(providerEnv))
for provider := range providerEnv {
providers = append(providers, provider)
}
slices.Sort(providers)
return strings.Join(providers, ",")
}
func knownProviders() string {
providers := make([]string, 0, len(providerEnv)+1)
providers = append(providers, "mock")
@@ -350,10 +323,6 @@ func localRPCEnv(env []string) []string {
}
func runHarness(provider, harness string, timeout time.Duration) error {
if harness == "agent" {
return runAgentConformance(provider, timeout)
}
ctx, cancel := context.WithTimeout(context.Background(), timeout)
defer cancel()
@@ -389,28 +358,3 @@ func runHarness(provider, harness string, timeout time.Duration) error {
}
return nil
}
func runAgentConformance(provider string, timeout time.Duration) error {
ctx, cancel := context.WithTimeout(context.Background(), timeout)
defer cancel()
testProvider := provider
if provider == "mock" {
testProvider = "fake"
}
cmd := exec.CommandContext(ctx, "go", "test", "./agent", "-run", "TestAgentProviderConformanceMatrix", "-count=1", "-v")
cmd.Dir = repoRoot()
cmd.Stdout = os.Stdout
cmd.Stderr = os.Stderr
cmd.Env = localRPCEnv(append(os.Environ(),
"GO_MICRO_AGENT_CONFORMANCE_LIVE=1",
"GO_MICRO_AGENT_CONFORMANCE_PROVIDERS="+testProvider,
))
if err := cmd.Run(); err != nil {
if ctx.Err() == context.DeadlineExceeded {
return fmt.Errorf("timed out after %s", timeout)
}
return err
}
return nil
}
@@ -36,18 +36,6 @@ func TestValidateSelectionRejectsUnsafeHarnessName(t *testing.T) {
}
}
func TestDefaultProvidersTracksLiveProviderSet(t *testing.T) {
got := defaultProviders()
for _, want := range []string{"anthropic", "openai", "gemini", "groq", "mistral", "together", "atlascloud"} {
if !strings.Contains(got, want) {
t.Fatalf("defaultProviders() = %q, want %q", got, want)
}
}
if strings.Contains(got, "mock") {
t.Fatalf("defaultProviders() = %q, should not include mock in live scheduled defaults", got)
}
}
func TestCapabilityMatrixHasRegisteredProviders(t *testing.T) {
rows := ai.CapabilityRows()
if len(rows) == 0 {
@@ -117,8 +105,6 @@ func TestWriteSummaryMarkdown(t *testing.T) {
for _, want := range []string{
"# Provider conformance summary",
"Passed: 1. Skipped providers: 1. Failed: 0.",
"Providers: —.",
"Harnesses: —.",
"| mock | ✅ | — | — | — |",
"| mock | agent-flow | passed | — |",
"| live | — | skipped | missing \\| key |",
@@ -129,14 +115,6 @@ func TestWriteSummaryMarkdown(t *testing.T) {
}
}
func TestMarkdownListEscapesBackticks(t *testing.T) {
got := markdownList([]string{"agent", "bad`name"})
want := "`agent`, `bad\\`name`"
if got != want {
t.Fatalf("markdownList() = %q, want %q", got, want)
}
}
func TestWriteSummaryJSON(t *testing.T) {
path := filepath.Join(t.TempDir(), "summary.json")
summary := conformanceSummary{
+7 -13
View File
@@ -4,22 +4,17 @@ This directory owns the no-secret reference scenario for the Go Micro
services → agents → workflows lifecycle. It is intentionally small and
scripted so CI can run it on every push without external services or model keys.
`run.sh` verifies four boundaries together:
`run.sh` verifies three boundaries together:
1. **Run**`micro run` remains available as the local development entry point.
2. **Chat**`micro chat` remains available as the interactive agent entry point.
3. **Inspect**`micro inspect agent <name>` and `micro inspect flow <name>`
remain available as the local run-history inspection step, with `micro flow
runs` preserving durable workflow history inspection.
4. **Deploy**`micro deploy --dry-run <target>` remains available as the
deployment-boundary checkpoint. The dry run resolves configured deploy targets
and services and prints the remote build/copy/systemd/health plan without
building binaries, opening SSH connections, running `rsync`, or touching
remote infrastructure.
After the CLI boundary smoke checks, the script runs the deterministic harnesses
that boot real services, agents, workflows, store-backed run history, plan/delegate,
and A2A with only the LLM mocked.
that boot real services, agents, workflows, store-backed run history, and A2A
with only the LLM mocked.
## Local and CI entry points
@@ -31,8 +26,7 @@ contract locally with:
make harness
```
That target intentionally exercises both 0→1 scaffold variants, the 0→hero
scenario, the event-driven agent-flow harness, and mock provider conformance, so
the public scaffold → run/chat → inspect → deploy lifecycle stays executable
outside CI as well. Live provider checks remain separate and gated by configured
API keys (`make provider-conformance` or the scheduled/manual CI job).
That target intentionally exercises the documented getting-started path before
the 0→hero scenario, so the public scaffold → run/chat → inspect lifecycle stays
executable outside CI as well. Live provider checks remain separate and gated by
configured API keys (`make provider-conformance` or the scheduled/manual CI job).
-1
View File
@@ -7,7 +7,6 @@ cd "$ROOT"
# Keep the developer inner-loop boundaries executable and discoverable in CI
# without secrets or long-running daemons.
go test ./cmd/micro -run TestZeroToHeroCLIBoundaries -count=1
go test ./cmd/micro/cli/deploy -run TestDeployDryRun -count=1
# Deterministic no-secret reference scenarios. These use the real Go Micro
# runtime and mock only the LLM provider.
+3 -3
View File
@@ -11,7 +11,7 @@ description: "Unified service creation, cleaner handler registration, and modula
*March 4, 2026 — By the Go Micro Team*
Go Micro has always prioritized getting out of your way. But over time, the API accumulated multiple ways to do the same thing — `micro.New(name)`, `micro.NewService(micro.Name(...))`, `service.New()`, three different handler registration patterns. If you're building something for AI agents or running a modular monolith, you shouldn't have to choose between equivalent APIs.
Go Micro has always prioritized getting out of your way. But over time, the API accumulated multiple ways to do the same thing — `micro.NewService()`, `micro.NewService()`, `service.New()`, three different handler registration patterns. If you're building something for AI agents or running a modular monolith, you shouldn't have to choose between equivalent APIs.
We've cleaned it up. Here's what changed and why.
@@ -33,7 +33,7 @@ service := micro.NewService("greeter")
service := micro.NewService("greeter", micro.Address(":8080"))
```
Name is always the first argument. Options follow. `micro.New` still works as a deprecated alias, but every example, doc, and guide now uses `micro.NewService("name")`.
Name is always the first argument. Options follow. `NewService` still works (it's deprecated, not removed), but every example, doc, and guide now uses `micro.NewService()`.
## Clean Handler Registration
@@ -107,7 +107,7 @@ Your Go comments become tool descriptions. Your struct tags become parameter sch
If you're building new services, use `micro.NewService("name", opts...)` and `service.Handle()`. That's it.
If you have existing code using `micro.New("name")` or `service.Server().Handle()`, it still works. For v6, update name-less `micro.NewService(opts...)` calls to `micro.NewService("name", opts...)`. The docs, examples, and guides all point to the new patterns now.
If you have existing code using `micro.NewService()` or `service.Server().Handle()`, everything still works — we didn't break anything. But the docs, examples, and guides all point to the new patterns now.
The goal is simple: when someone asks "how do I create a service?", there should be exactly one answer.
+1 -1
View File
@@ -16,7 +16,7 @@ Go Micro has three core abstractions:
## Prerequisites
- **Go 1.24+** for development. The `curl` install below gives you the `micro` binary without Go, but `micro run` compiles your services, so you'll want Go installed to build them.
- **Go 1.21+** for development. The `curl` install below gives you the `micro` binary without Go, but `micro run` compiles your services, so you'll want Go installed to build them.
- An **LLM provider key** (Anthropic, OpenAI, Gemini, …) *only* for the AI features — `micro run --prompt`, `micro chat`, and agents. Plain services need no key. Set it before running, e.g. `export ANTHROPIC_API_KEY=sk-ant-...`.
## Install
+1 -9
View File
@@ -69,18 +69,10 @@ A card looks like:
"capabilities": { "streaming": true, "pushNotifications": true },
"defaultInputModes": ["text/plain"],
"defaultOutputModes": ["text/plain"],
"skills": [
{ "id": "task", "name": "Task", "tags": ["task"] },
{ "id": "project", "name": "Project", "tags": ["project"] }
]
"skills": [{ "id": "chat", "name": "Chat", "tags": ["task", "project"] }]
}
```
Each managed service is advertised as its own typed skill. Clients can call the
whole agent at `/agents/task-mgr`, or address one skill directly at
`/agents/task-mgr/skills/task`; the skill endpoint serves a focused card and
routes the request to the same agent with that skill selected.
## Calling an agent
A2A uses JSON-RPC 2.0 over HTTP. Send a message with `message/send`; the gateway runs the agent and returns a completed `Task`:
@@ -69,14 +69,6 @@ if err != nil {
_ = resp
```
Choose the boundary deliberately: use a durable flow when the steps are known
(`reserve`, `charge`, `confirm`) and each step has deterministic retry/resume
semantics. Use a checkpointed agent run when the model is deciding which tools to
call or how many turns it needs, but the side effects of completed tool calls
still need crash-safe resume. Flows and agents share the same `Checkpoint`
interface, so a flow can safely dispatch to a checkpointed agent for the
open-ended part.
For human-in-the-loop runs that pause through the built-in `request_input` tool,
resume with the operator's response:
@@ -13,7 +13,7 @@ They are exposed to the model as ordinary tools. There is no separate graph runt
## Prerequisites
- Go 1.24+
- Go 1.21+
- An API key for any supported provider (Anthropic, OpenAI, Gemini, Groq, Mistral, Together, Atlas Cloud)
```bash