Compare commits
1 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| 52aa089618 |
@@ -21,8 +21,12 @@ changes, architectural rewrites. Those go to the human.
|
||||
|
||||
## Work queue (ranked)
|
||||
|
||||
1. **Trace agent `RunInfo` into OpenTelemetry spans** ([#3867](https://github.com/micro/go-micro/issues/3867)) — #3857/#3865 closed the streaming queue item, there are no open `codex` PRs in flight, and the remaining Next-phase gap with the strongest mission fit is operability of the services → agents → workflows lifecycle. The README, website, roadmap, and recent blog all promise that agents are services you can run, chat with, inspect, and trust under the same runtime; projecting run steps, tool calls, delegation, streaming, cancellation, and errors into OTel spans makes that lived path debuggable without a public-API rewrite. Keep this additive/no-op when tracing is disabled, with tests proving representative `RunInfo` events become trace data.
|
||||
2. **Add an AP2 mandate layer over A2A and x402** ([#3552](https://github.com/micro/go-micro/issues/3552)) — this remains a forward interop investment, not a Now/Next 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 ahead of adoption, resilience, streaming, or durable agent operation. Keep it additive and opt-in while the AP2/FIDO work settles.
|
||||
1. **Require real notification side effects in AtlasCloud plan/delegate conformance** ([#3711](https://github.com/micro/go-micro/issues/3711)) — PR #3713 shipped the no-secret first-agent transcript, so the adoption on-ramp item is done and drops out of the queue. The highest-value Now-phase gap is now live-provider correctness: AtlasCloud/Minimax can emit an unavailable provider-specific tool call while the conductor marks delegation complete, which breaks the services → agents → workflows promise more directly than parser polish. Make the harness fail/retry until the registered delegate/comms notification path actually executes, with a CI-safe regression for unknown tool calls.
|
||||
2. **Expose `fallback_echo` during A2A streaming fallback conformance** ([#3560](https://github.com/micro/go-micro/issues/3560)) — this remains the next Now-phase interop/conformance gap: it protects the A2A streaming promise developers see in the README and site by ensuring the non-native streaming fallback path receives the tool surface, while staying scoped and CI-verifiable.
|
||||
3. **Parse multi-event A2A SSE fallback responses in the harness** ([#3662](https://github.com/micro/go-micro/issues/3662)) — once the fallback tool path succeeds, the harness must accept legitimate multi-event `message/stream` responses instead of concatenating valid SSE events into invalid JSON. This is 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.
|
||||
|
||||
_Seeded by Claude Code from the roadmap + open issues; thereafter maintained by the
|
||||
architecture-review pass._
|
||||
|
||||
@@ -18,7 +18,7 @@ on:
|
||||
providers:
|
||||
description: "Comma-separated providers for live conformance (default: all supported)"
|
||||
required: false
|
||||
default: "anthropic,openai,gemini,groq,minimax,mistral,together,atlascloud"
|
||||
default: "anthropic,openai,gemini,groq,mistral,together,atlascloud"
|
||||
harnesses:
|
||||
description: "Comma-separated harnesses for live conformance"
|
||||
required: false
|
||||
@@ -64,7 +64,6 @@ 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 }}
|
||||
@@ -74,7 +73,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,minimax,mistral,together,atlascloud' }}"
|
||||
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' }}"
|
||||
|
||||
|
||||
@@ -16,30 +16,9 @@ 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
|
||||
|
||||
@@ -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, …) |
|
||||
| 9 LLM providers | Anthropic, OpenAI, Gemini, Groq, Mistral, Together, Atlas Cloud, MiniMax, Ollama (local + cloud) |
|
||||
| 8 LLM providers | Anthropic, OpenAI, Gemini, Groq, Mistral, Together, Atlas Cloud, 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,7 +429,6 @@ 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
|
||||
@@ -443,7 +442,6 @@ resp, _ := m.Generate(ctx, &ai.Request{Prompt: "hello"})
|
||||
- [multi-service](examples/multi-service/) — Multiple services in one binary
|
||||
- [mcp](examples/mcp/) — MCP integration with AI agents
|
||||
- [agent-plan-delegate](examples/agent-plan-delegate/) — Agent planning and multi-agent delegation
|
||||
- [agent-durable](examples/agent-durable/) — Checkpoint and resume an agent run without replaying completed tool side effects
|
||||
- [grpc-interop](examples/grpc-interop/) — Call go-micro from any gRPC client
|
||||
|
||||
See [all examples](examples/README.md).
|
||||
|
||||
+1
-5
@@ -222,16 +222,12 @@ func (a *agentImpl) Stream(ctx context.Context, message string) (ai.Stream, erro
|
||||
return nil, fmt.Errorf("discover tools: %w", err)
|
||||
}
|
||||
a.mem.Add("user", message)
|
||||
stream, err := a.model.Stream(ctx, &ai.Request{
|
||||
return a.model.Stream(ctx, &ai.Request{
|
||||
Prompt: message,
|
||||
SystemPrompt: a.buildPrompt(),
|
||||
Tools: toolList,
|
||||
Messages: a.mem.Messages(),
|
||||
})
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return &memoryRecordingStream{stream: stream, memory: a.mem}, nil
|
||||
}
|
||||
|
||||
// Pending returns checkpointed agent runs that have not completed. It mirrors
|
||||
|
||||
@@ -2,7 +2,6 @@ package agent
|
||||
|
||||
import (
|
||||
"context"
|
||||
"io"
|
||||
"strings"
|
||||
"testing"
|
||||
|
||||
@@ -17,7 +16,6 @@ import (
|
||||
// it with a deferred cleanup. Tests in this package are not parallel,
|
||||
// so a package-level hook is safe.
|
||||
var fakeGen func(ctx context.Context, opts ai.Options, req *ai.Request) (*ai.Response, error)
|
||||
var fakeStream func(ctx context.Context, opts ai.Options, req *ai.Request) (ai.Stream, error)
|
||||
|
||||
type fakeModel struct{ opts ai.Options }
|
||||
|
||||
@@ -35,33 +33,10 @@ func (m *fakeModel) Generate(ctx context.Context, req *ai.Request, _ ...ai.Gener
|
||||
return &ai.Response{Reply: "ok"}, nil
|
||||
}
|
||||
func (m *fakeModel) Stream(ctx context.Context, req *ai.Request, _ ...ai.GenerateOption) (ai.Stream, error) {
|
||||
if fakeStream != nil {
|
||||
return fakeStream(ctx, m.opts, req)
|
||||
}
|
||||
return &sliceStream{chunks: []string{"ok"}}, nil
|
||||
return nil, nil
|
||||
}
|
||||
func (m *fakeModel) String() string { return "fake" }
|
||||
|
||||
type sliceStream struct {
|
||||
chunks []string
|
||||
idx int
|
||||
closed bool
|
||||
}
|
||||
|
||||
func (s *sliceStream) Recv() (*ai.Response, error) {
|
||||
if s.idx >= len(s.chunks) {
|
||||
return nil, io.EOF
|
||||
}
|
||||
chunk := s.chunks[s.idx]
|
||||
s.idx++
|
||||
return &ai.Response{Reply: chunk}, nil
|
||||
}
|
||||
|
||||
func (s *sliceStream) Close() error {
|
||||
s.closed = true
|
||||
return nil
|
||||
}
|
||||
|
||||
func init() {
|
||||
ai.Register("fake", func(opts ...ai.Option) ai.Model {
|
||||
m := &fakeModel{}
|
||||
|
||||
@@ -45,7 +45,6 @@ const (
|
||||
AttrFlowStep = "agent.flow.step"
|
||||
AttrDispatch = "agent.dispatch"
|
||||
AttrTrigger = "agent.trigger"
|
||||
AttrRunEventKind = "agent.event.kind"
|
||||
)
|
||||
|
||||
type RunEvent struct {
|
||||
@@ -324,7 +323,6 @@ func runEventAttributes(e RunEvent) []attribute.KeyValue {
|
||||
attrs := []attribute.KeyValue{
|
||||
attribute.String(AttrRunID, e.RunID),
|
||||
attribute.String(AttrAgentName, e.Agent),
|
||||
attribute.String(AttrRunEventKind, e.Kind),
|
||||
}
|
||||
if e.ParentID != "" {
|
||||
attrs = append(attrs, attribute.String(AttrParentRunID, e.ParentID))
|
||||
|
||||
+1
-2
@@ -279,8 +279,7 @@ func spanEventHasRunInfo(events []trace.Event, name, runID, agentName string) bo
|
||||
continue
|
||||
}
|
||||
attrs := spanAttributes(event.Attributes)
|
||||
wantKind := strings.TrimPrefix(name, "agent.")
|
||||
if attrs[AttrRunID] == runID && attrs[AttrAgentName] == agentName && attrs[AttrRunEventKind] == wantKind {
|
||||
if attrs[AttrRunID] == runID && attrs[AttrAgentName] == agentName {
|
||||
return true
|
||||
}
|
||||
}
|
||||
|
||||
@@ -132,34 +132,6 @@ 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
|
||||
|
||||
@@ -185,45 +185,6 @@ type agentStreamAdapter struct {
|
||||
stream AgentStream
|
||||
}
|
||||
|
||||
type memoryRecordingStream struct {
|
||||
stream ai.Stream
|
||||
memory Memory
|
||||
|
||||
mu sync.Mutex
|
||||
chunks []string
|
||||
closed bool
|
||||
}
|
||||
|
||||
func (s *memoryRecordingStream) Recv() (*ai.Response, error) {
|
||||
resp, err := s.stream.Recv()
|
||||
if resp != nil && resp.Reply != "" {
|
||||
s.mu.Lock()
|
||||
s.chunks = append(s.chunks, resp.Reply)
|
||||
s.mu.Unlock()
|
||||
}
|
||||
if errors.Is(err, io.EOF) {
|
||||
s.recordAssistant()
|
||||
}
|
||||
return resp, err
|
||||
}
|
||||
|
||||
func (s *memoryRecordingStream) Close() error {
|
||||
s.recordAssistant()
|
||||
return s.stream.Close()
|
||||
}
|
||||
|
||||
func (s *memoryRecordingStream) recordAssistant() {
|
||||
s.mu.Lock()
|
||||
defer s.mu.Unlock()
|
||||
if s.closed {
|
||||
return
|
||||
}
|
||||
s.closed = true
|
||||
if reply := strings.Join(s.chunks, ""); reply != "" {
|
||||
s.memory.Add("assistant", reply)
|
||||
}
|
||||
}
|
||||
|
||||
func (s *agentStreamAdapter) Recv() (*ai.Response, error) {
|
||||
for {
|
||||
event, err := s.stream.Recv()
|
||||
|
||||
@@ -82,52 +82,6 @@ func TestStreamAskHelperRejectsUnsupportedAgent(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
func TestAgentStreamUsesProviderStreamingAndRecordsAssistantMemory(t *testing.T) {
|
||||
var sawRequest bool
|
||||
fakeStream = func(ctx context.Context, opts ai.Options, req *ai.Request) (ai.Stream, error) {
|
||||
sawRequest = true
|
||||
if req.Prompt != "stream the answer" {
|
||||
t.Fatalf("Prompt = %q, want stream the answer", req.Prompt)
|
||||
}
|
||||
if len(req.Messages) != 1 || req.Messages[0].Role != "user" || req.Messages[0].Content != "stream the answer" {
|
||||
t.Fatalf("Messages = %#v, want current user turn in memory", req.Messages)
|
||||
}
|
||||
return &sliceStream{chunks: []string{"hel", "lo"}}, nil
|
||||
}
|
||||
defer func() { fakeStream = nil }()
|
||||
|
||||
a := newTestAgent(Name("provider-stream"))
|
||||
stream, err := a.Stream(context.Background(), "stream the answer")
|
||||
if err != nil {
|
||||
t.Fatalf("Stream: %v", err)
|
||||
}
|
||||
|
||||
var reply string
|
||||
for {
|
||||
chunk, err := stream.Recv()
|
||||
if errors.Is(err, io.EOF) {
|
||||
break
|
||||
}
|
||||
if err != nil {
|
||||
t.Fatalf("Recv: %v", err)
|
||||
}
|
||||
reply += chunk.Reply
|
||||
}
|
||||
if err := stream.Close(); err != nil {
|
||||
t.Fatalf("Close: %v", err)
|
||||
}
|
||||
if !sawRequest {
|
||||
t.Fatal("provider Stream was not called")
|
||||
}
|
||||
if reply != "hello" {
|
||||
t.Fatalf("reply = %q, want hello", reply)
|
||||
}
|
||||
got := a.mem.Messages()
|
||||
if len(got) != 2 || got[0].Role != "user" || got[0].Content != "stream the answer" || got[1].Role != "assistant" || got[1].Content != "hello" {
|
||||
t.Fatalf("memory = %#v, want user turn and streamed assistant reply", got)
|
||||
}
|
||||
}
|
||||
|
||||
func TestResumeStreamAskDoesNotReplayCompletedTool(t *testing.T) {
|
||||
ctx := context.Background()
|
||||
cp := flow.StoreCheckpoint(store.NewStore(), "stream-resume-agent")
|
||||
|
||||
@@ -300,20 +300,6 @@ 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:
|
||||
|
||||
+14
-86
@@ -52,15 +52,6 @@ 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...)
|
||||
@@ -130,7 +121,7 @@ func (p *Provider) Generate(ctx context.Context, req *ai.Request, opts ...ai.Gen
|
||||
apiReq["tools"] = tools
|
||||
}
|
||||
|
||||
resp, rawMessage, err := p.callAPI(ctx, "chat", apiReq)
|
||||
resp, rawMessage, err := p.callAPI(ctx, apiReq)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
@@ -140,7 +131,6 @@ func (p *Provider) Generate(ctx context.Context, req *ai.Request, opts ...ai.Gen
|
||||
}
|
||||
|
||||
if p.opts.ToolHandler != nil {
|
||||
var toolResults []string
|
||||
followUpMessages := append(messages, map[string]any{
|
||||
"role": "assistant",
|
||||
"content": rawMessage["content"],
|
||||
@@ -149,9 +139,6 @@ func (p *Provider) Generate(ctx context.Context, req *ai.Request, opts ...ai.Gen
|
||||
|
||||
for _, tc := range resp.ToolCalls {
|
||||
content := p.opts.ToolHandler(ctx, tc).Content
|
||||
if content != "" {
|
||||
toolResults = append(toolResults, content)
|
||||
}
|
||||
followUpMessages = append(followUpMessages, map[string]any{
|
||||
"role": "tool",
|
||||
"tool_call_id": tc.ID,
|
||||
@@ -164,14 +151,9 @@ func (p *Provider) Generate(ctx context.Context, req *ai.Request, opts ...ai.Gen
|
||||
"messages": followUpMessages,
|
||||
}
|
||||
|
||||
followUpResp, _, err := p.callAPI(ctx, "tool-follow-up", followUpReq)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if followUpResp.Reply != "" {
|
||||
followUpResp, _, err := p.callAPI(ctx, followUpReq)
|
||||
if err == nil && followUpResp.Reply != "" {
|
||||
resp.Answer = followUpResp.Reply
|
||||
} else if len(toolResults) > 0 {
|
||||
resp.Answer = strings.Join(toolResults, "\n")
|
||||
}
|
||||
}
|
||||
|
||||
@@ -181,10 +163,6 @@ func (p *Provider) Generate(ctx context.Context, req *ai.Request, opts ...ai.Gen
|
||||
// Stream generates a streaming response from Atlas Cloud's OpenAI-compatible
|
||||
// chat completions endpoint, emitting content deltas as they arrive.
|
||||
func (p *Provider) Stream(ctx context.Context, req *ai.Request, opts ...ai.GenerateOption) (ai.Stream, error) {
|
||||
if len(req.Tools) > 0 {
|
||||
return nil, fmt.Errorf("%w: atlascloud streaming does not expose tools", ai.ErrStreamingUnsupported)
|
||||
}
|
||||
|
||||
messages := []map[string]any{
|
||||
{"role": "system", "content": req.SystemPrompt},
|
||||
}
|
||||
@@ -289,7 +267,7 @@ func (s *atlasStream) Close() error {
|
||||
return s.body.Close()
|
||||
}
|
||||
|
||||
func (p *Provider) callAPI(ctx context.Context, phase string, req map[string]any) (*ai.Response, map[string]any, error) {
|
||||
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)
|
||||
@@ -312,14 +290,20 @@ func (p *Provider) callAPI(ctx context.Context, phase string, req map[string]any
|
||||
|
||||
respBody, _ := io.ReadAll(httpResp.Body)
|
||||
if httpResp.StatusCode != http.StatusOK {
|
||||
return nil, nil, fmt.Errorf("API error (%s) during atlascloud %s request (%s): %s", httpResp.Status, phase, atlascloudRequestSummary(req), string(respBody))
|
||||
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 []atlasToolCall `json:"tool_calls"`
|
||||
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"`
|
||||
}
|
||||
@@ -351,68 +335,12 @@ func (p *Provider) callAPI(ctx context.Context, phase string, req map[string]any
|
||||
|
||||
rawMessage := map[string]any{
|
||||
"content": choice.Message.Content,
|
||||
"tool_calls": normalizeAtlasCloudToolCalls(choice.Message.ToolCalls),
|
||||
"tool_calls": 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.
|
||||
|
||||
@@ -7,7 +7,6 @@ import (
|
||||
"io"
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"strings"
|
||||
"testing"
|
||||
|
||||
"go-micro.dev/v6/ai"
|
||||
@@ -141,187 +140,6 @@ func TestProvider_Stream(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
func TestProvider_StreamWithToolsFallsBack(t *testing.T) {
|
||||
p := NewProvider(ai.WithAPIKey("test-key"))
|
||||
_, err := p.Stream(context.Background(), &ai.Request{
|
||||
Prompt: "call a tool",
|
||||
Tools: []ai.Tool{{
|
||||
Name: "fallback_echo",
|
||||
Description: "echo fallback marker",
|
||||
Properties: map[string]any{"value": map[string]any{"type": "string"}},
|
||||
}},
|
||||
})
|
||||
if !errors.Is(err, ai.ErrStreamingUnsupported) {
|
||||
t.Fatalf("Stream with tools error = %v, want ErrStreamingUnsupported", err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestProvider_GenerateToolCallEmptyFollowUpUsesToolResult(t *testing.T) {
|
||||
var calls int
|
||||
ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
if r.URL.Path != "/v1/chat/completions" {
|
||||
t.Errorf("path = %s, want /v1/chat/completions", r.URL.Path)
|
||||
}
|
||||
calls++
|
||||
w.Header().Set("Content-Type", "application/json")
|
||||
switch calls {
|
||||
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":""}}]}`))
|
||||
default:
|
||||
t.Fatalf("unexpected API call %d", calls)
|
||||
}
|
||||
}))
|
||||
defer ts.Close()
|
||||
|
||||
p := NewProvider(
|
||||
ai.WithAPIKey("test-key"),
|
||||
ai.WithBaseURL(ts.URL),
|
||||
ai.WithToolHandler(func(ctx context.Context, call ai.ToolCall) ai.ToolResult {
|
||||
if call.Name != "conformance_echo" {
|
||||
t.Fatalf("tool name = %q, want conformance_echo", call.Name)
|
||||
}
|
||||
return ai.ToolResult{ID: call.ID, Content: `{"marker":"agent-conformance-ok"}`}
|
||||
}),
|
||||
)
|
||||
resp, 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.Fatalf("Generate returned error: %v", err)
|
||||
}
|
||||
if calls != 2 {
|
||||
t.Fatalf("API calls = %d, want 2", calls)
|
||||
}
|
||||
if resp.Answer != `{"marker":"agent-conformance-ok"}` {
|
||||
t.Fatalf("Answer = %q, want tool result fallback", resp.Answer)
|
||||
}
|
||||
}
|
||||
|
||||
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 {
|
||||
|
||||
@@ -9,7 +9,6 @@ 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"
|
||||
@@ -17,7 +16,7 @@ import (
|
||||
|
||||
func TestRegisteredProviders(t *testing.T) {
|
||||
got := ai.RegisteredProviders("")
|
||||
want := []string{"anthropic", "atlascloud", "gemini", "groq", "minimax", "mistral", "openai", "together"}
|
||||
want := []string{"anthropic", "atlascloud", "gemini", "groq", "mistral", "openai", "together"}
|
||||
if !reflect.DeepEqual(got, want) {
|
||||
t.Fatalf("RegisteredProviders() = %#v, want %#v", got, want)
|
||||
}
|
||||
@@ -35,7 +34,7 @@ func TestRegisteredProviders(t *testing.T) {
|
||||
}
|
||||
|
||||
got = ai.RegisteredProviders("stream")
|
||||
want = []string{"atlascloud", "groq", "minimax", "mistral", "openai", "together"}
|
||||
want = []string{"atlascloud", "groq", "mistral", "openai", "together"}
|
||||
if !reflect.DeepEqual(got, want) {
|
||||
t.Fatalf("RegisteredProviders(stream) = %#v, want %#v", got, want)
|
||||
}
|
||||
@@ -48,7 +47,6 @@ 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}},
|
||||
@@ -61,7 +59,7 @@ func TestCapabilityRows(t *testing.T) {
|
||||
func TestCapabilityMatrix(t *testing.T) {
|
||||
matrix := ai.CapabilityMatrix()
|
||||
|
||||
for _, provider := range []string{"anthropic", "atlascloud", "gemini", "groq", "minimax", "mistral", "openai", "together"} {
|
||||
for _, provider := range []string{"anthropic", "atlascloud", "gemini", "groq", "mistral", "openai", "together"} {
|
||||
caps, ok := matrix[provider]
|
||||
if !ok {
|
||||
t.Fatalf("CapabilityMatrix missing %q", provider)
|
||||
@@ -90,7 +88,7 @@ func TestRegisterStream(t *testing.T) {
|
||||
}
|
||||
|
||||
got := ai.RegisteredProviders("stream")
|
||||
want := []string{"atlascloud", "groq", "minimax", "mistral", "openai", "test-stream", "together"}
|
||||
want := []string{"atlascloud", "groq", "mistral", "openai", "test-stream", "together"}
|
||||
if !reflect.DeepEqual(got, want) {
|
||||
t.Fatalf("RegisteredProviders(stream) = %#v, want %#v", got, want)
|
||||
}
|
||||
|
||||
@@ -1,196 +0,0 @@
|
||||
// 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
|
||||
}
|
||||
@@ -1,96 +0,0 @@
|
||||
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())
|
||||
}
|
||||
}
|
||||
@@ -212,8 +212,6 @@ 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"):
|
||||
|
||||
+4
-8
@@ -99,19 +99,15 @@ 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 {
|
||||
|
||||
@@ -18,7 +18,6 @@ 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"
|
||||
@@ -279,7 +278,6 @@ func conformingStreamProviders(t *testing.T) []string {
|
||||
allowed := map[string]struct{}{
|
||||
"atlascloud": {},
|
||||
"groq": {},
|
||||
"minimax": {},
|
||||
"mistral": {},
|
||||
"openai": {},
|
||||
"together": {},
|
||||
@@ -290,7 +288,7 @@ func conformingStreamProviders(t *testing.T) []string {
|
||||
out = append(out, provider)
|
||||
}
|
||||
}
|
||||
want := []string{"atlascloud", "groq", "minimax", "mistral", "openai", "together"}
|
||||
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)
|
||||
}
|
||||
|
||||
@@ -16,7 +16,6 @@ type preflightCheck struct {
|
||||
OK bool
|
||||
Detail string
|
||||
Fix string
|
||||
Next string
|
||||
}
|
||||
|
||||
type preflightDeps struct {
|
||||
@@ -53,9 +52,6 @@ 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)
|
||||
@@ -91,23 +87,19 @@ func agentPreflightChecks(deps preflightDeps) []preflightCheck {
|
||||
func checkGoToolchain(deps preflightDeps) preflightCheck {
|
||||
path, err := deps.lookPath("go")
|
||||
if err != nil {
|
||||
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."}
|
||||
return preflightCheck{Name: "Go toolchain", Fix: "Install Go 1.24 or newer and ensure go is on PATH."}
|
||||
}
|
||||
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 (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", Detail: strings.TrimSpace(string(out)), Fix: "Ensure the go command runs successfully."}
|
||||
}
|
||||
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)}
|
||||
return preflightCheck{Name: "Go toolchain", OK: true, Detail: fmt.Sprintf("%s (%s)", firstLine(out), path)}
|
||||
}
|
||||
|
||||
func checkMicroBinary(deps preflightDeps) preflightCheck {
|
||||
exe, err := deps.executable()
|
||||
if err != nil || exe == "" {
|
||||
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."}
|
||||
return preflightCheck{Name: "micro binary", Fix: "Install the micro CLI or run this check through go run ./cmd/micro agent preflight."}
|
||||
}
|
||||
version := deps.version()
|
||||
if version == "" {
|
||||
@@ -125,7 +117,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.", 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", 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", OK: true, Detail: "found " + strings.Join(found, ", ")}
|
||||
}
|
||||
@@ -133,7 +125,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 + " (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."}
|
||||
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."}
|
||||
}
|
||||
_ = ln.Close()
|
||||
return preflightCheck{Name: "local port " + addr, OK: true, Detail: "available for " + use}
|
||||
@@ -146,18 +138,3 @@ 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
|
||||
}
|
||||
|
||||
@@ -61,59 +61,13 @@ func TestRunAgentPreflightReportsActionableFailures(t *testing.T) {
|
||||
t.Fatal("runAgentPreflight() error = nil")
|
||||
}
|
||||
got := out.String()
|
||||
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"} {
|
||||
for _, want := range []string{"✗ Go toolchain", "Install Go 1.24", "✗ micro binary", "✗ provider API key", "ANTHROPIC_API_KEY", "✗ local port :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)
|
||||
|
||||
@@ -24,30 +24,6 @@ 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
|
||||
@@ -120,17 +96,6 @@ 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",
|
||||
|
||||
@@ -1,7 +1,6 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"strings"
|
||||
"testing"
|
||||
|
||||
@@ -22,7 +21,7 @@ func TestFirstAgentWalkthroughCLIBoundaries(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
for _, want := range []string{"new", "run", "chat", "inspect", "agent", "docs"} {
|
||||
for _, want := range []string{"new", "run", "chat", "inspect", "agent"} {
|
||||
if !commands[want] {
|
||||
t.Fatalf("first-agent walkthrough missing %q command", want)
|
||||
}
|
||||
@@ -38,31 +37,6 @@ 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 {
|
||||
|
||||
@@ -148,17 +148,13 @@ func main() {
|
||||
fmt.Fprintf(os.Stderr, "content-type = %q, want text/event-stream\n", ct)
|
||||
os.Exit(1)
|
||||
}
|
||||
summary, err := readSSESummary(res.Body)
|
||||
payload, err := readSSEData(res.Body)
|
||||
if err != nil {
|
||||
fmt.Fprintln(os.Stderr, err)
|
||||
os.Exit(1)
|
||||
}
|
||||
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)
|
||||
if !strings.Contains(payload, "a2a-fallback-ok") {
|
||||
fmt.Fprintf(os.Stderr, "stream payload missing marker: %s\n", payload)
|
||||
os.Exit(1)
|
||||
}
|
||||
if !sawTool || !sawRunInfo {
|
||||
@@ -168,77 +164,24 @@ func main() {
|
||||
fmt.Println("\n\033[32m✓ A2A message/stream fell back to Ask and preserved tool/run metadata\033[0m")
|
||||
}
|
||||
|
||||
type streamSummary struct {
|
||||
Payload string
|
||||
State string
|
||||
HasArtifactText bool
|
||||
}
|
||||
|
||||
func readSSESummary(r io.Reader) (streamSummary, error) {
|
||||
func readSSEData(r io.Reader) (string, error) {
|
||||
scanner := bufio.NewScanner(r)
|
||||
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
|
||||
}
|
||||
var payload strings.Builder
|
||||
for scanner.Scan() {
|
||||
line := scanner.Text()
|
||||
if strings.TrimSpace(line) == "" {
|
||||
if err := flush(); err != nil {
|
||||
return streamSummary{}, err
|
||||
}
|
||||
continue
|
||||
if data, ok := strings.CutPrefix(line, "data: "); ok {
|
||||
payload.WriteString(data)
|
||||
payload.WriteByte('\n')
|
||||
}
|
||||
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 streamSummary{}, err
|
||||
return "", err
|
||||
}
|
||||
if err := flush(); err != nil {
|
||||
return streamSummary{}, err
|
||||
if payload.Len() == 0 {
|
||||
return "", errors.New("no SSE data received")
|
||||
}
|
||||
if !seen {
|
||||
return streamSummary{}, 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 summary, nil
|
||||
return payload.String(), nil
|
||||
}
|
||||
|
||||
@@ -1,30 +0,0 @@
|
||||
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")
|
||||
}
|
||||
}
|
||||
@@ -127,10 +127,6 @@ 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"`
|
||||
@@ -139,11 +135,9 @@ type SendResponse struct {
|
||||
Sent bool `json:"sent"`
|
||||
}
|
||||
type NotifyService struct {
|
||||
mu sync.Mutex
|
||||
sent int
|
||||
attempts int
|
||||
duplicates int
|
||||
bySend map[string]bool
|
||||
mu sync.Mutex
|
||||
sent int
|
||||
bySend map[string]bool
|
||||
}
|
||||
|
||||
// Send delivers a notification message to a recipient. Duplicate delivery
|
||||
@@ -156,13 +150,11 @@ func (s *NotifyService) Send(ctx context.Context, req *SendRequest, rsp *SendRes
|
||||
s.bySend = map[string]bool{}
|
||||
}
|
||||
key := strings.ToLower(strings.TrimSpace(req.To)) + "\x00" + strings.ToLower(strings.TrimSpace(req.Message))
|
||||
s.attempts++
|
||||
if !s.bySend[key] {
|
||||
s.bySend[key] = true
|
||||
s.sent++
|
||||
fmt.Printf(" \033[35m[notify]\033[0m 📨 to=%s message=%q\n", req.To, req.Message)
|
||||
} else {
|
||||
s.duplicates++
|
||||
fmt.Printf(" \033[35m[notify]\033[0m reused to=%s message=%q\n", req.To, req.Message)
|
||||
}
|
||||
s.mu.Unlock()
|
||||
@@ -176,28 +168,13 @@ func (s *NotifyService) count() int {
|
||||
return s.sent
|
||||
}
|
||||
|
||||
func (s *NotifyService) duplicateAttempts() int {
|
||||
s.mu.Lock()
|
||||
defer s.mu.Unlock()
|
||||
return s.duplicates
|
||||
}
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// mock LLM provider — the ONLY fake. It "reasons" by simple heuristics
|
||||
// over the tools it's offered and the system prompt it's given, calling
|
||||
// the real tool handler exactly the way a real provider would.
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
type mockModel struct {
|
||||
opts ai.Options
|
||||
|
||||
// unknownDelegateOnce makes the mock emit one provider-style, unavailable
|
||||
// delegate tool name before using the registered delegate tool. This mirrors
|
||||
// live providers that occasionally hallucinate a provider-specific tool while
|
||||
// still keeping the regression deterministic and keyless.
|
||||
unknownDelegateOnce bool
|
||||
emittedUnknownDelegate bool
|
||||
}
|
||||
type mockModel struct{ opts ai.Options }
|
||||
|
||||
func newMock(opts ...ai.Option) ai.Model {
|
||||
m := &mockModel{}
|
||||
@@ -205,12 +182,6 @@ func newMock(opts ...ai.Option) ai.Model {
|
||||
return m
|
||||
}
|
||||
|
||||
func newMockUnknownDelegate(opts ...ai.Option) ai.Model {
|
||||
m := &mockModel{unknownDelegateOnce: true}
|
||||
_ = m.Init(opts...)
|
||||
return m
|
||||
}
|
||||
|
||||
func (m *mockModel) Init(opts ...ai.Option) error {
|
||||
for _, o := range opts {
|
||||
o(&m.opts)
|
||||
@@ -276,18 +247,10 @@ func (m *mockModel) Generate(ctx context.Context, req *ai.Request, _ ...ai.Gener
|
||||
}
|
||||
}
|
||||
if del := findTool(req.Tools, "delegate"); del != "" {
|
||||
if m.unknownDelegateOnce && !m.emittedUnknownDelegate {
|
||||
m.emittedUnknownDelegate = true
|
||||
m.call("conductor", "atlascloud_delegate", map[string]any{
|
||||
"task": delegatedNotifyTask,
|
||||
"to": "comms",
|
||||
})
|
||||
} else {
|
||||
m.call("conductor", del, map[string]any{
|
||||
"task": delegatedNotifyTask,
|
||||
"to": "comms",
|
||||
})
|
||||
}
|
||||
m.call("conductor", del, map[string]any{
|
||||
"task": "Notify owner@acme.com that the launch plan is ready",
|
||||
"to": "comms",
|
||||
})
|
||||
}
|
||||
return &ai.Response{Answer: "Created Design, Build and Ship, and had comms notify the owner."}, nil
|
||||
|
||||
@@ -315,12 +278,9 @@ func providerKey(provider string) string {
|
||||
|
||||
func runPlanDelegate(provider string) error {
|
||||
apiKey := ""
|
||||
switch provider {
|
||||
case "mock":
|
||||
if provider == "mock" {
|
||||
ai.Register("mock", newMock)
|
||||
case "mock-unknown-delegate":
|
||||
ai.Register("mock-unknown-delegate", newMockUnknownDelegate)
|
||||
default:
|
||||
} 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)
|
||||
@@ -358,7 +318,7 @@ func runPlanDelegate(provider string) error {
|
||||
agent.Name("comms"),
|
||||
agent.Address("127.0.0.1:0"),
|
||||
agent.Services("notify"),
|
||||
agent.Prompt(commsPrompt),
|
||||
agent.Prompt("You handle outbound notifications. Use the notify service."),
|
||||
agent.Provider(provider), agent.APIKey(apiKey),
|
||||
agent.WithRegistry(reg), agent.WithClient(cl), agent.WithStore(mem),
|
||||
agent.WithCheckpoint(commsCheckpoint),
|
||||
@@ -410,18 +370,8 @@ func runPlanDelegate(provider string) error {
|
||||
}
|
||||
|
||||
fmt.Print("\n\033[1m> flow:\033[0m services + agents + workflow + plan/delegate, no API key.\n\n")
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
defer cancel()
|
||||
executeDone := make(chan error, 1)
|
||||
go func() {
|
||||
executeDone <- f.Execute(ctx, "launch readiness")
|
||||
}()
|
||||
|
||||
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
|
||||
if err := f.Execute(context.Background(), "launch readiness"); err != nil {
|
||||
return fmt.Errorf("flow execute: %w", err)
|
||||
}
|
||||
|
||||
if rs := f.Results(); len(rs) > 0 {
|
||||
@@ -442,59 +392,8 @@ func runPlanDelegate(provider string) error {
|
||||
return nil
|
||||
}
|
||||
|
||||
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 {
|
||||
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:
|
||||
if dup := notifySvc.duplicateAttempts(); dup > 0 {
|
||||
return fmt.Errorf("duplicate notify attempts: got %d duplicate replay(s), want 0", dup)
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
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")
|
||||
provider := flag.String("provider", "mock", "LLM provider: mock (default), anthropic, openai, gemini, groq, mistral, together, atlascloud")
|
||||
flag.Parse()
|
||||
|
||||
if err := runPlanDelegate(*provider); err != nil {
|
||||
|
||||
@@ -2,8 +2,6 @@ package main
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
@@ -67,7 +65,7 @@ func TestPlanDelegateEndToEnd(t *testing.T) {
|
||||
agent.Name("comms"),
|
||||
agent.Address("127.0.0.1:0"),
|
||||
agent.Services("notify"),
|
||||
agent.Prompt(commsPrompt),
|
||||
agent.Prompt("You handle outbound notifications."),
|
||||
agent.Provider("mock"),
|
||||
agent.WithRegistry(reg),
|
||||
agent.WithClient(cl),
|
||||
@@ -149,7 +147,7 @@ func TestFlowDispatchesToAgentEndToEnd(t *testing.T) {
|
||||
agent.Name("comms"),
|
||||
agent.Address("127.0.0.1:0"),
|
||||
agent.Services("notify"),
|
||||
agent.Prompt(commsPrompt),
|
||||
agent.Prompt("You handle outbound notifications."),
|
||||
agent.Provider("mock"),
|
||||
agent.WithRegistry(reg),
|
||||
agent.WithClient(cl),
|
||||
@@ -222,15 +220,6 @@ func TestZeroToHeroContract(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
func TestPlanDelegateRetriesAfterUnknownDelegateTool(t *testing.T) {
|
||||
if testing.Short() {
|
||||
t.Skip("0→hero harness boots an end-to-end system; skipped with -short")
|
||||
}
|
||||
if err := runPlanDelegate("mock-unknown-delegate"); err != nil {
|
||||
t.Fatalf("0→hero harness with unknown delegate retry: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestTaskServiceAddIsIdempotentForLaunchTitles(t *testing.T) {
|
||||
svc := new(TaskService)
|
||||
for _, title := range []string{"Design", "design task", "Build", "Build launch task", "Ship", "ship readiness"} {
|
||||
@@ -247,137 +236,6 @@ func TestTaskServiceAddIsIdempotentForLaunchTitles(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
func TestPlanDelegateExecutionReportsDuplicateNotifyBeforeTimeout(t *testing.T) {
|
||||
notifySvc := new(NotifyService)
|
||||
for i := 0; i < 2; i++ {
|
||||
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 attempt %d: %v", i+1, err)
|
||||
}
|
||||
}
|
||||
|
||||
done := make(chan error)
|
||||
errCh := make(chan error, 1)
|
||||
go func() { errCh <- waitForPlanDelegateExecution(done, new(TaskService), notifySvc, nil) }()
|
||||
|
||||
select {
|
||||
case err := <-errCh:
|
||||
if err == nil {
|
||||
t.Fatal("waitForPlanDelegateExecution returned nil, want duplicate notify error")
|
||||
}
|
||||
if got := err.Error(); !strings.Contains(got, "duplicate notify attempts") {
|
||||
t.Fatalf("error = %q, want duplicate notify attempts", got)
|
||||
}
|
||||
case <-time.After(time.Second):
|
||||
t.Fatal("waitForPlanDelegateExecution did not report duplicate notify before timeout")
|
||||
}
|
||||
}
|
||||
|
||||
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,7 +51,6 @@ 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,7 +29,6 @@ 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"
|
||||
@@ -50,7 +49,6 @@ 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", "minimax", "mistral", "together", "atlascloud"} {
|
||||
for _, want := range []string{"anthropic", "openai", "gemini", "groq", "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, foundMiniMax bool
|
||||
var foundOpenAI bool
|
||||
for _, row := range rows {
|
||||
if row.Provider == "openai" {
|
||||
foundOpenAI = true
|
||||
@@ -62,19 +62,10 @@ 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) {
|
||||
|
||||
@@ -27,7 +27,6 @@ package main
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"flag"
|
||||
"fmt"
|
||||
"net/http/httptest"
|
||||
@@ -40,8 +39,6 @@ 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"
|
||||
@@ -112,10 +109,9 @@ type SendResponse struct {
|
||||
}
|
||||
|
||||
type Notify struct {
|
||||
mu sync.Mutex
|
||||
sent int64
|
||||
seen map[string]struct{}
|
||||
lastRejected *SendRequest
|
||||
mu sync.Mutex
|
||||
sent int64
|
||||
seen map[string]struct{}
|
||||
}
|
||||
|
||||
// Send delivers a notification.
|
||||
@@ -126,14 +122,11 @@ 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 {
|
||||
@@ -165,48 +158,13 @@ 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 {
|
||||
switch strings.ToLower(strings.TrimSpace(to)) {
|
||||
case "buyer", "buyer@acme.com":
|
||||
return "buyer@acme.com"
|
||||
default:
|
||||
return ""
|
||||
}
|
||||
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', ',', ';', ':', '/', '\\', '(', ')', '[', ']', '{', '}':
|
||||
return true
|
||||
default:
|
||||
return false
|
||||
}
|
||||
}) {
|
||||
switch field {
|
||||
case "buyer", "buyer@acme.com":
|
||||
return "buyer@acme.com"
|
||||
}
|
||||
}
|
||||
return ""
|
||||
}
|
||||
|
||||
func notificationDedupeKeys(req *SendRequest) []string {
|
||||
@@ -228,40 +186,18 @@ func notificationDedupeKeys(req *SendRequest) []string {
|
||||
return keys
|
||||
}
|
||||
|
||||
func dispatchNotifyStep(agentName string, cl client.Client, ntf *Notify) flow.StepFunc {
|
||||
func dispatchNotifyStep(agentName string, ntf *Notify) flow.StepFunc {
|
||||
dispatch := flow.Dispatch(agentName)
|
||||
return func(ctx context.Context, in flow.State) (flow.State, error) {
|
||||
before := atomic.LoadInt64(&ntf.sent)
|
||||
out, err := dispatchBuyerNotification(ctx, agentName, cl, in)
|
||||
if err != nil {
|
||||
out = in
|
||||
out, err := dispatch(ctx, in)
|
||||
if err == nil {
|
||||
return out, nil
|
||||
}
|
||||
return completeNotifyOnObservedSideEffect(ctx, out, ntf, before, 2*time.Second, err)
|
||||
return completeNotifyOnObservedSideEffect(ctx, in, 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) {
|
||||
@@ -271,17 +207,11 @@ func completeNotifyOnObservedSideEffect(ctx context.Context, in flow.State, ntf
|
||||
}
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
if dispatchErr != nil {
|
||||
return in, dispatchErr
|
||||
}
|
||||
return in, ctx.Err()
|
||||
return in, dispatchErr
|
||||
case <-time.After(25 * time.Millisecond):
|
||||
}
|
||||
}
|
||||
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())
|
||||
return in, dispatchErr
|
||||
}
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
@@ -454,7 +384,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", cl, ntf)},
|
||||
flow.Step{Name: "notify", Run: dispatchNotifyStep("concierge", ntf)},
|
||||
),
|
||||
)
|
||||
if err := checkout.Register(reg, br, cl); err != nil {
|
||||
|
||||
@@ -69,27 +69,6 @@ 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()
|
||||
@@ -175,62 +154,3 @@ func TestNotifyIgnoresNonBuyerRecipients(t *testing.T) {
|
||||
t.Fatalf("buyer notifications sent = %d, want 1", got)
|
||||
}
|
||||
}
|
||||
|
||||
func TestNotifyAcceptsOrderScopedBuyerRecipient(t *testing.T) {
|
||||
ntf := new(Notify)
|
||||
ctx := context.Background()
|
||||
|
||||
var rsp SendResponse
|
||||
if err := ntf.Send(ctx, &SendRequest{
|
||||
To: "buyer-of-order-1",
|
||||
Message: "order-1 confirmed",
|
||||
}, &rsp); err != nil {
|
||||
t.Fatalf("send order-scoped buyer notification: %v", err)
|
||||
}
|
||||
if !rsp.Sent {
|
||||
t.Fatal("order-scoped buyer notification did not report sent")
|
||||
}
|
||||
if got := atomic.LoadInt64(&ntf.sent); got != 1 {
|
||||
t.Fatalf("order-scoped buyer notifications sent = %d, want 1", got)
|
||||
}
|
||||
|
||||
if err := ntf.Send(ctx, &SendRequest{
|
||||
To: "non-buyer",
|
||||
Message: "order-1 confirmed",
|
||||
}, &rsp); err != nil {
|
||||
t.Fatalf("send hyphenated non-buyer notification: %v", err)
|
||||
}
|
||||
if rsp.Sent {
|
||||
t.Fatal("hyphenated non-buyer notification reported sent")
|
||||
}
|
||||
if got := atomic.LoadInt64(&ntf.sent); got != 1 {
|
||||
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)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -16,7 +16,6 @@ agents → workflows lifecycle.
|
||||
| First service-backed agent | [`examples/agent-demo`](https://github.com/micro/go-micro/tree/master/examples/agent-demo) | Multi-service project/task/team app with agent playground integration. |
|
||||
| 0→hero lifecycle | [`examples/support`](https://github.com/micro/go-micro/tree/master/examples/support) | No-secret support-desk story: typed services, an agent, an event-driven flow, and a guardrail. |
|
||||
| Planning and delegation | [`examples/agent-plan-delegate`](https://github.com/micro/go-micro/tree/master/examples/agent-plan-delegate) | Two agents collaborate through `plan` and `delegate` over normal Go Micro RPC. |
|
||||
| Durable agent runs | [`examples/agent-durable`](https://github.com/micro/go-micro/tree/master/examples/agent-durable) | Checkpoint and resume a model-directed run without replaying completed tool side effects. |
|
||||
| Durable workflows | [`examples/flow-durable`](https://github.com/micro/go-micro/tree/master/examples/flow-durable) | Ordered, checkpointed flow steps resume without duplicating completed side effects. |
|
||||
| AI-callable services | [`examples/mcp`](https://github.com/micro/go-micro/tree/master/examples/mcp) | MCP examples that expose service endpoints as model tools. |
|
||||
|
||||
@@ -36,11 +35,7 @@ agents → workflows lifecycle.
|
||||
[`examples/agent-plan-delegate`](https://github.com/micro/go-micro/tree/master/examples/agent-plan-delegate).
|
||||
- [Agents and Workflows](../guides/agents-and-workflows.html) → run
|
||||
[`examples/flow-durable`](https://github.com/micro/go-micro/tree/master/examples/flow-durable)
|
||||
for deterministic checkpointed steps,
|
||||
[`examples/agent-durable`](https://github.com/micro/go-micro/tree/master/examples/agent-durable)
|
||||
for model-directed checkpointed runs, and
|
||||
[`examples/support`](https://github.com/micro/go-micro/tree/master/examples/support)
|
||||
for the full services → agents → workflows lifecycle.
|
||||
and [`examples/support`](https://github.com/micro/go-micro/tree/master/examples/support).
|
||||
|
||||
## Repository examples
|
||||
|
||||
|
||||
@@ -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 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.
|
||||
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.
|
||||
|
||||
## Install
|
||||
|
||||
|
||||
@@ -42,7 +42,6 @@ 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,9 +17,7 @@ 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. 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.
|
||||
any workflow that handed off to the agent.
|
||||
|
||||
## 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 `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
|
||||
If chat behaves unexpectedly, continue to
|
||||
[Debugging your agent](debugging-agents.html) for provider checks, run history,
|
||||
memory, and tool-call inspection.
|
||||
|
||||
@@ -13,7 +13,6 @@ 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 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.
|
||||
It checks Go, the `micro` binary, provider-key setup, and the default local gateway port without contacting a provider.
|
||||
|
||||
## 1. Create a workspace
|
||||
|
||||
|
||||
+54
-39
@@ -3,6 +3,7 @@ package store
|
||||
import (
|
||||
"fmt"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
@@ -11,20 +12,15 @@ import (
|
||||
"github.com/kr/pretty"
|
||||
)
|
||||
|
||||
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 cleanup(db string, s Store) {
|
||||
s.Close()
|
||||
dir := filepath.Join(DefaultDir, db+"/")
|
||||
os.RemoveAll(dir)
|
||||
}
|
||||
|
||||
func TestFileStoreReInit(t *testing.T) {
|
||||
s := newTestFileStore(t, Table("aaa"))
|
||||
s := NewStore(Table("aaa"))
|
||||
defer cleanup(DefaultDatabase, s)
|
||||
s.Init(Table("bbb"))
|
||||
if s.Options().Table != "bbb" {
|
||||
t.Error("Init didn't reinitialise the store")
|
||||
@@ -32,22 +28,26 @@ func TestFileStoreReInit(t *testing.T) {
|
||||
}
|
||||
|
||||
func TestFileStoreBasic(t *testing.T) {
|
||||
s := newTestFileStore(t)
|
||||
s := NewStore()
|
||||
defer cleanup(DefaultDatabase, s)
|
||||
fileTest(s, t)
|
||||
}
|
||||
|
||||
func TestFileStoreTable(t *testing.T) {
|
||||
s := newTestFileStore(t, Table("testTable"))
|
||||
s := NewStore(Table("testTable"))
|
||||
defer cleanup(DefaultDatabase, s)
|
||||
fileTest(s, t)
|
||||
}
|
||||
|
||||
func TestFileStoreDatabase(t *testing.T) {
|
||||
s := newTestFileStore(t, Database("testdb"))
|
||||
s := NewStore(Database("testdb"))
|
||||
defer cleanup("testdb", s)
|
||||
fileTest(s, t)
|
||||
}
|
||||
|
||||
func TestFileStoreDatabaseTable(t *testing.T) {
|
||||
s := newTestFileStore(t, Table("testTable"), Database("testdb"))
|
||||
s := NewStore(Table("testTable"), Database("testdb"))
|
||||
defer cleanup("testdb", s)
|
||||
fileTest(s, t)
|
||||
}
|
||||
|
||||
@@ -94,7 +94,7 @@ func fileTest(s Store, t *testing.T) {
|
||||
{
|
||||
Key: "foobar",
|
||||
Value: []byte("foobarfoobar"),
|
||||
Expiry: time.Second, // wide window: CI I/O under -race can exceed a 100ms expiry before the read below
|
||||
Expiry: time.Millisecond * 100,
|
||||
},
|
||||
}
|
||||
|
||||
@@ -113,8 +113,8 @@ func fileTest(s Store, t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
// wait for the expiry (must exceed the 1s Expiry above, with margin for slow CI)
|
||||
time.Sleep(time.Second * 2)
|
||||
// wait for the expiry
|
||||
time.Sleep(time.Millisecond * 200)
|
||||
|
||||
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 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.
|
||||
// Write 3 records with various expiry and get with Suffix
|
||||
records = []*Record{
|
||||
{
|
||||
Key: "foo",
|
||||
Value: []byte("foofoo"),
|
||||
},
|
||||
{
|
||||
Key: "barfoo",
|
||||
Value: []byte("barfoobarfoo"),
|
||||
Expiry: -time.Second,
|
||||
Key: "barfoo",
|
||||
Value: []byte("barfoobarfoo"),
|
||||
|
||||
Expiry: time.Millisecond * 100,
|
||||
},
|
||||
{
|
||||
Key: "bazbarfoo",
|
||||
Value: []byte("bazbarfoobazbarfoo"),
|
||||
Key: "bazbarfoo",
|
||||
Value: []byte("bazbarfoobazbarfoo"),
|
||||
Expiry: 2 * time.Millisecond * 100,
|
||||
},
|
||||
}
|
||||
for _, r := range records {
|
||||
@@ -160,44 +160,59 @@ 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) != 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)
|
||||
} else {
|
||||
if len(results) != 3 {
|
||||
t.Errorf("Expected 3 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 unexpired suffix item, got %d (%# v)", len(results), spew.Sdump(results))
|
||||
} 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))
|
||||
}
|
||||
}
|
||||
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.Second)); err != nil {
|
||||
}, WriteTTL(time.Millisecond*100)); err != nil {
|
||||
t.Error(err)
|
||||
}
|
||||
if err := s.Write(&Record{
|
||||
Key: "foofoo",
|
||||
Value: []byte("something"),
|
||||
}, WriteExpiry(time.Now().Add(time.Second))); err != nil {
|
||||
}, WriteExpiry(time.Now().Add(time.Millisecond*100))); 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.Second)); err != nil {
|
||||
}, WriteExpiry(time.Now().Add(time.Hour)), WriteTTL(time.Millisecond*100)); err != nil {
|
||||
t.Error(err)
|
||||
}
|
||||
|
||||
@@ -209,7 +224,7 @@ func fileTest(s Store, t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
time.Sleep(time.Second * 2) // exceed the 1s TTL/expiry above so everything has expired
|
||||
time.Sleep(time.Millisecond * 100)
|
||||
|
||||
if results, err := s.List(); err != nil {
|
||||
t.Errorf("List failed: %s", err)
|
||||
|
||||
Reference in New Issue
Block a user