Compare commits

..

39 Commits

Author SHA1 Message Date
Codex bc1dd0382b docs(priorities): refresh planner queue for 3852
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-07-03 21:34:36 +00:00
Asim Aslam 3cafff8789 Make agent preflight failures actionable (#3851)
Co-authored-by: Codex <codex@openai.com>
2026-07-03 22:00:02 +01:00
Asim Aslam 69e23eb1f6 docs(priorities): refresh planner queue for 3846 (#3848)
Co-authored-by: Codex <codex@openai.com>
2026-07-03 21:34:55 +01:00
Asim Aslam 1841599980 Ensure canceled agent runs fail after tool calls (#3845)
Co-authored-by: Codex <codex@openai.com>
2026-07-03 21:07:58 +01:00
Asim Aslam 6f9994b3ea docs(priorities): refresh planner queue for 3840 (#3842)
Co-authored-by: Codex <codex@openai.com>
2026-07-03 20:33:13 +01:00
Asim Aslam 55070f1597 Harden A2A fallback stream validation (#3839)
Co-authored-by: Codex <codex@openai.com>
2026-07-03 20:03:40 +01:00
Asim Aslam b3940860b6 docs(priorities): refresh planner queue for 3835 (#3836)
Co-authored-by: Codex <codex@openai.com>
2026-07-03 19:39:51 +01:00
Asim Aslam 5c38a23fe4 Classify plan-delegate timeout side effects (#3834)
Co-authored-by: Codex <codex@openai.com>
2026-07-03 19:07:08 +01:00
Asim Aslam 9e7273dde1 docs(priorities): refresh planner queue for 3829 (#3830)
Co-authored-by: Codex <codex@openai.com>
2026-07-03 18:42:19 +01:00
Asim Aslam 42080b6d6d Stabilize file store suffix expiry test (#3828)
Co-authored-by: Codex <codex@openai.com>
2026-07-03 18:18:26 +01:00
Asim Aslam c3df8b833d docs(priorities): refresh planner queue for 3823 (#3824)
Co-authored-by: Codex <codex@openai.com>
2026-07-03 17:38:33 +01:00
Asim Aslam 8694e3a20c test(store): isolate file store tests (#3822)
Co-authored-by: Codex <codex@openai.com>
2026-07-03 17:18:57 +01:00
Asim Aslam 9def03d4e3 docs(priorities): refresh planner queue for 3817 (#3819)
Co-authored-by: Codex <codex@openai.com>
2026-07-03 16:51:21 +01:00
Asim Aslam bd7625f947 Tighten delegated notify harness prompt (#3816)
Co-authored-by: Codex <codex@openai.com>
2026-07-03 16:25:49 +01:00
Asim Aslam 0a547fbe17 docs(priorities): refresh planner queue for 3812 (#3813)
Co-authored-by: Codex <codex@openai.com>
2026-07-03 15:55:48 +01:00
Asim Aslam 331e3f32bd Fix universe concierge notify handoff (#3811)
Co-authored-by: Codex <codex@openai.com>
2026-07-03 15:31:48 +01:00
Asim Aslam 7db1d986ab docs(priorities): refresh planner queue for 3808 (#3809)
Co-authored-by: Codex <codex@openai.com>
2026-07-03 14:48:47 +01:00
Asim Aslam 6042a6c5f5 Add CLI docs wayfinding (#3807)
Co-authored-by: Codex <codex@openai.com>
2026-07-03 14:21:34 +01:00
Asim Aslam bff916a3ac docs(priorities): refresh planner queue for 3800 (#3802)
Co-authored-by: Codex <codex@openai.com>
2026-07-03 13:42:57 +01:00
Asim Aslam b8c03dafa2 Stabilize plan-delegate notify recovery (#3799)
Co-authored-by: Codex <codex@openai.com>
2026-07-03 13:30:58 +01:00
Asim Aslam 04bcef47ac docs(priorities): refresh planner queue for 3795 (#3796)
Co-authored-by: Codex <codex@openai.com>
2026-07-03 12:58:27 +01:00
Asim Aslam e92978f3eb Add minimax to provider conformance (#3794)
Co-authored-by: Codex <codex@openai.com>
2026-07-03 12:30:56 +01:00
Asim Aslam 695a24432a docs(priorities): refresh planner queue for 3790 (#3791)
Co-authored-by: Codex <codex@openai.com>
2026-07-03 12:00:11 +01:00
Asim Aslam 2889b98dcf store: de-flake TestFileStoreTable timing windows (#3789)
TestFileStoreTable failed intermittently on CI ("Expected 2 items, got 1"):
the same commit passed the Unit Tests job on a PR run and failed on the master
push. Cause: records written with a 100ms expiry are read back immediately and
expected to still be present, but under `-race` on a loaded runner the write
loop + file I/O + read can exceed 100ms, so a record expires before the read.

Widen the expiry/TTL windows (100ms -> 1s) and the paired post-expiry sleeps
(-> 2s). The "read before expiry" reads happen in well under 200ms, so they stay
inside the 1s window on any runner; the "read after expiry" waits comfortably
exceed it. Test-only; the store's expiry behavior is unchanged.

Verified: `go test -race -count=5 -run TestFileStoreTable ./store/` green.


Claude-Session: https://claude.ai/code/session_01CmdEY7pYmV5zzwCjNJ4ykL

Co-authored-by: Claude <noreply@anthropic.com>
2026-07-03 11:35:56 +01:00
Asim Aslam a356ab36a8 ai/minimax: complete provider surface (matrix, conformance, changelog) (#3784)
* ai/minimax: complete provider surface (matrix, conformance, changelog)

Follow-up after merging the MiniMax provider (#3769), mirroring the Ollama
completeness pass (#3637):

- Add the `minimax` row to the AI provider capability matrix and blank-import
  ai/minimax in provider_capabilities_test.go so the matrix stays enforced
  against the registry.
- Add minimax to the stream-conformance allowlist (+ import) so its streaming
  is actually exercised against the OpenAI-compatible SSE contract, not just
  registered. It passes via the shared ai/internal/openaiapi path.
- Record the provider in CHANGELOG [Unreleased].

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01CmdEY7pYmV5zzwCjNJ4ykL

* ai: update capabilities_test provider assertions for minimax

Adding the minimax blank-import to the shared ai_test binary (for stream
conformance) also registers it for TestRegisteredProviders / TestCapabilityRows
/ TestCapabilityMatrix in capabilities_test.go, which pin the exact provider
set. Update those assertions to include minimax. (Fixes the Unit Tests failure
my scoped `-run TestStreamProviders` check missed — go compiles all _test.go in
a package into one binary.)

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01CmdEY7pYmV5zzwCjNJ4ykL

---------

Co-authored-by: Claude <noreply@anthropic.com>
2026-07-03 11:10:29 +01:00
Octopus 2078748e7f feat: add MiniMax provider (#3769)
Co-authored-by: octo-patch <266937838+octo-patch@users.noreply.github.com>
2026-07-03 10:56:12 +01:00
Asim Aslam d283f9b08b Accept order-scoped buyer notifications (#3783)
Co-authored-by: Codex <codex@openai.com>
2026-07-03 10:45:47 +01:00
Asim Aslam af7b4f3d50 docs(priorities): refresh planner queue for 3777 (#3778)
Co-authored-by: Codex <codex@openai.com>
2026-07-03 10:17:04 +01:00
Asim Aslam 954f8fa79e docs: update changelog for coherence audit (#3776)
Co-authored-by: Codex <codex@openai.com>
2026-07-03 09:07:28 +01:00
Asim Aslam cdfe9c0947 Fix plan-delegate timeout completion (#3774)
Co-authored-by: Codex <codex@openai.com>
2026-07-03 08:49:33 +01:00
Asim Aslam c645c5faa7 docs(priorities): refresh planner queue for 3768 (#3770)
Co-authored-by: Codex <codex@openai.com>
2026-07-03 08:18:32 +01:00
Asim Aslam 8bde01bdac docs(priorities): refresh planner queue for 3763 (#3764)
Co-authored-by: Codex <codex@openai.com>
2026-07-03 06:31:05 +01:00
Asim Aslam 1373ceec21 harness: parse multi-event A2A SSE (#3762)
Co-authored-by: Codex <codex@openai.com>
2026-07-03 04:53:40 +01:00
Asim Aslam 838b7f73f8 docs(priorities): refresh planner queue for 3757 (#3758)
Co-authored-by: Codex <codex@openai.com>
2026-07-03 04:20:35 +01:00
Asim Aslam c56423a33c Require harness notification side effects (#3756)
Co-authored-by: Codex <codex@openai.com>
2026-07-03 03:29:12 +01:00
Asim Aslam 6edc0de7bd docs(priorities): refresh planner queue for 3752 (#3753)
Co-authored-by: Codex <codex@openai.com>
2026-07-03 02:46:16 +01:00
Asim Aslam 3d31fe37db fix atlascloud tool-call request diagnostics (#3749)
Co-authored-by: Codex <codex@openai.com>
2026-07-03 02:30:44 +01:00
Asim Aslam 6840ae0fb1 docs(priorities): refresh planner queue for 3745 (#3746)
Co-authored-by: Codex <codex@openai.com>
2026-07-03 01:45:59 +01:00
Asim Aslam af61d6327a atlascloud: fall back to tool results (#3744)
Co-authored-by: Codex <codex@openai.com>
2026-07-03 01:00:20 +01:00
34 changed files with 1150 additions and 136 deletions
+3 -6
View File
@@ -21,12 +21,9 @@ changes, architectural rewrites. Those go to the human.
## Work queue (ranked)
1. **Surface AtlasCloud tool results when the follow-up reply is empty** ([#3735](https://github.com/micro/go-micro/issues/3735)) — PR #3738 closed the prior A2A streaming fallback blocker (#3560), and the latest live provider run exposed the next adoption-critical Now-phase conformance seam: AtlasCloud can execute the tool but return an empty assistant follow-up, leaving first-agent/provider conformance with a blank reply instead of the deterministic tool result. Fixing this keeps the service-as-tool story dependable before chasing deeper harness parser work.
2. **Require AtlasCloud notification side effects in multi-step harnesses** ([#3736](https://github.com/micro/go-micro/issues/3736)) — after the basic tool-result path is reliable, the multi-step universe and plan/delegate harnesses must not accept provider replies that claim completion while skipping the required notify/delegate side effect. This protects the services → agents → workflows lifecycle promised by the first-agent and 0→hero paths: an agent should operate the system, not only narrate success.
3. **Parse multi-event A2A SSE fallback responses in the harness** ([#3662](https://github.com/micro/go-micro/issues/3662)) — once the current AtlasCloud tool/side-effect failures are resolved, the harness must accept legitimate multi-event `message/stream` responses instead of concatenating valid SSE events into invalid JSON. This remains a small CI-verifiable harness fix that keeps cross-provider streaming conformance focused on real gateway failures rather than parser brittleness.
4. **Propagate agent run cancellation and deadlines through model and tool calls** ([#3544](https://github.com/micro/go-micro/issues/3544)) — the highest-value remaining resilience gap is predictable failure semantics across agent runs, model calls, tool calls, plan/delegate, and flow handoffs. Tool retries, live-provider deadline tuning, and delegated-plan completion are in place; the lifecycle still needs cancellation/deadline propagation so work fails safely instead of becoming opaque loops.
5. **Emit OpenTelemetry spans for agent run timelines** ([#3525](https://github.com/micro/go-micro/issues/3525)) — recent work made runs inspectable, correlated trace metadata through scheduled dispatch, verified restart resume, added opt-in tool retries, hardened provider conformance, and fixed provider-emitted text tool calls. The next Next-phase step is to turn that RunInfo foundation into standard OTel spans for agent runs, model calls, tool calls, checkpoint/resume, cancellation/deadlines, and failures.
6. **Add an AP2 mandate layer over A2A and x402** ([#3552](https://github.com/micro/go-micro/issues/3552)) — this is a forward interop investment, not a Now-phase blocker: Go Micro already has A2A agents and x402 paid tools, so a small signed-mandate foundation can keep agent payments aligned with the open-protocol story without pulling the queue away from adoption, resilience, or observability. Keep it additive and opt-in while the AP2/FIDO work settles.
1. **Emit OpenTelemetry spans for agent run timelines** ([#3525](https://github.com/micro/go-micro/issues/3525)) — #3851 shipped the remaining Now-phase first-agent preflight doctor, and the README/docs now provide a no-secret first-agent route, debugging guide, examples wayfinding, and 0→hero harness. The highest-value remaining Next-phase increment is turning the `RunInfo` foundation into standard OTel spans for agent runs, model calls, tool calls, checkpoint/resume, cancellation/deadlines, and failures so operators can understand the service → agent → workflow lifecycle in production.
2. **Resume long-running agent runs from checkpoints** ([#3847](https://github.com/micro/go-micro/issues/3847)) — the roadmap and blog promise a harness that recovers from failure, and flows already checkpoint/resume. After the observability foundation, durable agent resume closes the most visible lifecycle seam between flows and agents: long agent work should survive restarts without replaying completed tool side effects or resuming canceled/expired runs.
3. **Add an AP2 mandate layer over A2A and x402** ([#3552](https://github.com/micro/go-micro/issues/3552)) — this is a forward interop investment, not a Now-phase blocker: Go Micro already has A2A agents and x402 paid tools, so a small signed-mandate foundation can keep agent payments aligned with the open-protocol story without pulling the queue away from adoption, resilience, observability, or durable agent operation. Keep it additive and opt-in while the AP2/FIDO work settles.
_Seeded by Claude Code from the roadmap + open issues; thereafter maintained by the
architecture-review pass._
+3 -2
View File
@@ -18,7 +18,7 @@ on:
providers:
description: "Comma-separated providers for live conformance (default: all supported)"
required: false
default: "anthropic,openai,gemini,groq,mistral,together,atlascloud"
default: "anthropic,openai,gemini,groq,minimax,mistral,together,atlascloud"
harnesses:
description: "Comma-separated harnesses for live conformance"
required: false
@@ -64,6 +64,7 @@ jobs:
OPENAI_API_KEY: ${{ secrets.OPENAI_API_KEY }}
GEMINI_API_KEY: ${{ secrets.GEMINI_API_KEY }}
GROQ_API_KEY: ${{ secrets.GROQ_API_KEY }}
MINIMAX_API_KEY: ${{ secrets.MINIMAX_API_KEY }}
MISTRAL_API_KEY: ${{ secrets.MISTRAL_API_KEY }}
TOGETHER_API_KEY: ${{ secrets.TOGETHER_API_KEY }}
ATLASCLOUD_API_KEY: ${{ secrets.ATLASCLOUD_API_KEY }}
@@ -73,7 +74,7 @@ jobs:
# catalog id differs (Atlas uses org/model ids).
ATLASCLOUD_MODEL: ${{ vars.ATLASCLOUD_MODEL || 'minimaxai/minimax-m3' }}
run: |
PROVIDERS="${{ github.event.inputs.providers || 'anthropic,openai,gemini,groq,mistral,together,atlascloud' }}"
PROVIDERS="${{ github.event.inputs.providers || 'anthropic,openai,gemini,groq,minimax,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' }}"
+21
View File
@@ -16,9 +16,30 @@ next version when it ships.
## [Unreleased]
### Added
- **MiniMax provider** — run agents against MiniMax's `MiniMax-M3` model via its OpenAI-compatible endpoint, with tool calling and streaming; auto-detected from the base URL. (`ai/minimax/`)
### Fixed
- **Plan/delegate completion** — agents now continue unfinished plan steps more reliably, fail checkpointed runs that leave delegated plans unfinished, recover from unknown plan-delegate tool calls, avoid duplicate side effects, and complete timeout paths deterministically. (`agent/`)
- **AtlasCloud tool calls** — streaming and request fallback handling now recovers tool-call results from provider responses that omit the expected structured fields. (`ai/atlascloud/`)
---
## [6.3.13] - July 2026
### Added
- **`micro loop`** — scaffold an autonomous improvement loop into any repository: GitHub Actions workflows dispatched to an @mention-driven coding agent, across up to five roles — `planner` (ranked queue), `builder` (top item as a single-concern PR, auto-merged on green CI), `triage` (CI failures → fix issues), and opt-in `coherence` (docs/CHANGELOG alignment) and `release` (daily patch tag). Each dispatch role's instruction lives in an editable `.github/loop/prompts/<role>.md` file — the workflow is the mechanism, the prompt is the policy — so a repo customizes behavior without forking the CLI. `micro loop init --roles …` writes it all; `micro loop verify` checks the wiring. This is the loop that maintains go-micro itself, generalized. (`cmd/micro/loop/`)
### Changed
- **x402 payments** — settlement now covers CDP facilitator authentication and conformance edge cases. (`wrapper/x402/`)
### Fixed
- **Plan/delegate harnessing** — side effects and notifications are now idempotent and deterministic across duplicate, alias, order-scoped, and reachability scenarios. (`agent/`, `internal/harness/`)
### Documentation
- **First-agent on-ramp** — quickstart docs now connect the no-secret first-agent transcript, example map, and 0→hero path. (`README.md`, `internal/website/docs/`)
- **Ollama provider docs** — the provider surface, capability matrix, and examples now document local and cloud behavior. (`internal/website/docs/`, `examples/agent-ollama/`)
---
## [6.3.12] - July 2026
+2 -1
View File
@@ -332,7 +332,7 @@ MCP exposes your services as tools; A2A exposes your agents as agents. See the [
| MCP gateway | Every endpoint is an AI tool automatically |
| A2A gateway | Every agent is reachable over the Agent2Agent protocol; cards generated from the registry (`micro a2a`) |
| Payments (x402) | Opt-in per-call payments for tools via the x402 standard; pluggable facilitator (Base, Solana, …) |
| 8 LLM providers | Anthropic, OpenAI, Gemini, Groq, Mistral, Together, Atlas Cloud, Ollama (local + cloud) |
| 9 LLM providers | Anthropic, OpenAI, Gemini, Groq, Mistral, Together, Atlas Cloud, MiniMax, Ollama (local + cloud) |
| Interactive console | `micro run` includes a chat console for talking to services |
| Service generation | `micro run --prompt` — describe a system, get running services |
@@ -429,6 +429,7 @@ Swap providers with a single import — same interface everywhere:
| Mistral | `mistral-large-latest` |
| Together AI | `meta-llama/Llama-3.3-70B-Instruct-Turbo` |
| Atlas Cloud | `deepseek-ai/DeepSeek-V3-0324` |
| MiniMax | `MiniMax-M3` |
| Ollama | `llama3.2` (local) |
```go
+28
View File
@@ -132,6 +132,34 @@ func TestToolCallTimeoutPropagatesDeadlineToCustomTool(t *testing.T) {
}
}
func TestAskCancellationDuringToolCallFailsRun(t *testing.T) {
fakeGen = func(ctx context.Context, opts ai.Options, req *ai.Request) (*ai.Response, error) {
if opts.ToolHandler == nil {
t.Fatal("missing tool handler")
}
res := opts.ToolHandler(ctx, ai.ToolCall{ID: "call-1", Name: "cancel-self"})
if !strings.Contains(res.Content, context.Canceled.Error()) {
t.Fatalf("tool result = %q, want cancellation error", res.Content)
}
return &ai.Response{Reply: "should not succeed"}, nil
}
defer func() { fakeGen = nil }()
ctx, cancel := context.WithCancel(context.Background())
a := newTestAgent(
Name("cancel-during-tool"),
WithTool("cancel-self", "cancel the run context", nil, func(context.Context, map[string]any) (string, error) {
cancel()
return "", context.Canceled
}),
)
_, err := a.Ask(ctx, "cancel during tool")
if !errors.Is(err, context.Canceled) {
t.Fatalf("Ask error = %v, want context canceled", err)
}
}
func TestAskCheckpointRecordsTerminalOperationalFailureStatus(t *testing.T) {
tests := []struct {
name string
+14
View File
@@ -300,6 +300,20 @@ Default base URL: `https://api.atlascloud.ai`
Atlas Cloud is an enterprise AI infrastructure platform offering high-performance LLM APIs. It exposes an OpenAI-compatible chat completions endpoint with tool calling support.
### MiniMax
```go
m := ai.New("minimax",
ai.WithAPIKey("your-key"),
ai.WithModel("MiniMax-M3"), // default
)
```
Default model: `MiniMax-M3`
Default base URL: `https://api.minimax.io`
MiniMax offers its flagship MiniMax-M3 model via an OpenAI-compatible chat completions endpoint.
## Auto-Detection
Use `AutoDetectProvider()` to detect the provider from a base URL:
+76 -14
View File
@@ -52,6 +52,15 @@ type Provider struct {
opts ai.Options
}
type atlasToolCall struct {
ID string `json:"id"`
Type string `json:"type"`
Function struct {
Name string `json:"name"`
Arguments string `json:"arguments"`
} `json:"function"`
}
// NewProvider creates a new Atlas Cloud provider.
func NewProvider(opts ...ai.Option) *Provider {
options := ai.NewOptions(opts...)
@@ -121,7 +130,7 @@ func (p *Provider) Generate(ctx context.Context, req *ai.Request, opts ...ai.Gen
apiReq["tools"] = tools
}
resp, rawMessage, err := p.callAPI(ctx, apiReq)
resp, rawMessage, err := p.callAPI(ctx, "chat", apiReq)
if err != nil {
return nil, err
}
@@ -155,8 +164,11 @@ func (p *Provider) Generate(ctx context.Context, req *ai.Request, opts ...ai.Gen
"messages": followUpMessages,
}
followUpResp, _, err := p.callAPI(ctx, followUpReq)
if err == nil && followUpResp.Reply != "" {
followUpResp, _, err := p.callAPI(ctx, "tool-follow-up", followUpReq)
if err != nil {
return nil, err
}
if followUpResp.Reply != "" {
resp.Answer = followUpResp.Reply
} else if len(toolResults) > 0 {
resp.Answer = strings.Join(toolResults, "\n")
@@ -277,7 +289,7 @@ func (s *atlasStream) Close() error {
return s.body.Close()
}
func (p *Provider) callAPI(ctx context.Context, req map[string]any) (*ai.Response, map[string]any, error) {
func (p *Provider) callAPI(ctx context.Context, phase string, req map[string]any) (*ai.Response, map[string]any, error) {
reqBody, err := json.Marshal(req)
if err != nil {
return nil, nil, fmt.Errorf("failed to marshal request: %w", err)
@@ -300,20 +312,14 @@ func (p *Provider) callAPI(ctx context.Context, req map[string]any) (*ai.Respons
respBody, _ := io.ReadAll(httpResp.Body)
if httpResp.StatusCode != http.StatusOK {
return nil, nil, fmt.Errorf("API error (%s): %s", httpResp.Status, string(respBody))
return nil, nil, fmt.Errorf("API error (%s) during atlascloud %s request (%s): %s", httpResp.Status, phase, atlascloudRequestSummary(req), string(respBody))
}
var chatResp struct {
Choices []struct {
Message struct {
Content string `json:"content"`
ToolCalls []struct {
ID string `json:"id"`
Function struct {
Name string `json:"name"`
Arguments string `json:"arguments"`
} `json:"function"`
} `json:"tool_calls"`
Content string `json:"content"`
ToolCalls []atlasToolCall `json:"tool_calls"`
} `json:"message"`
} `json:"choices"`
}
@@ -345,12 +351,68 @@ func (p *Provider) callAPI(ctx context.Context, req map[string]any) (*ai.Respons
rawMessage := map[string]any{
"content": choice.Message.Content,
"tool_calls": choice.Message.ToolCalls,
"tool_calls": normalizeAtlasCloudToolCalls(choice.Message.ToolCalls),
}
return response, rawMessage, nil
}
func normalizeAtlasCloudToolCalls(toolCalls []atlasToolCall) []map[string]any {
out := make([]map[string]any, 0, len(toolCalls))
for _, tc := range toolCalls {
toolType := tc.Type
if toolType == "" {
toolType = "function"
}
out = append(out, map[string]any{
"id": tc.ID,
"type": toolType,
"function": map[string]any{
"name": tc.Function.Name,
"arguments": tc.Function.Arguments,
},
})
}
return out
}
func atlascloudRequestSummary(req map[string]any) string {
parts := []string{}
if model, ok := req["model"].(string); ok && model != "" {
parts = append(parts, "model="+model)
}
if messages, ok := req["messages"].([]map[string]any); ok {
parts = append(parts, fmt.Sprintf("messages=%d", len(messages)))
if len(messages) > 0 {
last := messages[len(messages)-1]
if role, ok := last["role"].(string); ok && role != "" {
parts = append(parts, "last_role="+role)
}
if _, ok := last["tool_call_id"].(string); ok {
parts = append(parts, "last_has_tool_call_id=true")
}
}
}
if tools, ok := req["tools"].([]map[string]any); ok {
names := make([]string, 0, len(tools))
for _, tool := range tools {
fn, _ := tool["function"].(map[string]any)
name, _ := fn["name"].(string)
if name != "" {
names = append(names, name)
}
}
parts = append(parts, fmt.Sprintf("tools=%d", len(tools)))
if len(names) > 0 {
parts = append(parts, "tool_names="+strings.Join(names, ","))
}
}
if len(parts) == 0 {
return "request_context=unavailable"
}
return strings.Join(parts, " ")
}
const defaultImageModel = "openai/gpt-image-2/text-to-image"
// GenerateImage creates an image using Atlas Cloud's async image API.
+119
View File
@@ -7,6 +7,7 @@ import (
"io"
"net/http"
"net/http/httptest"
"strings"
"testing"
"go-micro.dev/v6/ai"
@@ -203,6 +204,124 @@ func TestProvider_GenerateToolCallEmptyFollowUpUsesToolResult(t *testing.T) {
}
}
func TestProvider_GenerateMinimaxToolRequests(t *testing.T) {
var bodies []map[string]any
ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
var body map[string]any
if err := json.NewDecoder(r.Body).Decode(&body); err != nil {
t.Fatalf("decode request: %v", err)
}
bodies = append(bodies, body)
w.Header().Set("Content-Type", "application/json")
switch len(bodies) {
case 1:
_, _ = w.Write([]byte(`{"choices":[{"message":{"content":"","tool_calls":[{"id":"call-1","function":{"name":"conformance_echo","arguments":"{\"value\":\"agent-conformance\"}"}}]}}]}`))
case 2:
_, _ = w.Write([]byte(`{"choices":[{"message":{"content":"done"}}]}`))
default:
t.Fatalf("unexpected API call %d", len(bodies))
}
}))
defer ts.Close()
p := NewProvider(
ai.WithAPIKey("test-key"),
ai.WithBaseURL(ts.URL),
ai.WithModel("minimaxai/minimax-m3"),
ai.WithToolHandler(func(ctx context.Context, call ai.ToolCall) ai.ToolResult {
return ai.ToolResult{ID: call.ID, Content: `{"marker":"agent-conformance-ok"}`}
}),
)
resp, err := p.Generate(context.Background(), &ai.Request{
SystemPrompt: "You are helpful.",
Prompt: "call a tool",
Tools: []ai.Tool{{
Name: "conformance_echo",
Description: "echo conformance marker",
Properties: map[string]any{"value": map[string]any{"type": "string"}},
}},
})
if err != nil {
t.Fatalf("Generate returned error: %v", err)
}
if resp.Answer != "done" {
t.Fatalf("Answer = %q, want done", resp.Answer)
}
if len(bodies) != 2 {
t.Fatalf("captured requests = %d, want 2", len(bodies))
}
if got := bodies[0]["model"]; got != "minimaxai/minimax-m3" {
t.Fatalf("initial model = %v", got)
}
tools, ok := bodies[0]["tools"].([]any)
if !ok || len(tools) != 1 {
t.Fatalf("initial tools = %#v, want one tool", bodies[0]["tools"])
}
tool := tools[0].(map[string]any)
if tool["type"] != "function" {
t.Fatalf("tool type = %v, want function", tool["type"])
}
fn := tool["function"].(map[string]any)
if fn["name"] != "conformance_echo" {
t.Fatalf("tool function name = %v", fn["name"])
}
params := fn["parameters"].(map[string]any)
if params["type"] != "object" {
t.Fatalf("parameters type = %v, want object", params["type"])
}
followUpMessages := bodies[1]["messages"].([]any)
if len(followUpMessages) != 4 {
t.Fatalf("follow-up messages = %d, want 4", len(followUpMessages))
}
assistant := followUpMessages[2].(map[string]any)
if assistant["role"] != "assistant" {
t.Fatalf("assistant role = %v", assistant["role"])
}
assistantCalls := assistant["tool_calls"].([]any)
assistantCall := assistantCalls[0].(map[string]any)
if assistantCall["type"] != "function" {
t.Fatalf("assistant tool call type = %v, want function", assistantCall["type"])
}
toolResult := followUpMessages[3].(map[string]any)
if toolResult["role"] != "tool" || toolResult["tool_call_id"] != "call-1" {
t.Fatalf("tool result message = %#v", toolResult)
}
}
func TestProvider_GenerateToolCallHTTPErrorIncludesRequestContext(t *testing.T) {
ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
http.Error(w, `{"code":400,"msg":"bad request"}`, http.StatusBadRequest)
}))
defer ts.Close()
p := NewProvider(
ai.WithAPIKey("test-key"),
ai.WithBaseURL(ts.URL),
ai.WithModel("minimaxai/minimax-m3"),
)
_, err := p.Generate(context.Background(), &ai.Request{
Prompt: "call a tool",
Tools: []ai.Tool{{
Name: "conformance_echo",
Description: "echo conformance marker",
Properties: map[string]any{"value": map[string]any{"type": "string"}},
}},
})
if err == nil {
t.Fatal("Generate error = nil, want 400")
}
msg := err.Error()
for _, want := range []string{"400 Bad Request", "atlascloud chat request", "model=minimaxai/minimax-m3", "tools=1", "tool_names=conformance_echo"} {
if !strings.Contains(msg, want) {
t.Fatalf("error %q missing %q", msg, want)
}
}
if strings.Contains(msg, "test-key") {
t.Fatalf("error leaked API key: %s", msg)
}
}
func TestProvider_Registration(t *testing.T) {
m := ai.New("atlascloud", ai.WithAPIKey("test"))
if m == nil {
+6 -4
View File
@@ -9,6 +9,7 @@ import (
_ "go-micro.dev/v6/ai/atlascloud"
_ "go-micro.dev/v6/ai/gemini"
_ "go-micro.dev/v6/ai/groq"
_ "go-micro.dev/v6/ai/minimax"
_ "go-micro.dev/v6/ai/mistral"
_ "go-micro.dev/v6/ai/openai"
_ "go-micro.dev/v6/ai/together"
@@ -16,7 +17,7 @@ import (
func TestRegisteredProviders(t *testing.T) {
got := ai.RegisteredProviders("")
want := []string{"anthropic", "atlascloud", "gemini", "groq", "mistral", "openai", "together"}
want := []string{"anthropic", "atlascloud", "gemini", "groq", "minimax", "mistral", "openai", "together"}
if !reflect.DeepEqual(got, want) {
t.Fatalf("RegisteredProviders() = %#v, want %#v", got, want)
}
@@ -34,7 +35,7 @@ func TestRegisteredProviders(t *testing.T) {
}
got = ai.RegisteredProviders("stream")
want = []string{"atlascloud", "groq", "mistral", "openai", "together"}
want = []string{"atlascloud", "groq", "minimax", "mistral", "openai", "together"}
if !reflect.DeepEqual(got, want) {
t.Fatalf("RegisteredProviders(stream) = %#v, want %#v", got, want)
}
@@ -47,6 +48,7 @@ func TestCapabilityRows(t *testing.T) {
{Provider: "atlascloud", Capabilities: ai.Capabilities{Model: true, Image: true, Video: true, Stream: true}},
{Provider: "gemini", Capabilities: ai.Capabilities{Model: true}},
{Provider: "groq", Capabilities: ai.Capabilities{Model: true, Stream: true}},
{Provider: "minimax", Capabilities: ai.Capabilities{Model: true, Stream: true}},
{Provider: "mistral", Capabilities: ai.Capabilities{Model: true, Stream: true}},
{Provider: "openai", Capabilities: ai.Capabilities{Model: true, Image: true, Stream: true}},
{Provider: "together", Capabilities: ai.Capabilities{Model: true, Stream: true}},
@@ -59,7 +61,7 @@ func TestCapabilityRows(t *testing.T) {
func TestCapabilityMatrix(t *testing.T) {
matrix := ai.CapabilityMatrix()
for _, provider := range []string{"anthropic", "atlascloud", "gemini", "groq", "mistral", "openai", "together"} {
for _, provider := range []string{"anthropic", "atlascloud", "gemini", "groq", "minimax", "mistral", "openai", "together"} {
caps, ok := matrix[provider]
if !ok {
t.Fatalf("CapabilityMatrix missing %q", provider)
@@ -88,7 +90,7 @@ func TestRegisterStream(t *testing.T) {
}
got := ai.RegisteredProviders("stream")
want := []string{"atlascloud", "groq", "mistral", "openai", "test-stream", "together"}
want := []string{"atlascloud", "groq", "minimax", "mistral", "openai", "test-stream", "together"}
if !reflect.DeepEqual(got, want) {
t.Fatalf("RegisteredProviders(stream) = %#v, want %#v", got, want)
}
+196
View File
@@ -0,0 +1,196 @@
// Package minimax implements the MiniMax model provider.
//
// MiniMax offers its flagship MiniMax-M3 model via an OpenAI-compatible
// chat completions endpoint.
//
// Usage:
//
// import _ "go-micro.dev/v6/ai/minimax"
//
// m := ai.New("minimax",
// ai.WithAPIKey("your-api-key"),
// )
package minimax
import (
"bytes"
"context"
"encoding/json"
"fmt"
"io"
"net/http"
"strings"
"go-micro.dev/v6/ai"
"go-micro.dev/v6/ai/internal/openaiapi"
)
func init() {
ai.Register("minimax", func(opts ...ai.Option) ai.Model {
return NewProvider(opts...)
})
ai.RegisterStream("minimax")
}
type Provider struct {
opts ai.Options
}
func NewProvider(opts ...ai.Option) *Provider {
options := ai.NewOptions(opts...)
if options.Model == "" {
options.Model = "MiniMax-M3"
}
if options.BaseURL == "" {
options.BaseURL = "https://api.minimax.io"
}
return &Provider{opts: options}
}
func (p *Provider) Init(opts ...ai.Option) error {
for _, o := range opts {
o(&p.opts)
}
return nil
}
func (p *Provider) Options() ai.Options { return p.opts }
func (p *Provider) String() string { return "minimax" }
func (p *Provider) Generate(ctx context.Context, req *ai.Request, opts ...ai.GenerateOption) (*ai.Response, error) {
var tools []map[string]any
for _, t := range req.Tools {
tools = append(tools, map[string]any{
"type": "function",
"function": map[string]any{
"name": t.Name,
"description": t.Description,
"parameters": map[string]any{
"type": "object",
"properties": t.Properties,
},
},
})
}
messages := []map[string]any{
{"role": "system", "content": req.SystemPrompt},
{"role": "user", "content": req.Prompt},
}
apiReq := map[string]any{
"model": p.opts.Model,
"messages": messages,
}
if len(tools) > 0 {
apiReq["tools"] = tools
}
resp, rawMessage, err := p.callAPI(ctx, apiReq)
if err != nil {
return nil, err
}
if len(resp.ToolCalls) == 0 {
return resp, nil
}
if p.opts.ToolHandler != nil {
followUpMessages := append(messages, map[string]any{
"role": "assistant",
"content": rawMessage["content"],
"tool_calls": rawMessage["tool_calls"],
})
for _, tc := range resp.ToolCalls {
content := p.opts.ToolHandler(ctx, tc).Content
followUpMessages = append(followUpMessages, map[string]any{
"role": "tool",
"tool_call_id": tc.ID,
"content": content,
})
}
followUpResp, _, err := p.callAPI(ctx, map[string]any{
"model": p.opts.Model,
"messages": followUpMessages,
})
if err == nil && followUpResp.Reply != "" {
resp.Answer = followUpResp.Reply
}
}
return resp, nil
}
func (p *Provider) Stream(ctx context.Context, req *ai.Request, opts ...ai.GenerateOption) (ai.Stream, error) {
return openaiapi.Stream(ctx, p.opts, req, "/v1/chat/completions")
}
func (p *Provider) callAPI(ctx context.Context, req map[string]any) (*ai.Response, map[string]any, error) {
reqBody, err := json.Marshal(req)
if err != nil {
return nil, nil, fmt.Errorf("failed to marshal request: %w", err)
}
apiURL := strings.TrimRight(p.opts.BaseURL, "/") + "/v1/chat/completions"
httpReq, err := http.NewRequestWithContext(ctx, http.MethodPost, apiURL, bytes.NewReader(reqBody))
if err != nil {
return nil, nil, fmt.Errorf("failed to create request: %w", err)
}
httpReq.Header.Set("Content-Type", "application/json")
httpReq.Header.Set("Authorization", "Bearer "+p.opts.APIKey)
httpResp, err := http.DefaultClient.Do(httpReq)
if err != nil {
return nil, nil, fmt.Errorf("API request failed: %w", err)
}
defer httpResp.Body.Close()
respBody, _ := io.ReadAll(httpResp.Body)
if httpResp.StatusCode != http.StatusOK {
return nil, nil, fmt.Errorf("API error (%s): %s", httpResp.Status, string(respBody))
}
var chatResp struct {
Choices []struct {
Message struct {
Content string `json:"content"`
ToolCalls []struct {
ID string `json:"id"`
Function struct {
Name string `json:"name"`
Arguments string `json:"arguments"`
} `json:"function"`
} `json:"tool_calls"`
} `json:"message"`
} `json:"choices"`
}
if err := json.Unmarshal(respBody, &chatResp); err != nil {
return nil, nil, fmt.Errorf("failed to parse response: %w", err)
}
if len(chatResp.Choices) == 0 {
return nil, nil, fmt.Errorf("no response from API")
}
choice := chatResp.Choices[0]
response := &ai.Response{Reply: choice.Message.Content}
for _, tc := range choice.Message.ToolCalls {
var input map[string]any
if err := json.Unmarshal([]byte(tc.Function.Arguments), &input); err != nil {
input = map[string]any{}
}
response.ToolCalls = append(response.ToolCalls, ai.ToolCall{
ID: tc.ID,
Name: tc.Function.Name,
Input: input,
})
}
rawMessage := map[string]any{
"content": choice.Message.Content,
"tool_calls": choice.Message.ToolCalls,
}
return response, rawMessage, nil
}
+96
View File
@@ -0,0 +1,96 @@
package minimax
import (
"context"
"encoding/json"
"errors"
"io"
"net/http"
"net/http/httptest"
"testing"
"go-micro.dev/v6/ai"
)
func TestProvider_String(t *testing.T) {
if NewProvider().String() != "minimax" {
t.Errorf("got %q", NewProvider().String())
}
}
func TestProvider_Defaults(t *testing.T) {
opts := NewProvider().Options()
if opts.Model != "MiniMax-M3" {
t.Errorf("default model = %q", opts.Model)
}
if opts.BaseURL != "https://api.minimax.io" {
t.Errorf("default base URL = %q", opts.BaseURL)
}
}
func TestProvider_Init(t *testing.T) {
p := NewProvider()
if err := p.Init(ai.WithModel("m"), ai.WithAPIKey("k")); err != nil {
t.Fatal(err)
}
if p.Options().Model != "m" || p.Options().APIKey != "k" {
t.Error("Init did not apply options")
}
}
func TestProvider_Generate_NoAPIKey(t *testing.T) {
if _, err := NewProvider().Generate(context.Background(), &ai.Request{Prompt: "hi"}); err == nil {
t.Error("expected error without API key")
}
}
func TestProvider_Stream(t *testing.T) {
var sawStream bool
ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
if r.URL.Path != "/v1/chat/completions" {
t.Fatalf("path = %s, want /v1/chat/completions", r.URL.Path)
}
var body map[string]any
if err := json.NewDecoder(r.Body).Decode(&body); err != nil {
t.Fatalf("decode request: %v", err)
}
sawStream, _ = body["stream"].(bool)
w.Header().Set("Content-Type", "text/event-stream")
_, _ = w.Write([]byte("data: {\"choices\":[{\"delta\":{\"content\":\"hel\"}}]}\n\n"))
_, _ = w.Write([]byte("data: {\"choices\":[{\"delta\":{\"content\":\"lo\"}}]}\n\n"))
_, _ = w.Write([]byte("data: [DONE]\n\n"))
}))
defer ts.Close()
p := NewProvider(ai.WithAPIKey("test-key"), ai.WithBaseURL(ts.URL))
stream, err := p.Stream(context.Background(), &ai.Request{Prompt: "Hello"})
if err != nil {
t.Fatalf("Stream returned error: %v", err)
}
defer stream.Close()
if !sawStream {
t.Fatal("stream request did not set stream=true")
}
first, err := stream.Recv()
if err != nil || first.Reply != "hel" {
t.Fatalf("first chunk = %#v, %v; want hel", first, err)
}
second, err := stream.Recv()
if err != nil || second.Reply != "lo" {
t.Fatalf("second chunk = %#v, %v; want lo", second, err)
}
if _, err := stream.Recv(); !errors.Is(err, io.EOF) {
t.Fatalf("final error = %v, want EOF", err)
}
}
func TestProvider_Registration(t *testing.T) {
m := ai.New("minimax", ai.WithAPIKey("test"))
if m == nil {
t.Fatal("provider not registered")
}
if m.String() != "minimax" {
t.Errorf("got %q", m.String())
}
}
+2
View File
@@ -212,6 +212,8 @@ func AutoDetectProvider(baseURL string) string {
return "gemini"
case strings.Contains(baseURL, "groq"):
return "groq"
case strings.Contains(baseURL, "minimax"):
return "minimax"
case strings.Contains(baseURL, "mistral"):
return "mistral"
case strings.Contains(baseURL, "together"):
+8 -4
View File
@@ -99,15 +99,19 @@ func GenerateWithRetry(ctx context.Context, m Model, req *Request, policy Genera
}
resp, err := m.Generate(callCtx, req, opts...)
cancel()
// Caller cancellation/deadline always wins and is not retried, even if
// a provider or tool loop swallowed the canceled tool result and returned
// a final response. This keeps agent runs from appearing successful after
// their controlling context was abandoned.
if ctxErr := ctx.Err(); ctxErr != nil {
return nil, ctxErr
}
if err == nil {
return resp, nil
}
last = err
// Caller cancellation/deadline always wins and is not retried.
if ctx.Err() != nil {
return nil, ctx.Err()
}
transient := IsTransientError(err)
if attempt == policy.MaxAttempts || !transient {
if attempt > 1 || transient {
+3 -1
View File
@@ -18,6 +18,7 @@ import (
_ "go-micro.dev/v6/ai/atlascloud"
_ "go-micro.dev/v6/ai/gemini"
_ "go-micro.dev/v6/ai/groq"
_ "go-micro.dev/v6/ai/minimax"
_ "go-micro.dev/v6/ai/mistral"
_ "go-micro.dev/v6/ai/openai"
_ "go-micro.dev/v6/ai/together"
@@ -278,6 +279,7 @@ func conformingStreamProviders(t *testing.T) []string {
allowed := map[string]struct{}{
"atlascloud": {},
"groq": {},
"minimax": {},
"mistral": {},
"openai": {},
"together": {},
@@ -288,7 +290,7 @@ func conformingStreamProviders(t *testing.T) []string {
out = append(out, provider)
}
}
want := []string{"atlascloud", "groq", "mistral", "openai", "together"}
want := []string{"atlascloud", "groq", "minimax", "mistral", "openai", "together"}
if !reflect.DeepEqual(out, want) {
t.Fatalf("conforming stream providers = %#v, want %#v (registered stream providers: %#v)", out, want, providers)
}
+29 -6
View File
@@ -16,6 +16,7 @@ type preflightCheck struct {
OK bool
Detail string
Fix string
Next string
}
type preflightDeps struct {
@@ -52,6 +53,9 @@ func runAgentPreflight(w io.Writer, deps preflightDeps) error {
if !check.OK && check.Fix != "" {
fmt.Fprintf(w, " Fix: %s\n", check.Fix)
}
if !check.OK && check.Next != "" {
fmt.Fprintf(w, " Next: %s\n", check.Next)
}
}
if failures > 0 {
return fmt.Errorf("first-agent preflight failed: %d check(s) need attention", failures)
@@ -87,19 +91,23 @@ func agentPreflightChecks(deps preflightDeps) []preflightCheck {
func checkGoToolchain(deps preflightDeps) preflightCheck {
path, err := deps.lookPath("go")
if err != nil {
return preflightCheck{Name: "Go toolchain", Fix: "Install Go 1.24 or newer and ensure go is on PATH."}
return preflightCheck{Name: "Go toolchain", Detail: "go was not found on PATH", Fix: "Install Go 1.24 or newer from https://go.dev/doc/install and ensure go is on PATH.", Next: "After installing Go, rerun micro agent preflight, then continue with docs/guides/your-first-agent.html."}
}
out, err := deps.commandOutput("go", "version")
if err != nil {
return preflightCheck{Name: "Go toolchain", Detail: strings.TrimSpace(string(out)), Fix: "Ensure the go command runs successfully."}
return preflightCheck{Name: "Go toolchain", Detail: strings.TrimSpace(string(out)), Fix: "Ensure the go command runs successfully (try `go version`) before starting the agent walkthrough.", Next: "Use docs/guides/debugging-agents.html after the toolchain check passes if an agent run still fails."}
}
return preflightCheck{Name: "Go toolchain", OK: true, Detail: fmt.Sprintf("%s (%s)", firstLine(out), path)}
version := firstLine(out)
if !goVersionAtLeast(version, 1, 24) {
return preflightCheck{Name: "Go toolchain", Detail: fmt.Sprintf("%s (%s)", version, path), Fix: "Upgrade to Go 1.24 or newer before running generated services.", Next: "Rerun micro agent preflight, then continue with docs/guides/your-first-agent.html."}
}
return preflightCheck{Name: "Go toolchain", OK: true, Detail: fmt.Sprintf("%s (%s)", version, path)}
}
func checkMicroBinary(deps preflightDeps) preflightCheck {
exe, err := deps.executable()
if err != nil || exe == "" {
return preflightCheck{Name: "micro binary", Fix: "Install the micro CLI or run this check through go run ./cmd/micro agent preflight."}
return preflightCheck{Name: "micro binary", Detail: "micro executable path is unavailable", Fix: "Install the micro CLI or run this check through `go run ./cmd/micro agent preflight` from the repository.", Next: "Then follow docs/getting-started.html for the scaffold -> run path."}
}
version := deps.version()
if version == "" {
@@ -117,7 +125,7 @@ func checkProviderKey(deps preflightDeps) preflightCheck {
}
}
if len(found) == 0 {
return preflightCheck{Name: "provider API key", Detail: "no supported provider key found", Fix: "Export MICRO_AI_API_KEY or a provider key such as ANTHROPIC_API_KEY before running provider-backed agents."}
return preflightCheck{Name: "provider API key", Detail: "no supported provider key found", Fix: "Export MICRO_AI_API_KEY or a provider key such as ANTHROPIC_API_KEY before running provider-backed agents.", Next: "For a no-secret path, run the mock-model walkthrough in docs/guides/no-secret-first-agent.html; for real providers, see docs/guides/debugging-agents.html#provider-failures."}
}
return preflightCheck{Name: "provider API key", OK: true, Detail: "found " + strings.Join(found, ", ")}
}
@@ -125,7 +133,7 @@ func checkProviderKey(deps preflightDeps) preflightCheck {
func checkPortAvailable(deps preflightDeps, addr, use string) preflightCheck {
ln, err := deps.listen("tcp", addr)
if err != nil {
return preflightCheck{Name: "local port " + addr, Detail: "busy or unavailable for " + use, Fix: "Stop the process using " + addr + " or run micro run --address with a free port."}
return preflightCheck{Name: "local port " + addr, Detail: "busy or unavailable for " + use, Fix: "Stop the process using " + addr + " (for example, `lsof -i :8080`) or run `micro run --address` with a free port.", Next: "Once the gateway starts, open http://localhost:8080/agent or continue with docs/guides/your-first-agent.html#chat-with-your-agent."}
}
_ = ln.Close()
return preflightCheck{Name: "local port " + addr, OK: true, Detail: "available for " + use}
@@ -138,3 +146,18 @@ func firstLine(b []byte) string {
}
return s
}
func goVersionAtLeast(line string, wantMajor, wantMinor int) bool {
idx := strings.Index(line, "go1.")
if idx < 0 {
return false
}
var major, minor int
if _, err := fmt.Sscanf(line[idx:], "go%d.%d", &major, &minor); err != nil {
return false
}
if major != wantMajor {
return major > wantMajor
}
return minor >= wantMinor
}
+47 -1
View File
@@ -61,13 +61,59 @@ func TestRunAgentPreflightReportsActionableFailures(t *testing.T) {
t.Fatal("runAgentPreflight() error = nil")
}
got := out.String()
for _, want := range []string{"✗ Go toolchain", "Install Go 1.24", "✗ micro binary", "✗ provider API key", "ANTHROPIC_API_KEY", "✗ local port :8080", "micro run --address"} {
for _, want := range []string{"✗ Go toolchain", "go was not found on PATH", "https://go.dev/doc/install", "docs/guides/your-first-agent.html", "✗ micro binary", "go run ./cmd/micro agent preflight", "✗ provider API key", "docs/guides/no-secret-first-agent.html", "docs/guides/debugging-agents.html#provider-failures", "✗ local port :8080", "lsof -i :8080", "micro run --address"} {
if !strings.Contains(got, want) {
t.Fatalf("output missing %q:\n%s", want, got)
}
}
}
func TestRunAgentPreflightReportsOldGoVersion(t *testing.T) {
deps := preflightDeps{
lookPath: func(name string) (string, error) { return "/usr/bin/" + name, nil },
commandOutput: func(name string, args ...string) ([]byte, error) {
return []byte("go version go1.23.9 linux/amd64\n"), nil
},
executable: func() (string, error) { return "/usr/local/bin/micro", nil },
getenv: func(key string) string {
if key == "ANTHROPIC_API_KEY" {
return "set"
}
return ""
},
listen: func(network, address string) (net.Listener, error) { return stubListener{}, nil },
}
var out bytes.Buffer
err := runAgentPreflight(&out, deps)
if err == nil {
t.Fatal("runAgentPreflight() error = nil")
}
got := out.String()
for _, want := range []string{"✗ Go toolchain", "go1.23.9", "Upgrade to Go 1.24 or newer", "Rerun micro agent preflight"} {
if !strings.Contains(got, want) {
t.Fatalf("output missing %q:\n%s", want, got)
}
}
}
func TestGoVersionAtLeast(t *testing.T) {
tests := []struct {
line string
want bool
}{
{line: "go version go1.24.0 linux/amd64", want: true},
{line: "go version go1.25.1 linux/amd64", want: true},
{line: "go version go1.23.9 linux/amd64", want: false},
{line: "unexpected", want: false},
}
for _, tt := range tests {
if got := goVersionAtLeast(tt.line, 1, 24); got != tt.want {
t.Fatalf("goVersionAtLeast(%q) = %v, want %v", tt.line, got, tt.want)
}
}
}
func TestFirstLine(t *testing.T) {
if got := firstLine([]byte("one\ntwo")); got != "one" {
t.Fatalf("firstLine() = %q", got)
+35
View File
@@ -24,6 +24,30 @@ import (
_ "go-micro.dev/v6/cmd/micro/cli/remote"
)
const docsWayfinding = `First-agent and 0→hero docs:
1. No-secret first-agent transcript
https://go-micro.dev/docs/guides/no-secret-first-agent.html
Run the maintained support agent without a provider key:
go test ./internal/harness/zero-to-hero-ci -run TestNoSecretFirstAgentTranscript -count=1
2. Your First Agent
https://go-micro.dev/docs/guides/your-first-agent.html
Build a service-backed agent, then use:
micro agent preflight
micro run
micro chat
3. Debugging your agent
https://go-micro.dev/docs/guides/debugging-agents.html
Inspect agent runs and memory with:
micro inspect agent
micro runs <agent>
4. 0→hero Reference
https://go-micro.dev/docs/guides/zero-to-hero.html
Walk the scaffold → run → chat → inspect → deploy dry-run lifecycle.`
func genProtoHandler(c *cli.Context) error {
cmd := exec.Command("find", ".", "-name", "*.proto", "-exec", "protoc", "--proto_path=.", "--micro_out=.", "--go_out=.", `{}`, `;`)
cmd.Stdout = os.Stdout
@@ -96,6 +120,17 @@ func init() {
return nil
},
},
{
Name: "docs",
Usage: "Show the first-agent and 0→hero documentation path",
Description: `Print the maintained adoption on-ramp for new Go Micro developers:
the no-secret first-agent transcript, Your First Agent, debugging guide, and
0→hero lifecycle reference.`,
Action: func(ctx *cli.Context) error {
fmt.Fprintln(ctx.App.Writer, docsWayfinding)
return nil
},
},
{
Name: "call",
Usage: "Call a service",
+27 -1
View File
@@ -1,6 +1,7 @@
package main
import (
"bytes"
"strings"
"testing"
@@ -21,7 +22,7 @@ func TestFirstAgentWalkthroughCLIBoundaries(t *testing.T) {
}
}
for _, want := range []string{"new", "run", "chat", "inspect", "agent"} {
for _, want := range []string{"new", "run", "chat", "inspect", "agent", "docs"} {
if !commands[want] {
t.Fatalf("first-agent walkthrough missing %q command", want)
}
@@ -37,6 +38,31 @@ func TestFirstAgentWalkthroughCLIBoundaries(t *testing.T) {
if !strings.Contains(chat.Description, "services") || !strings.Contains(chat.Description, "agent") {
t.Fatalf("micro chat should describe the service-to-agent walkthrough boundary; description was %q", chat.Description)
}
docs := commandByName(t, "docs")
if !strings.Contains(docs.Usage, "first-agent") || !strings.Contains(docs.Usage, "0→hero") {
t.Fatalf("micro docs should advertise the first-agent and 0→hero docs path; usage was %q", docs.Usage)
}
var out bytes.Buffer
app := cli.NewApp()
app.Writer = &out
if err := docs.Action(cli.NewContext(app, nil, nil)); err != nil {
t.Fatalf("micro docs failed: %v", err)
}
for _, want := range []string{
"no-secret-first-agent.html",
"your-first-agent.html",
"debugging-agents.html",
"zero-to-hero.html",
"micro agent preflight",
"micro run",
"micro chat",
"micro inspect agent",
} {
if !strings.Contains(out.String(), want) {
t.Fatalf("micro docs output missing %q:\n%s", want, out.String())
}
}
}
func commandByName(t *testing.T, name string) *cli.Command {
+71 -14
View File
@@ -148,13 +148,17 @@ func main() {
fmt.Fprintf(os.Stderr, "content-type = %q, want text/event-stream\n", ct)
os.Exit(1)
}
payload, err := readSSEData(res.Body)
summary, err := readSSESummary(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)
if summary.State != "completed" {
fmt.Fprintf(os.Stderr, "stream final state = %q, want completed; payload: %s\n", summary.State, summary.Payload)
os.Exit(1)
}
if !summary.HasArtifactText {
fmt.Fprintf(os.Stderr, "stream completed without artifact text: %s\n", summary.Payload)
os.Exit(1)
}
if !sawTool || !sawRunInfo {
@@ -164,24 +168,77 @@ func main() {
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) {
type streamSummary struct {
Payload string
State string
HasArtifactText bool
}
func readSSESummary(r io.Reader) (streamSummary, error) {
scanner := bufio.NewScanner(r)
var payload strings.Builder
var event strings.Builder
var summary streamSummary
seen := false
flush := func() error {
data := strings.TrimSpace(event.String())
event.Reset()
if data == "" {
return nil
}
var envelope struct {
Result struct {
Status struct {
State string `json:"state"`
} `json:"status"`
Artifacts []struct {
Parts []struct {
Text string `json:"text"`
} `json:"parts"`
} `json:"artifacts"`
} `json:"result"`
}
if err := json.Unmarshal([]byte(data), &envelope); err != nil {
return fmt.Errorf("SSE data event is not JSON: %s", data)
}
seen = true
summary.Payload += data + "\n"
if envelope.Result.Status.State != "" {
summary.State = envelope.Result.Status.State
}
for _, artifact := range envelope.Result.Artifacts {
for _, part := range artifact.Parts {
if strings.TrimSpace(part.Text) != "" {
summary.HasArtifactText = true
}
}
}
return nil
}
for scanner.Scan() {
line := scanner.Text()
if data, ok := strings.CutPrefix(line, "data: "); ok {
payload.WriteString(data)
payload.WriteByte('\n')
if strings.TrimSpace(line) == "" {
if err := flush(); err != nil {
return streamSummary{}, err
}
continue
}
data, ok := strings.CutPrefix(line, "data:")
if !ok {
continue
}
if event.Len() > 0 {
event.WriteByte('\n')
}
event.WriteString(strings.TrimSpace(data))
}
if err := scanner.Err(); err != nil {
return "", err
return streamSummary{}, err
}
if payload.Len() == 0 {
return "", errors.New("no SSE data received")
if err := flush(); err != nil {
return streamSummary{}, err
}
if !json.Valid([]byte(strings.TrimSpace(payload.String()))) {
return "", fmt.Errorf("SSE data is not JSON: %s", payload.String())
if !seen {
return streamSummary{}, errors.New("no SSE data received")
}
return payload.String(), nil
return summary, nil
}
@@ -0,0 +1,30 @@
package main
import (
"strings"
"testing"
)
func TestReadSSESummaryUsesCompletedTaskInvariants(t *testing.T) {
summary, err := readSSESummary(strings.NewReader("data: {\"jsonrpc\":\"2.0\",\"result\":{\"status\":{\"state\":\"working\"}}}\n\n" +
"data: {\"jsonrpc\":\"2.0\",\"result\":{\"status\":{\"state\":\"completed\"},\"artifacts\":[{\"parts\":[{\"kind\":\"text\",\"text\":\"provider-specific answer\"}]}]}}\n\n"))
if err != nil {
t.Fatalf("readSSESummary() error = %v", err)
}
if summary.State != "completed" {
t.Fatalf("State = %q, want completed", summary.State)
}
if !summary.HasArtifactText {
t.Fatal("HasArtifactText = false, want true")
}
if strings.Contains(summary.Payload, "a2a-fallback-ok") {
t.Fatalf("test fixture should not rely on marker text: %s", summary.Payload)
}
}
func TestReadSSESummaryRejectsNonJSONData(t *testing.T) {
_, err := readSSESummary(strings.NewReader("data: not-json\n\n"))
if err == nil {
t.Fatal("readSSESummary() error = nil, want non-JSON error")
}
}
+46 -6
View File
@@ -127,6 +127,10 @@ func (s *TaskService) count() int {
return len(s.tasks)
}
const delegatedNotifyTask = "Use the notify Send tool exactly once to tell owner@acme.com: The launch plan is ready. Do not answer until the notify tool call has succeeded."
const commsPrompt = "You handle outbound notifications. When asked to notify someone, you must call the notify Send tool exactly once before replying. Never claim a notification was sent unless the notify tool returned success."
type SendRequest struct {
To string `json:"to" description:"Recipient address"`
Message string `json:"message" description:"Message body"`
@@ -275,12 +279,12 @@ func (m *mockModel) Generate(ctx context.Context, req *ai.Request, _ ...ai.Gener
if m.unknownDelegateOnce && !m.emittedUnknownDelegate {
m.emittedUnknownDelegate = true
m.call("conductor", "atlascloud_delegate", map[string]any{
"task": "Notify owner@acme.com that the launch plan is ready",
"task": delegatedNotifyTask,
"to": "comms",
})
} else {
m.call("conductor", del, map[string]any{
"task": "Notify owner@acme.com that the launch plan is ready",
"task": delegatedNotifyTask,
"to": "comms",
})
}
@@ -354,7 +358,7 @@ func runPlanDelegate(provider string) error {
agent.Name("comms"),
agent.Address("127.0.0.1:0"),
agent.Services("notify"),
agent.Prompt("You handle outbound notifications. Use the notify service."),
agent.Prompt(commsPrompt),
agent.Provider(provider), agent.APIKey(apiKey),
agent.WithRegistry(reg), agent.WithClient(cl), agent.WithStore(mem),
agent.WithCheckpoint(commsCheckpoint),
@@ -413,7 +417,10 @@ func runPlanDelegate(provider string) error {
executeDone <- f.Execute(ctx, "launch readiness")
}()
if err := waitForPlanDelegateExecution(executeDone, notifySvc); err != nil {
if err := waitForPlanDelegateExecution(executeDone, taskSvc, notifySvc, func(ctx context.Context) error {
_, err := conductor.Ask(ctx, "The Design, Build, and Ship tasks already exist, but the owner notification is still missing. Delegate exactly one notification to the \"comms\" agent now with this exact subtask: "+delegatedNotifyTask+" Do not create more tasks and do not answer until comms has handled the notification.")
return err
}); err != nil {
return err
}
@@ -435,14 +442,38 @@ func runPlanDelegate(provider string) error {
return nil
}
func waitForPlanDelegateExecution(done <-chan error, notifySvc *NotifyService) error {
func waitForPlanDelegateExecution(done <-chan error, taskSvc *TaskService, notifySvc *NotifyService, recoverMissingNotify func(context.Context) error) error {
ticker := time.NewTicker(50 * time.Millisecond)
defer ticker.Stop()
for {
select {
case err := <-done:
tasks := taskSvc.count()
notify := notifySvc.count()
if err != nil {
return fmt.Errorf("flow execute: %w", err)
if isClientTimeout(err) {
if tasks == 3 && notify == 1 {
fmt.Printf("\n\033[33mwarning:\033[0m flow execute returned after completed side effects: %v\n", err)
return nil
}
return classifiedPlanDelegateTimeout(tasks, notify, err)
}
return fmt.Errorf("flow execute after side effects tasks=%d notify=%d: %w", tasks, notify, err)
}
if notify != 1 {
if recoverMissingNotify == nil || tasks != 3 || notify != 0 {
return fmt.Errorf("delegation completed without required notify side effect: notify=%d, want 1", notify)
}
fmt.Print("\n\033[33mwarning:\033[0m flow completed before delegated notify; retrying the missing comms handoff once.\n")
ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
retryErr := recoverMissingNotify(ctx)
cancel()
if retryErr != nil {
return fmt.Errorf("delegation completed without required notify side effect and recovery failed: notify=%d, want 1: %w", notify, retryErr)
}
if notify = notifySvc.count(); notify != 1 {
return fmt.Errorf("delegation recovery completed without required notify side effect: notify=%d, want 1", notify)
}
}
return nil
case <-ticker.C:
@@ -453,6 +484,15 @@ func waitForPlanDelegateExecution(done <-chan error, notifySvc *NotifyService) e
}
}
func classifiedPlanDelegateTimeout(tasks, notify int, err error) error {
return fmt.Errorf("provider latency/outage during plan-delegate before required side effects completed (tasks=%d/3 notify=%d/1); retry live provider or inspect provider logs if this recurs: %w", tasks, notify, err)
}
func isClientTimeout(err error) bool {
msg := strings.ToLower(err.Error())
return strings.Contains(msg, "request timeout") || strings.Contains(msg, "code=408") || strings.Contains(msg, "code\":408")
}
func main() {
provider := flag.String("provider", "mock", "LLM provider: mock (default), mock-unknown-delegate, anthropic, openai, gemini, groq, mistral, together, atlascloud")
flag.Parse()
+109 -3
View File
@@ -2,6 +2,7 @@ package main
import (
"context"
"errors"
"strings"
"testing"
"time"
@@ -66,7 +67,7 @@ func TestPlanDelegateEndToEnd(t *testing.T) {
agent.Name("comms"),
agent.Address("127.0.0.1:0"),
agent.Services("notify"),
agent.Prompt("You handle outbound notifications."),
agent.Prompt(commsPrompt),
agent.Provider("mock"),
agent.WithRegistry(reg),
agent.WithClient(cl),
@@ -148,7 +149,7 @@ func TestFlowDispatchesToAgentEndToEnd(t *testing.T) {
agent.Name("comms"),
agent.Address("127.0.0.1:0"),
agent.Services("notify"),
agent.Prompt("You handle outbound notifications."),
agent.Prompt(commsPrompt),
agent.Provider("mock"),
agent.WithRegistry(reg),
agent.WithClient(cl),
@@ -257,7 +258,7 @@ func TestPlanDelegateExecutionReportsDuplicateNotifyBeforeTimeout(t *testing.T)
done := make(chan error)
errCh := make(chan error, 1)
go func() { errCh <- waitForPlanDelegateExecution(done, notifySvc) }()
go func() { errCh <- waitForPlanDelegateExecution(done, new(TaskService), notifySvc, nil) }()
select {
case err := <-errCh:
@@ -272,6 +273,111 @@ func TestPlanDelegateExecutionReportsDuplicateNotifyBeforeTimeout(t *testing.T)
}
}
func TestPlanDelegateExecutionRejectsClaimedCompletionWithoutNotify(t *testing.T) {
notifySvc := new(NotifyService)
done := make(chan error, 1)
done <- nil
err := waitForPlanDelegateExecution(done, new(TaskService), notifySvc, nil)
if err == nil {
t.Fatal("waitForPlanDelegateExecution returned nil, want missing notify side-effect error")
}
if got := err.Error(); !strings.Contains(got, "without required notify side effect") {
t.Fatalf("error = %q, want missing notify side-effect error", got)
}
}
func TestPlanDelegateExecutionRecoversMissingNotifyOnce(t *testing.T) {
taskSvc := new(TaskService)
for _, title := range []string{"Design", "Build", "Ship"} {
var rsp AddResponse
if err := taskSvc.Add(context.Background(), &AddRequest{Title: title}, &rsp); err != nil {
t.Fatalf("Add(%q): %v", title, err)
}
}
notifySvc := new(NotifyService)
done := make(chan error, 1)
done <- nil
recovered := false
err := waitForPlanDelegateExecution(done, taskSvc, notifySvc, func(ctx context.Context) error {
recovered = true
var rsp SendResponse
return notifySvc.Send(ctx, &SendRequest{To: "owner@acme.com", Message: "The launch plan is ready"}, &rsp)
})
if err != nil {
t.Fatalf("waitForPlanDelegateExecution returned %v, want recovery success", err)
}
if !recovered {
t.Fatal("missing notify recovery was not invoked")
}
if got := notifySvc.count(); got != 1 {
t.Fatalf("notify count = %d, want 1 after recovery", got)
}
}
func TestPlanDelegateExecutionAcceptsClientTimeoutAfterSideEffects(t *testing.T) {
taskSvc := new(TaskService)
for _, title := range []string{"Design", "Build", "Ship"} {
var rsp AddResponse
if err := taskSvc.Add(context.Background(), &AddRequest{Title: title}, &rsp); err != nil {
t.Fatalf("Add(%q): %v", title, err)
}
}
notifySvc := new(NotifyService)
var rsp SendResponse
if err := notifySvc.Send(context.Background(), &SendRequest{To: "owner@acme.com", Message: "The launch plan is ready"}, &rsp); err != nil {
t.Fatalf("Send: %v", err)
}
done := make(chan error, 1)
done <- errors.New(`{"id":"go.micro.client","code":408,"detail":"<nil>","status":"Request Timeout"}`)
if err := waitForPlanDelegateExecution(done, taskSvc, notifySvc, nil); err != nil {
t.Fatalf("waitForPlanDelegateExecution returned %v, want completed side effects to satisfy client timeout", err)
}
}
func TestPlanDelegateExecutionClassifiesClientTimeoutBeforeSideEffects(t *testing.T) {
done := make(chan error, 1)
done <- errors.New(`{"id":"go.micro.client","code":408,"detail":"<nil>","status":"Request Timeout"}`)
err := waitForPlanDelegateExecution(done, new(TaskService), new(NotifyService), nil)
if err == nil {
t.Fatal("waitForPlanDelegateExecution returned nil, want timeout before side effects to fail")
}
for _, want := range []string{
"provider latency/outage during plan-delegate",
"tasks=0/3 notify=0/1",
"retry live provider or inspect provider logs",
"Request Timeout",
} {
if got := err.Error(); !strings.Contains(got, want) {
t.Fatalf("error = %q, want %q", got, want)
}
}
}
func TestPlanDelegateExecutionClassifiesPartialClientTimeout(t *testing.T) {
taskSvc := new(TaskService)
for _, title := range []string{"Design", "Build", "Ship"} {
var rsp AddResponse
if err := taskSvc.Add(context.Background(), &AddRequest{Title: title}, &rsp); err != nil {
t.Fatalf("Add(%q): %v", title, err)
}
}
done := make(chan error, 1)
done <- errors.New(`{"id":"go.micro.client","code":408,"detail":"<nil>","status":"Request Timeout"}`)
err := waitForPlanDelegateExecution(done, taskSvc, new(NotifyService), nil)
if err == nil {
t.Fatal("waitForPlanDelegateExecution returned nil, want timeout before notify to fail")
}
if got := err.Error(); !strings.Contains(got, "tasks=3/3 notify=0/1") {
t.Fatalf("error = %q, want partial side-effect counts", got)
}
}
func TestNotifyServiceSendIsIdempotentForDuplicateDelivery(t *testing.T) {
svc := new(NotifyService)
for i := 0; i < 3; i++ {
@@ -51,6 +51,7 @@ Provider keys are read from `MICRO_AI_API_KEY` or the provider-specific variable
| OpenAI | `OPENAI_API_KEY` |
| Gemini | `GEMINI_API_KEY` |
| Groq | `GROQ_API_KEY` |
| MiniMax | `MINIMAX_API_KEY` |
| Mistral | `MISTRAL_API_KEY` |
| Together | `TOGETHER_API_KEY` |
| AtlasCloud | `ATLASCLOUD_API_KEY` |
@@ -29,6 +29,7 @@ import (
_ "go-micro.dev/v6/ai/atlascloud"
_ "go-micro.dev/v6/ai/gemini"
_ "go-micro.dev/v6/ai/groq"
_ "go-micro.dev/v6/ai/minimax"
_ "go-micro.dev/v6/ai/mistral"
_ "go-micro.dev/v6/ai/openai"
_ "go-micro.dev/v6/ai/together"
@@ -49,6 +50,7 @@ var providerEnv = map[string]string{
"openai": "OPENAI_API_KEY",
"gemini": "GEMINI_API_KEY",
"groq": "GROQ_API_KEY",
"minimax": "MINIMAX_API_KEY",
"mistral": "MISTRAL_API_KEY",
"together": "TOGETHER_API_KEY",
"atlascloud": "ATLASCLOUD_API_KEY",
@@ -38,7 +38,7 @@ func TestValidateSelectionRejectsUnsafeHarnessName(t *testing.T) {
func TestDefaultProvidersTracksLiveProviderSet(t *testing.T) {
got := defaultProviders()
for _, want := range []string{"anthropic", "openai", "gemini", "groq", "mistral", "together", "atlascloud"} {
for _, want := range []string{"anthropic", "openai", "gemini", "groq", "minimax", "mistral", "together", "atlascloud"} {
if !strings.Contains(got, want) {
t.Fatalf("defaultProviders() = %q, want %q", got, want)
}
@@ -54,7 +54,7 @@ func TestCapabilityMatrixHasRegisteredProviders(t *testing.T) {
t.Fatal("CapabilityRows returned no providers")
}
var foundOpenAI bool
var foundOpenAI, foundMiniMax bool
for _, row := range rows {
if row.Provider == "openai" {
foundOpenAI = true
@@ -62,10 +62,19 @@ func TestCapabilityMatrixHasRegisteredProviders(t *testing.T) {
t.Fatalf("openai capabilities = %#v, want model+image only", row.Capabilities)
}
}
if row.Provider == "minimax" {
foundMiniMax = true
if !row.Model || !row.Stream || row.Image || row.Video {
t.Fatalf("minimax capabilities = %#v, want model+stream only", row.Capabilities)
}
}
}
if !foundOpenAI {
t.Fatalf("CapabilityRows = %#v, want openai row", rows)
}
if !foundMiniMax {
t.Fatalf("CapabilityRows = %#v, want minimax row", rows)
}
}
func TestWriteCapabilityMarkdown(t *testing.T) {
+69 -12
View File
@@ -27,6 +27,7 @@ package main
import (
"context"
"encoding/json"
"flag"
"fmt"
"net/http/httptest"
@@ -39,6 +40,8 @@ import (
"go-micro.dev/v6/agent"
"go-micro.dev/v6/ai"
"go-micro.dev/v6/broker"
"go-micro.dev/v6/client"
codecbytes "go-micro.dev/v6/codec/bytes"
"go-micro.dev/v6/flow"
"go-micro.dev/v6/gateway/a2a"
"go-micro.dev/v6/internal/harness/harnessutil"
@@ -109,9 +112,10 @@ type SendResponse struct {
}
type Notify struct {
mu sync.Mutex
sent int64
seen map[string]struct{}
mu sync.Mutex
sent int64
seen map[string]struct{}
lastRejected *SendRequest
}
// Send delivers a notification.
@@ -122,11 +126,14 @@ func (s *Notify) Send(_ context.Context, req *SendRequest, rsp *SendResponse) er
if req != nil {
to, message = req.To, req.Message
}
s.recordRejected(to, message)
fmt.Printf(" \033[35m[notify]\033[0m 📨 ignored non-buyer notification to=%s %q\n", to, message)
rsp.Sent = false
return nil
}
s.recordRejected("", "")
keys := notificationDedupeKeys(req)
s.mu.Lock()
if s.seen == nil {
@@ -158,12 +165,34 @@ func isBuyerNotification(req *SendRequest) bool {
return canonicalBuyerRecipient(req.To) != ""
}
func (s *Notify) recordRejected(to, message string) {
s.mu.Lock()
defer s.mu.Unlock()
if strings.TrimSpace(to) == "" && strings.TrimSpace(message) == "" {
s.lastRejected = nil
return
}
s.lastRejected = &SendRequest{To: to, Message: message}
}
func (s *Notify) rejectedSummary() string {
s.mu.Lock()
defer s.mu.Unlock()
if s.lastRejected == nil {
return "no rejected notify call observed"
}
return fmt.Sprintf("last notify args to=%q message=%q", s.lastRejected.To, s.lastRejected.Message)
}
func canonicalBuyerRecipient(to string) string {
recipient := strings.ToLower(strings.TrimSpace(to))
switch recipient {
case "buyer", "buyer@acme.com":
return "buyer@acme.com"
}
if strings.HasPrefix(recipient, "buyer-of-order-") && len(recipient) > len("buyer-of-order-") {
return "buyer@acme.com"
}
for _, field := range strings.FieldsFunc(recipient, func(r rune) bool {
switch r {
case ' ', '\t', '\n', '\r', ',', ';', ':', '/', '\\', '(', ')', '[', ']', '{', '}':
@@ -199,18 +228,40 @@ func notificationDedupeKeys(req *SendRequest) []string {
return keys
}
func dispatchNotifyStep(agentName string, ntf *Notify) flow.StepFunc {
dispatch := flow.Dispatch(agentName)
func dispatchNotifyStep(agentName string, cl client.Client, ntf *Notify) flow.StepFunc {
return func(ctx context.Context, in flow.State) (flow.State, error) {
before := atomic.LoadInt64(&ntf.sent)
out, err := dispatch(ctx, in)
if err == nil {
return out, nil
out, err := dispatchBuyerNotification(ctx, agentName, cl, in)
if err != nil {
out = in
}
return completeNotifyOnObservedSideEffect(ctx, in, ntf, before, 2*time.Second, err)
return completeNotifyOnObservedSideEffect(ctx, out, ntf, before, 2*time.Second, err)
}
}
func dispatchBuyerNotification(ctx context.Context, agentName string, cl client.Client, in flow.State) (flow.State, error) {
if cl == nil {
cl = client.DefaultClient
}
info, _ := ai.RunInfoFrom(ctx)
message := fmt.Sprintf(
"Checkout flow confirmed this order: %s. Use notify.Send exactly once to notify buyer@acme.com that the order is confirmed. Do not reply until the notify tool call has completed.",
strings.TrimSpace(in.String()),
)
body, _ := json.Marshal(map[string]string{"message": message, "parent_id": info.RunID})
req := cl.NewRequest(agentName, "Agent.Chat", &codecbytes.Frame{Data: body})
var rsp codecbytes.Frame
if err := cl.Call(ctx, req, &rsp); err != nil {
return in, err
}
var out struct {
Reply string `json:"reply"`
}
_ = json.Unmarshal(rsp.Data, &out)
in.Data = []byte(out.Reply)
return in, nil
}
func completeNotifyOnObservedSideEffect(ctx context.Context, in flow.State, ntf *Notify, before int64, wait time.Duration, dispatchErr error) (flow.State, error) {
deadline := time.Now().Add(wait)
for time.Now().Before(deadline) {
@@ -220,11 +271,17 @@ func completeNotifyOnObservedSideEffect(ctx context.Context, in flow.State, ntf
}
select {
case <-ctx.Done():
return in, dispatchErr
if dispatchErr != nil {
return in, dispatchErr
}
return in, ctx.Err()
case <-time.After(25 * time.Millisecond):
}
}
return in, dispatchErr
if dispatchErr != nil {
return in, dispatchErr
}
return in, fmt.Errorf("concierge completed without notifying buyer: notify count stayed at %d; expected recipient buyer@acme.com, buyer, or buyer-of-order-<id>; %s", before, ntf.rejectedSummary())
}
// ---------------------------------------------------------------------------
@@ -397,7 +454,7 @@ func runUniverse(provider string) int {
flow.Step{Name: "reserve", Run: flow.Call("inventory", "Inventory.Reserve")},
flow.Step{Name: "charge", Run: flow.Call("payment", "Payment.Charge")},
flow.Step{Name: "confirm", Run: flow.Call("orders", "Orders.Confirm")},
flow.Step{Name: "notify", Run: dispatchNotifyStep("concierge", ntf)},
flow.Step{Name: "notify", Run: dispatchNotifyStep("concierge", cl, ntf)},
),
)
if err := checkout.Register(reg, br, cl); err != nil {
+49 -1
View File
@@ -69,6 +69,27 @@ func TestNotifyStepCompletesAfterObservedSideEffectTimeout(t *testing.T) {
}
}
func TestNotifyStepRejectsClaimedCompletionWithoutSideEffect(t *testing.T) {
ntf := new(Notify)
before := atomic.LoadInt64(&ntf.sent)
_, err := completeNotifyOnObservedSideEffect(
context.Background(),
flow.State{Data: []byte(`claimed success`)},
ntf,
before,
25*time.Millisecond,
nil,
)
if err == nil {
t.Fatal("notify completion returned nil, want missing buyer notification error")
}
want := `concierge completed without notifying buyer: notify count stayed at 0; expected recipient buyer@acme.com, buyer, or buyer-of-order-<id>; no rejected notify call observed`
if got := err.Error(); got != want {
t.Fatalf("error = %q, want %q", got, want)
}
}
func TestNotifySuppressesEquivalentConfirmationMessages(t *testing.T) {
ntf := new(Notify)
ctx := context.Background()
@@ -161,7 +182,7 @@ func TestNotifyAcceptsOrderScopedBuyerRecipient(t *testing.T) {
var rsp SendResponse
if err := ntf.Send(ctx, &SendRequest{
To: "order-1 buyer",
To: "buyer-of-order-1",
Message: "order-1 confirmed",
}, &rsp); err != nil {
t.Fatalf("send order-scoped buyer notification: %v", err)
@@ -186,3 +207,30 @@ func TestNotifyAcceptsOrderScopedBuyerRecipient(t *testing.T) {
t.Fatalf("notifications sent after hyphenated non-buyer = %d, want 1", got)
}
}
func TestNotifyStepReportsRejectedRecipientDiagnostics(t *testing.T) {
ntf := new(Notify)
var rsp SendResponse
if err := ntf.Send(context.Background(), &SendRequest{
To: "order-1",
Message: "order-1 confirmed",
}, &rsp); err != nil {
t.Fatalf("send rejected notification: %v", err)
}
_, err := completeNotifyOnObservedSideEffect(
context.Background(),
flow.State{Data: []byte(`claimed success`)},
ntf,
0,
25*time.Millisecond,
nil,
)
if err == nil {
t.Fatal("notify completion returned nil, want diagnostics")
}
want := `concierge completed without notifying buyer: notify count stayed at 0; expected recipient buyer@acme.com, buyer, or buyer-of-order-<id>; last notify args to="order-1" message="order-1 confirmed"`
if got := err.Error(); got != want {
t.Fatalf("error = %q, want %q", got, want)
}
}
+1 -1
View File
@@ -25,7 +25,7 @@ Before your first provider-backed agent run, check the local path with:
micro agent preflight
```
The preflight is read-only: it verifies Go, the `micro` binary, provider-key setup, and whether the default `micro run` gateway port is free, without calling an LLM provider.
The preflight is read-only: it verifies Go 1.24+, the `micro` binary, provider-key setup, and whether the default `micro run` gateway port is free, without calling an LLM provider. When a check fails it prints the exact fix plus the next guide to open, so the scaffold → run → chat path stays walkable.
## Install
@@ -42,6 +42,7 @@ The built-in providers currently register these capability interfaces:
| `atlascloud` | Yes | Yes | Yes | Yes |
| `gemini` | Yes | No | No | No |
| `groq` | Yes | No | No | Yes |
| `minimax` | Yes | No | No | Yes |
| `mistral` | Yes | No | No | Yes |
| `ollama` | Yes | No | No | Yes |
| `openai` | Yes | Yes | No | Yes |
@@ -17,7 +17,9 @@ micro inspect ... # read the recorded run or workflow history
Debug the lifecycle in the same order Go Micro runs it: first prove the service is
registered and callable, then inspect the agent run that chose tools, then inspect
any workflow that handed off to the agent.
any workflow that handed off to the agent. If the first local run fails before a
chat turn, run `micro agent preflight`; failed checks include `Fix:` and `Next:`
lines for Go, CLI installation, provider-key setup, and the local gateway port.
## 1. Reproduce one small turn
@@ -89,6 +89,6 @@ CI keeps those CLI boundaries present with:
go test ./cmd/micro -run TestFirstAgentWalkthroughCLIBoundaries -count=1
```
If chat behaves unexpectedly, continue to
If `micro agent preflight` reports a missing provider key, you can still use this no-secret path because it runs against the mock model; the command now prints this guide as the next step for that failure. If chat behaves unexpectedly, continue to
[Debugging your agent](debugging-agents.html) for provider checks, run history,
memory, and tool-call inspection.
@@ -13,6 +13,7 @@ import (
_ "go-micro.dev/v6/ai/atlascloud"
_ "go-micro.dev/v6/ai/gemini"
_ "go-micro.dev/v6/ai/groq"
_ "go-micro.dev/v6/ai/minimax"
_ "go-micro.dev/v6/ai/mistral"
_ "go-micro.dev/v6/ai/ollama"
_ "go-micro.dev/v6/ai/openai"
@@ -53,7 +53,7 @@ Run the read-only first-agent preflight before starting the walkthrough. The sam
micro agent preflight
```
It checks Go, the `micro` binary, provider-key setup, and the default local gateway port without contacting a provider.
It checks Go 1.24+, the `micro` binary, provider-key setup, and the default local gateway port without contacting a provider. Failed checks include a `Fix:` line and a `Next:` line that points back to this guide, the no-secret walkthrough, or the debugging guide.
## 1. Create a workspace
+39 -54
View File
@@ -3,7 +3,6 @@ package store
import (
"fmt"
"os"
"path/filepath"
"strings"
"testing"
"time"
@@ -12,15 +11,20 @@ import (
"github.com/kr/pretty"
)
func cleanup(db string, s Store) {
s.Close()
dir := filepath.Join(DefaultDir, db+"/")
os.RemoveAll(dir)
func newTestFileStore(t *testing.T, opts ...Option) Store {
t.Helper()
opts = append(opts, DirOption(t.TempDir()))
s := NewStore(opts...)
t.Cleanup(func() {
if err := s.Close(); err != nil {
t.Errorf("failed to close file store: %v", err)
}
})
return s
}
func TestFileStoreReInit(t *testing.T) {
s := NewStore(Table("aaa"))
defer cleanup(DefaultDatabase, s)
s := newTestFileStore(t, Table("aaa"))
s.Init(Table("bbb"))
if s.Options().Table != "bbb" {
t.Error("Init didn't reinitialise the store")
@@ -28,26 +32,22 @@ func TestFileStoreReInit(t *testing.T) {
}
func TestFileStoreBasic(t *testing.T) {
s := NewStore()
defer cleanup(DefaultDatabase, s)
s := newTestFileStore(t)
fileTest(s, t)
}
func TestFileStoreTable(t *testing.T) {
s := NewStore(Table("testTable"))
defer cleanup(DefaultDatabase, s)
s := newTestFileStore(t, Table("testTable"))
fileTest(s, t)
}
func TestFileStoreDatabase(t *testing.T) {
s := NewStore(Database("testdb"))
defer cleanup("testdb", s)
s := newTestFileStore(t, Database("testdb"))
fileTest(s, t)
}
func TestFileStoreDatabaseTable(t *testing.T) {
s := NewStore(Table("testTable"), Database("testdb"))
defer cleanup("testdb", s)
s := newTestFileStore(t, Table("testTable"), Database("testdb"))
fileTest(s, t)
}
@@ -94,7 +94,7 @@ func fileTest(s Store, t *testing.T) {
{
Key: "foobar",
Value: []byte("foobarfoobar"),
Expiry: time.Millisecond * 100,
Expiry: time.Second, // wide window: CI I/O under -race can exceed a 100ms expiry before the read below
},
}
@@ -113,8 +113,8 @@ func fileTest(s Store, t *testing.T) {
}
}
// wait for the expiry
time.Sleep(time.Millisecond * 200)
// wait for the expiry (must exceed the 1s Expiry above, with margin for slow CI)
time.Sleep(time.Second * 2)
if results, err := s.Read("foo", ReadPrefix()); err != nil {
t.Errorf("Couldn't read all \"foo\" keys, got %# v (%s)", spew.Sdump(results), err)
@@ -135,22 +135,22 @@ func fileTest(s Store, t *testing.T) {
}
}
// Write 3 records with various expiry and get with Suffix
// Write records with suffix matches and an already-expired record. Avoid
// wall-clock boundary sleeps here: under -race/-cover, sleeping exactly the
// TTL made this assertion flaky on slower CI runners.
records = []*Record{
{
Key: "foo",
Value: []byte("foofoo"),
},
{
Key: "barfoo",
Value: []byte("barfoobarfoo"),
Expiry: time.Millisecond * 100,
Key: "barfoo",
Value: []byte("barfoobarfoo"),
Expiry: -time.Second,
},
{
Key: "bazbarfoo",
Value: []byte("bazbarfoobazbarfoo"),
Expiry: 2 * time.Millisecond * 100,
Key: "bazbarfoo",
Value: []byte("bazbarfoobazbarfoo"),
},
}
for _, r := range records {
@@ -160,59 +160,44 @@ func fileTest(s Store, t *testing.T) {
}
if results, err := s.Read("foo", ReadSuffix()); err != nil {
t.Errorf("Couldn't read all \"foo\" keys, got %# v (%s)", spew.Sdump(results), err)
} else {
if len(results) != 3 {
t.Errorf("Expected 3 items, got %d", len(results))
// t.Logf("Table test: %v\n", spew.Sdump(results))
}
} else if len(results) != 2 {
t.Errorf("Expected 2 unexpired suffix items, got %d (%# v)", len(results), spew.Sdump(results))
}
if err := s.Delete("bazbarfoo"); err != nil {
t.Errorf("Delete failed (%v)", err)
}
time.Sleep(time.Millisecond * 100)
if results, err := s.Read("foo", ReadSuffix()); err != nil {
t.Errorf("Couldn't read all \"foo\" keys, got %# v (%s)", spew.Sdump(results), err)
} else {
if len(results) != 2 {
t.Errorf("Expected 2 items, got %d", len(results))
// t.Logf("Table test: %v\n", spew.Sdump(results))
}
}
time.Sleep(time.Millisecond * 100)
if results, err := s.Read("foo", ReadSuffix()); err != nil {
t.Errorf("Couldn't read all \"foo\" keys, got %# v (%s)", spew.Sdump(results), err)
} else {
if len(results) != 1 {
t.Errorf("Expected 1 item, got %d", len(results))
// t.Logf("Table test: %# v\n", spew.Sdump(results))
}
} else if len(results) != 1 {
t.Errorf("Expected 1 unexpired suffix item, got %d (%# v)", len(results), spew.Sdump(results))
}
if err := s.Delete("foo"); err != nil {
t.Errorf("Delete failed (%v)", err)
}
if results, err := s.Read("foo", ReadSuffix()); err != nil {
t.Errorf("Couldn't read all \"foo\" keys, got %# v (%s)", spew.Sdump(results), err)
} else {
if len(results) != 0 {
t.Errorf("Expected 0 items, got %d (%# v)", len(results), spew.Sdump(results))
}
} else if len(results) != 0 {
t.Errorf("Expected 0 items, got %d (%# v)", len(results), spew.Sdump(results))
}
// Test Table, Suffix and WriteOptions
if err := s.Write(&Record{
Key: "foofoobarbar",
Value: []byte("something"),
}, WriteTTL(time.Millisecond*100)); err != nil {
}, WriteTTL(time.Second)); err != nil {
t.Error(err)
}
if err := s.Write(&Record{
Key: "foofoo",
Value: []byte("something"),
}, WriteExpiry(time.Now().Add(time.Millisecond*100))); err != nil {
}, WriteExpiry(time.Now().Add(time.Second))); err != nil {
t.Error(err)
}
if err := s.Write(&Record{
Key: "barbar",
Value: []byte("something"),
// TTL has higher precedence than expiry
}, WriteExpiry(time.Now().Add(time.Hour)), WriteTTL(time.Millisecond*100)); err != nil {
}, WriteExpiry(time.Now().Add(time.Hour)), WriteTTL(time.Second)); err != nil {
t.Error(err)
}
@@ -224,7 +209,7 @@ func fileTest(s Store, t *testing.T) {
}
}
time.Sleep(time.Millisecond * 100)
time.Sleep(time.Second * 2) // exceed the 1s TTL/expiry above so everything has expired
if results, err := s.List(); err != nil {
t.Errorf("List failed: %s", err)