Compare commits

..

1 Commits

Author SHA1 Message Date
Codex 821e768769 docs(priorities): refresh planner queue
Harness (E2E) / Harnesses (mock LLM) (push) Waiting to run
Harness (E2E) / Provider harnesses (live LLM conformance) (push) Waiting to run
Lint / golangci-lint (push) Waiting to run
Run Tests / Unit Tests (push) Waiting to run
Run Tests / Etcd Integration Tests (push) Waiting to run
2026-07-02 12:01:49 +00:00
45 changed files with 308 additions and 2425 deletions
+7 -4
View File
@@ -21,10 +21,13 @@ changes, architectural rewrites. Those go to the human.
## Work queue (ranked)
1. **Propagate agent run cancellation and deadlines through model and tool calls** ([#3544](https://github.com/micro/go-micro/issues/3544)) — #3839 closed the remaining atlascloud A2A fallback marker issue, so the top Now-phase resilience gap is predictable failure semantics across agent runs, model calls, tool calls, plan/delegate, and flow handoffs. Tool retries, live-provider timeout classification, delegated-plan completion, side-effect enforcement, and file-store isolation are in place; the lifecycle still needs cancellation/deadline propagation so work fails safely instead of becoming opaque loops.
2. **Make first-agent preflight failures actionable** ([#3841](https://github.com/micro/go-micro/issues/3841)) — developer adoption is the current goal, and the README/docs now provide a no-secret first-agent route, `micro agent preflight`, debugging guide, and 0→hero harness. The next highest-value adoption increment is to make preflight output behave like a local doctor: when Go, the CLI, provider-key setup, or the gateway port is wrong, it should tell a new user exactly what to fix and where to continue, without needing secrets or provider calls.
3. **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, fixed provider-emitted text tool calls, and narrowed the latest provider failures to diagnosability/contract gaps. 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.
4. **Add an AP2 mandate layer over A2A and x402** ([#3552](https://github.com/micro/go-micro/issues/3552)) — this is a forward interop investment, not a Now-phase blocker: Go Micro already has A2A agents and x402 paid tools, so a small signed-mandate foundation can keep agent payments aligned with the open-protocol story without pulling the queue away from adoption, resilience, or observability. Keep it additive and opt-in while the AP2/FIDO work settles.
1. **Make universe checkout conformance send exactly one concierge notification** ([#3633](https://github.com/micro/go-micro/issues/3633)) — durable checkout/universe is the strongest 0→hero proof for services → agents → workflows, and resume idempotency must be boring. Keep this first because duplicate buyer notifications are user-visible side effects in the adoption story and the plan/delegate duplicate-side-effect issue is now closed by PR #3664.
2. **Make the universe A2A reachability check deterministic** ([#3653](https://github.com/micro/go-micro/issues/3653)) — the latest live-provider scan shows the final universe A2A smoke can trigger extra concierge notifications and time out even after checkout durability assertions pass. Rank it with the universe side-effect work because gateway reachability should prove interop without adding more side effects or making CI flaky.
3. **Expose `fallback_echo` during A2A streaming fallback conformance** ([#3560](https://github.com/micro/go-micro/issues/3560)) — this remains the next scoped 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, without letting protocol depth outrank the on-ramp or side-effect safety.
4. **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.
5. **Propagate agent run cancellation and deadlines through model and tool calls** ([#3544](https://github.com/micro/go-micro/issues/3544)) — the highest-value remaining Now-phase resilience gap after the live-provider side-effect fixes is predictable failure semantics across agent runs, model calls, tool calls, plan/delegate, and flow handoffs. Tool retries and live-provider deadline tuning are in place; the lifecycle still needs cancellation/deadline propagation so work fails safely instead of becoming opaque loops.
6. **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.
7. **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._
+6 -6
View File
@@ -1,14 +1,14 @@
<!--
The TRIAGE prompt — go-micro's CI-failure feedback path. Editable policy; the
workflow prepends the agent @mention and substitutes __ISSUE__ (this tracking
issue) and __RUNURL__ (the failed run) before posting. Keep both literal.
The TRIAGE prompt — go-micro's harness-failure feedback path. Editable policy;
the workflow prepends the agent @mention and substitutes __ISSUE__ (this
tracking issue) and __RUNURL__ (the failed run) before posting. Keep both literal.
-->
Triage the failed CI run at __RUNURL__. It may be the linter (Lint), the unit/integration tests (Run Tests), or the provider-conformance harness (Harness (E2E)).
Triage the failed provider-conformance harness run at __RUNURL__.
Read the logs and root-cause each distinct failure. DEDUPE against open issues — if a failure matches an existing issue, comment "recurred" there instead of filing a duplicate.
For each genuine, self-contained defect, file a scoped issue (`gh issue create --label codex --label enhancement --title "<scoped fix>" --body "<root cause, where, acceptance criteria>"`) so the increment loop builds it and the next CI/harness run verifies it. A lint or test failure on master is a real regression — file it so it is fixed promptly; do NOT ignore it.
For each genuine, self-contained defect, file a scoped issue (`gh issue create --label codex --label enhancement --title "<scoped fix>" --body "<root cause, where, acceptance criteria>"`) so the increment loop builds it and the next harness run verifies it.
IGNORE only genuine transient flakes — live-model latency, provider outages, rate limits, network timeouts with no code cause (mostly relevant to the harness). Anything needing a breaking or architectural change: file it as `needs-human` and describe it, rather than auto-queuing it as a routine fix.
IGNORE transient flakes — live-model latency, provider outages, rate limits, timeouts with no code cause. Anything needing a breaking or architectural change: file it as `needs-human` and describe it, rather than auto-queuing it as a routine fix.
Close this issue (`gh issue close __ISSUE__`) when triage is done. Open any PR yourself from the shell with `gh`; do not use the make_pr tool.
+2 -3
View File
@@ -18,7 +18,7 @@ on:
providers:
description: "Comma-separated providers for live conformance (default: all supported)"
required: false
default: "anthropic,openai,gemini,groq,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' }}"
+5 -6
View File
@@ -1,14 +1,14 @@
name: "Loop: Triage"
# Generated by `micro loop init`. The feedback path of the evaluator: when a CI
# workflow (Harness (E2E), Lint, Run Tests) fails on a non-PR run, dispatch the agent
# Generated by `micro loop init`. The feedback path of the evaluator: when the
# CI workflow ("Harness (E2E)") fails on a non-PR run, dispatch the agent
# (@codex) with the instruction in .github/loop/prompts/triage.md
# to root-cause the failure and file scoped fix issues back into the queue — so
# failures become fixes with no human in the middle. Gated on CODEX_TRIGGER_TOKEN.
on:
workflow_run:
workflows: ["Harness (E2E)", "Lint", "Run Tests"]
workflows: ["Harness (E2E)"]
types: [completed]
permissions:
@@ -33,7 +33,6 @@ jobs:
REPO: ${{ github.repository }}
RUN_ID: ${{ github.event.workflow_run.id }}
RUN_URL: ${{ github.event.workflow_run.html_url }}
WORKFLOW_NAME: ${{ github.event.workflow_run.name }}
run: |
if [ "$HAS_TOKEN" != "true" ]; then
echo "CODEX_TRIGGER_TOKEN is not set — skipping."
@@ -45,8 +44,8 @@ jobs:
exit 1
fi
ISSUE_URL=$(gh issue create --repo "$REPO" \
--title "Loop: triage failed run $RUN_ID ($WORKFLOW_NAME)" \
--body "The '$WORKFLOW_NAME' workflow failed on a non-PR run: $RUN_URL")
--title "Loop: triage failed run $RUN_ID" \
--body "The 'Harness (E2E)' workflow failed: $RUN_URL")
ISSUE_NUM="${ISSUE_URL##*/}"
echo "Opened issue #$ISSUE_NUM — dispatching triage."
{
-21
View File
@@ -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
+4 -7
View File
@@ -87,14 +87,12 @@ make harness
After install and the first `micro new`/`micro run` smoke check, take the
walkable agent path in this order:
1. [No-secret first-agent transcript](internal/website/docs/guides/no-secret-first-agent.md) — run the
maintained support agent with a mock model and see services → agents → workflows succeed without a key.
2. [Your First Agent](internal/website/docs/guides/your-first-agent.md) — build a
1. [Your First Agent](internal/website/docs/guides/your-first-agent.md) — build a
service-backed agent and talk to it with `micro chat`.
3. [Debugging your agent](internal/website/docs/guides/debugging-agents.md) — use
2. [Debugging your agent](internal/website/docs/guides/debugging-agents.md) — use
`micro agent inspect`, run history, memory, and provider checks when the first
conversation does something unexpected.
4. [0→hero Reference](internal/website/docs/guides/zero-to-hero.md) — complete the
3. [0→hero Reference](internal/website/docs/guides/zero-to-hero.md) — complete the
services → agents → workflows loop with scaffold, run, chat, inspect, flow
history, and deploy dry-run commands that match the maintained harness.
@@ -332,7 +330,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 +427,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
+46 -84
View File
@@ -293,77 +293,57 @@ func (a *agentImpl) askLocked(ctx context.Context, runID, message, parentRunID s
}
}
const maxPlanCompletionTurns = 3
var resp *ai.Response
for planCompletionTurn := 0; ; planCompletionTurn++ {
resp, err = ai.GenerateWithRetry(ctx, a.model, &ai.Request{
Prompt: message,
SystemPrompt: a.buildPrompt(),
Tools: toolList,
Messages: messages,
}, ai.GeneratePolicy{
Timeout: a.opts.ModelTimeout,
MaxAttempts: a.opts.ModelMaxAttempts,
Backoff: a.opts.ModelRetryBackoff,
})
if err != nil {
run.Status = agentRunFailureStatus(err)
if a.currentRun != nil {
run.Steps = a.currentRun.Steps
}
if len(run.Steps) == 0 {
run.Steps = []flow.StepRecord{{Name: agentAskStep}}
}
run.Steps[0].Status = run.Status
run.Steps[0].Error = err.Error()
_ = a.saveRun(ctx, run)
resp, err := ai.GenerateWithRetry(ctx, a.model, &ai.Request{
Prompt: message,
SystemPrompt: a.buildPrompt(),
Tools: toolList,
Messages: messages,
}, ai.GeneratePolicy{
Timeout: a.opts.ModelTimeout,
MaxAttempts: a.opts.ModelMaxAttempts,
Backoff: a.opts.ModelRetryBackoff,
})
if err != nil {
run.Status = agentRunFailureStatus(err)
if a.currentRun != nil {
run.Steps = a.currentRun.Steps
}
if len(run.Steps) == 0 {
run.Steps = []flow.StepRecord{{Name: agentAskStep}}
}
run.Steps[0].Status = run.Status
run.Steps[0].Error = err.Error()
_ = a.saveRun(ctx, run)
return nil, err
}
if a.pause != nil && a.opts.Checkpoint != nil {
run.Status = "paused"
run.State.Stage = agentApprovalStep
run.State.Data = []byte(message)
if a.pause.Tool == toolHumanInput {
run.State.Stage = agentInputStep
_ = run.State.Set(inputPause{OriginalMessage: message, Prompt: a.pause.Message})
}
run.Steps[0].Status = "paused"
run.Steps[0].Error = a.pause.Message
run.Steps[0].Result = a.pause.Tool
if err := a.saveRun(ctx, run); err != nil {
return nil, err
}
if a.pause != nil && a.opts.Checkpoint != nil {
run.Status = "paused"
run.State.Stage = agentApprovalStep
run.State.Data = []byte(message)
if a.pause.Tool == toolHumanInput {
run.State.Stage = agentInputStep
_ = run.State.Set(inputPause{OriginalMessage: message, Prompt: a.pause.Message})
}
run.Steps[0].Status = "paused"
run.Steps[0].Error = a.pause.Message
run.Steps[0].Result = a.pause.Tool
if err := a.saveRun(ctx, run); err != nil {
return nil, err
}
return nil, fmt.Errorf("agent run %s paused for approval: %s", run.ID, a.pause.Message)
}
return nil, fmt.Errorf("agent run %s paused for approval: %s", run.ID, a.pause.Message)
}
if len(resp.ToolCalls) == 0 {
if calls, answer, ok := a.executeTextToolCalls(ctx, resp.Reply, toolList); ok {
resp.ToolCalls = calls
if resp.Answer == "" {
resp.Answer = answer
}
trimmedReply := strings.TrimSpace(resp.Reply)
if strings.HasPrefix(trimmedReply, "{") || strings.HasPrefix(trimmedReply, "[") || strings.HasPrefix(trimmedReply, "```") {
resp.Reply = ""
}
if len(resp.ToolCalls) == 0 {
if calls, answer, ok := a.executeTextToolCalls(ctx, resp.Reply, toolList); ok {
resp.ToolCalls = calls
if resp.Answer == "" {
resp.Answer = answer
}
trimmedReply := strings.TrimSpace(resp.Reply)
if strings.HasPrefix(trimmedReply, "{") || strings.HasPrefix(trimmedReply, "[") || strings.HasPrefix(trimmedReply, "```") {
resp.Reply = ""
}
}
if a.opts.Checkpoint != nil {
if unfinished := a.unfinishedPlanSteps(); len(unfinished) > 0 && planCompletionTurn < maxPlanCompletionTurns {
if resp.Reply != "" {
a.mem.Add("assistant", resp.Reply)
}
if resp.Answer != "" {
a.mem.Add("assistant", resp.Answer)
}
message = "Continue the run. These plan steps are still unfinished and must be completed before a final answer: " + strings.Join(unfinished, ", ")
a.mem.Add("user", message)
messages = a.mem.Messages()
continue
}
}
break
}
if resp.Reply != "" {
@@ -388,24 +368,6 @@ func (a *agentImpl) askLocked(ctx context.Context, runID, message, parentRunID s
RunID: a.runID,
ParentID: parentRunID,
}
if a.opts.Checkpoint != nil {
if unfinished := a.unfinishedPlanSteps(); len(unfinished) > 0 {
err = fmt.Errorf("agent run %s has unfinished plan steps: %s", run.ID, strings.Join(unfinished, ", "))
run.Status = "failed"
run.State.Stage = agentAskStep
run.State.Data = []byte(message)
if a.currentRun != nil {
run.Steps = a.currentRun.Steps
}
if len(run.Steps) == 0 {
run.Steps = []flow.StepRecord{{Name: agentAskStep}}
}
run.Steps[0].Status = "failed"
run.Steps[0].Error = err.Error()
_ = a.saveRun(ctx, run)
return nil, err
}
}
run.Status = "done"
run.State.Stage = ""
if b, marshalErr := json.Marshal(res); marshalErr == nil {
-32
View File
@@ -402,38 +402,6 @@ func (a *agentImpl) completeNextPlanStep() {
}
}
func (a *agentImpl) unfinishedPlanSteps() []string {
plan := a.loadPlan()
if plan == "" {
return nil
}
var data map[string]any
if err := json.Unmarshal([]byte(plan), &data); err != nil {
return nil
}
steps, ok := data["steps"].([]any)
if !ok {
return nil
}
var unfinished []string
for _, raw := range steps {
step, ok := raw.(map[string]any)
if !ok {
continue
}
status, _ := step["status"].(string)
if status != "" && status != "pending" && status != "in_progress" {
continue
}
task, _ := step["task"].(string)
if task == "" {
task = "<unnamed>"
}
unfinished = append(unfinished, task)
}
return unfinished
}
// handleHumanInput records that the model needs operator input before it can continue.
func (a *agentImpl) handleHumanInput(call ai.ToolCall) ai.ToolResult {
prompt, _ := call.Input["prompt"].(string)
-78
View File
@@ -7,10 +7,7 @@ import (
"testing"
"go-micro.dev/v6/ai"
"go-micro.dev/v6/client"
codecBytes "go-micro.dev/v6/codec/bytes"
"go-micro.dev/v6/flow"
"go-micro.dev/v6/registry"
"go-micro.dev/v6/store"
)
@@ -144,81 +141,6 @@ func TestCheckpointSkipsDuplicateToolWithinAsk(t *testing.T) {
}
}
func TestCheckpointContinuesRunWithUnfinishedPlanStep(t *testing.T) {
ctx := context.Background()
cp := flow.StoreCheckpoint(store.NewMemoryStore(), "unfinished-plan-agent")
reg := registry.NewMemoryRegistry()
if err := reg.Register(&registry.Service{
Name: "comms",
Metadata: map[string]string{"type": "agent"},
Nodes: []*registry.Node{{Id: "comms-1", Address: "127.0.0.1:0"}},
}); err != nil {
t.Fatalf("register comms agent: %v", err)
}
delegateCalls := 0
fc := &fakeClient{Client: client.DefaultClient}
fc.callFn = func(ctx context.Context, req client.Request, rsp interface{}) error {
delegateCalls++
if req.Service() != "comms" || req.Endpoint() != "Agent.Chat" {
t.Fatalf("delegate RPC = %s %s, want comms Agent.Chat", req.Service(), req.Endpoint())
}
frame := rsp.(*codecBytes.Frame)
frame.Data = []byte(`{"reply":"owner notified","agent":"comms"}`)
return nil
}
modelCalls := 0
fakeGen = func(ctx context.Context, opts ai.Options, req *ai.Request) (*ai.Response, error) {
modelCalls++
if opts.ToolHandler == nil {
t.Fatal("missing tool handler")
}
switch modelCalls {
case 1:
opts.ToolHandler(ctx, ai.ToolCall{ID: "plan-1", Name: toolPlan, Input: map[string]any{
"steps": []any{
map[string]any{"task": "create launch tasks", "status": "done"},
map[string]any{"task": "delegate readiness notification to comms", "status": "in_progress"},
},
}})
return &ai.Response{Reply: "tasks are ready"}, nil
case 2:
if !strings.Contains(req.Prompt, "delegate readiness notification to comms") {
t.Fatalf("continuation prompt = %q, want unfinished step", req.Prompt)
}
res := opts.ToolHandler(ctx, ai.ToolCall{ID: "delegate-1", Name: toolDelegate, Input: map[string]any{"task": "Notify owner@acme.com that the launch plan is ready", "to": "comms"}})
if !strings.Contains(res.Content, "owner notified") {
t.Fatalf("delegate result = %q, want owner notified", res.Content)
}
return &ai.Response{Reply: "all done"}, nil
default:
t.Fatalf("unexpected model call %d", modelCalls)
return nil, nil
}
}
defer func() { fakeGen = nil }()
a := newTestAgent(Name("unfinished-plan-agent"), WithCheckpoint(cp), WithRegistry(reg), WithClient(fc))
resp, err := a.Ask(ctx, "create tasks and notify owner")
if err != nil {
t.Fatalf("Ask: %v", err)
}
if resp.Reply != "all done" {
t.Fatalf("reply = %q, want final continuation reply", resp.Reply)
}
if modelCalls != 2 {
t.Fatalf("model calls = %d, want initial plus continuation", modelCalls)
}
if delegateCalls != 1 {
t.Fatalf("delegate calls = %d, want exactly one", delegateCalls)
}
if unfinished := a.unfinishedPlanSteps(); len(unfinished) != 0 {
t.Fatalf("unfinished plan steps = %v, want none", unfinished)
}
}
func TestResumeFailedCheckpointAfterFreshAgentRestart(t *testing.T) {
ctx := context.Background()
cp := flow.StoreCheckpoint(store.NewMemoryStore(), "restart-resume-agent")
-14
View File
@@ -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
View File
@@ -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.
-182
View File
@@ -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 {
+4 -6
View File
@@ -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)
}
-196
View File
@@ -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
}
-96
View File
@@ -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())
}
}
-2
View File
@@ -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"):
+1 -3
View File
@@ -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)
}
-35
View File
@@ -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 -27
View File
@@ -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 {
+11 -38
View File
@@ -40,11 +40,10 @@ var templatesFS embed.FS
// given repo tunes.
type config struct {
// Shared.
DefaultBranch string // base branch for the loop's PRs (e.g. main)
AgentMention string // how the workflows summon the agent (e.g. @codex)
TokenSecret string // repo secret holding the user PAT that drives dispatch
CIWorkflow string // human-readable CI workflow name(s) triage watches
CIWorkflowsYAML string // the same as a YAML array literal, e.g. ["Lint", "Run Tests"]
DefaultBranch string // base branch for the loop's PRs (e.g. main)
AgentMention string // how the workflows summon the agent (e.g. @codex)
TokenSecret string // repo secret holding the user PAT that drives dispatch
CIWorkflow string // name: of the CI workflow triage watches for failures
// Per-dispatch-role (set while rendering each one).
Role string
@@ -123,7 +122,7 @@ Examples:
&cli.StringFlag{Name: "branch", Usage: "Base branch for the loop's PRs (auto-detected if empty)"},
&cli.StringFlag{Name: "agent", Usage: "How the workflows summon the agent (an @mention)", Value: "@codex"},
&cli.StringFlag{Name: "token-secret", Usage: "Repo secret holding the user PAT that drives dispatch", Value: "LOOP_TOKEN"},
&cli.StringFlag{Name: "ci-workflow", Usage: "CI workflow name(s) triage watches for failures (comma-separated)", Value: "CI"},
&cli.StringFlag{Name: "ci-workflow", Usage: "name: of the CI workflow triage watches for failures", Value: "CI"},
&cli.StringFlag{Name: "planner-cron", Usage: "Cron schedule for the planner", Value: "0 * * * *"},
&cli.StringFlag{Name: "builder-cron", Usage: "Cron schedule for the builder", Value: "30 * * * *"},
&cli.StringFlag{Name: "coherence-cron", Usage: "Cron schedule for the coherence role", Value: "0 7 * * *"},
@@ -150,15 +149,13 @@ func runInit(c *cli.Context) error {
return err
}
ciNames := splitCSV(c.String("ci-workflow"))
cfg := config{
DefaultBranch: c.String("branch"),
AgentMention: strings.TrimSpace(c.String("agent")),
TokenSecret: strings.TrimSpace(c.String("token-secret")),
CIWorkflow: strings.Join(ciNames, ", "),
CIWorkflowsYAML: yamlStringArray(ciNames),
TagPrefix: c.String("tag-prefix"),
ReleaseCron: c.String("release-cron"),
DefaultBranch: c.String("branch"),
AgentMention: strings.TrimSpace(c.String("agent")),
TokenSecret: strings.TrimSpace(c.String("token-secret")),
CIWorkflow: c.String("ci-workflow"),
TagPrefix: c.String("tag-prefix"),
ReleaseCron: c.String("release-cron"),
}
if cfg.DefaultBranch == "" {
cfg.DefaultBranch = detectDefaultBranch(dir)
@@ -208,30 +205,6 @@ func parseRoles(spec string) ([]string, error) {
return out, nil
}
// splitCSV splits a comma-separated flag into trimmed, non-empty values.
func splitCSV(s string) []string {
var out []string
for _, v := range strings.Split(s, ",") {
if v = strings.TrimSpace(v); v != "" {
out = append(out, v)
}
}
if len(out) == 0 {
out = []string{"CI"}
}
return out
}
// yamlStringArray renders names as a YAML/JSON flow array, e.g. ["Lint", "Run Tests"].
// Names are known workflow display names (no embedded quotes), so a simple quote is safe.
func yamlStringArray(names []string) string {
quoted := make([]string, len(names))
for i, n := range names {
quoted[i] = fmt.Sprintf("%q", n)
}
return "[" + strings.Join(quoted, ", ") + "]"
}
func isRole(r string) bool {
for _, x := range allRoles {
if x == r {
+6 -26
View File
@@ -8,13 +8,12 @@ import (
)
var testCfg = config{
DefaultBranch: "main",
AgentMention: "@codex",
TokenSecret: "LOOP_TOKEN",
CIWorkflow: "CI",
CIWorkflowsYAML: `["CI"]`,
TagPrefix: "v",
ReleaseCron: "0 23 * * *",
DefaultBranch: "main",
AgentMention: "@codex",
TokenSecret: "LOOP_TOKEN",
CIWorkflow: "CI",
TagPrefix: "v",
ReleaseCron: "0 23 * * *",
}
var testCrons = map[string]string{"planner": "0 * * * *", "builder": "30 * * * *", "coherence": "0 7 * * *"}
@@ -202,25 +201,6 @@ func TestReinitForceKeepsPromptsRefreshesWorkflows(t *testing.T) {
}
}
func TestCIWorkflowListRendersAsYAMLArray(t *testing.T) {
if got := yamlStringArray([]string{"Harness (E2E)", "Lint", "Run Tests"}); got != `["Harness (E2E)", "Lint", "Run Tests"]` {
t.Errorf("yamlStringArray = %q", got)
}
if got := splitCSV("Harness (E2E), Lint ,Run Tests"); strings.Join(got, "|") != "Harness (E2E)|Lint|Run Tests" {
t.Errorf("splitCSV = %v", got)
}
if got := splitCSV(" "); strings.Join(got, "|") != "CI" {
t.Errorf("splitCSV empty should default to CI, got %v", got)
}
// The triage workflow must embed the array so workflow_run watches all of them.
cfg := testCfg
cfg.CIWorkflowsYAML = `["Harness (E2E)", "Lint", "Run Tests"]`
s := mustRender(t, "templates/loop-triage.yml.tmpl", cfg)
if !strings.Contains(s, `workflows: ["Harness (E2E)", "Lint", "Run Tests"]`) {
t.Errorf("triage workflow does not watch the CI workflow list:\n%s", s)
}
}
func TestParseRoles(t *testing.T) {
if got, err := parseRoles("all"); err != nil || len(got) != len(allRoles) {
t.Errorf("all => %v, %v", got, err)
@@ -1,14 +1,14 @@
name: "Loop: Triage"
# Generated by `micro loop init`. The feedback path of the evaluator: when a CI
# workflow (<< .CIWorkflow >>) fails on a non-PR run, dispatch the agent
# Generated by `micro loop init`. The feedback path of the evaluator: when the
# CI workflow ("<< .CIWorkflow >>") fails on a non-PR run, dispatch the agent
# (<< .AgentMention >>) with the instruction in .github/loop/prompts/triage.md
# to root-cause the failure and file scoped fix issues back into the queue — so
# failures become fixes with no human in the middle. Gated on << .TokenSecret >>.
on:
workflow_run:
workflows: << .CIWorkflowsYAML >>
workflows: ["<< .CIWorkflow >>"]
types: [completed]
permissions:
@@ -33,7 +33,6 @@ jobs:
REPO: ${{ github.repository }}
RUN_ID: ${{ github.event.workflow_run.id }}
RUN_URL: ${{ github.event.workflow_run.html_url }}
WORKFLOW_NAME: ${{ github.event.workflow_run.name }}
run: |
if [ "$HAS_TOKEN" != "true" ]; then
echo "<< .TokenSecret >> is not set — skipping."
@@ -45,8 +44,8 @@ jobs:
exit 1
fi
ISSUE_URL=$(gh issue create --repo "$REPO" \
--title "Loop: triage failed run $RUN_ID ($WORKFLOW_NAME)" \
--body "The '$WORKFLOW_NAME' workflow failed on a non-PR run: $RUN_URL")
--title "Loop: triage failed run $RUN_ID" \
--body "The '<< .CIWorkflow >>' workflow failed: $RUN_URL")
ISSUE_NUM="${ISSUE_URL##*/}"
echo "Opened issue #$ISSUE_NUM — dispatching triage."
{
+31 -82
View File
@@ -1,34 +1,14 @@
# Go Micro Examples
This directory contains runnable examples that take you through the Go Micro
lifecycle: start with a service, expose it as agent-usable capability, then
coordinate work with workflows.
This directory contains runnable examples demonstrating various go-micro features and patterns.
## Quick Start
Each example can be run with `go run .` from its directory unless its README says
otherwise. If you are new to the repo, follow the first-agent path below instead
of reading the directories alphabetically.
Each example can be run with `go run .` from its directory.
## Recommended first-agent path
## Examples
| Step | Start here | What you learn | Next step |
|------|------------|----------------|-----------|
| 1. First service | [`hello-world`](./hello-world/) | Create and register a basic RPC service, add a handler, call it with a client, and expose health checks. | Move to [`agent-demo`](./agent-demo/) to see services used by an agent. |
| 2. First agent | [`agent-demo`](./agent-demo/) | Run a small project-management app with Projects, Tasks, and Team services plus an agent playground. | Compare with the maintained 0-to-hero path in [`support`](./support/). |
| 3. First workflow | [`support`](./support/) | Follow typed services into an agent chat loop, an event-driven `intake` flow, and an approval gate in one runnable reference. | Deepen the workflow model with [`flow-durable`](./flow-durable/). |
For the shortest AI-tooling bridge, the MCP path is
[`mcp/hello`](./mcp/hello/) → [`mcp/crud`](./mcp/crud/) →
[`mcp/workflow`](./mcp/workflow/). For debugging and production hardening, keep
[`agent-wrap-tool`](./agent-wrap-tool/), [`agent-durable`](./agent-durable/), and
[`deployment`](./deployment/) nearby.
## Lifecycle map
### 1. Services — learn the runtime foundation
#### [hello-world](./hello-world/)
### [hello-world](./hello-world/)
Basic RPC service demonstrating core concepts:
- Service creation and registration
- Handler implementation
@@ -41,7 +21,7 @@ cd hello-world
go run .
```
#### [web-service](./web-service/)
### [web-service](./web-service/)
HTTP web service with service discovery:
- HTTP handlers
- Service registration
@@ -54,7 +34,7 @@ cd web-service
go run .
```
#### [multi-service](./multi-service/)
### [multi-service](./multi-service/)
Multiple services in a single binary — the modular monolith pattern:
- Isolated server, client, store, and cache per service
- Shared registry and broker for inter-service communication
@@ -67,76 +47,44 @@ cd multi-service
go run .
```
#### [deployment](./deployment/)
### [deployment](./deployment/)
Docker Compose deployment with MCP gateway, Consul registry, and Jaeger tracing:
- Production-like architecture in one `docker-compose up`
- Standalone MCP gateway connected to service registry
- Distributed tracing with OpenTelemetry + Jaeger
### 2. Agents — turn services into tool-using teammates
#### [agent-demo](./agent-demo/)
Recommended first agent: a multi-service project management app with Projects,
Tasks, and Team services, seed data, and agent playground integration.
#### [agent-plan-delegate](./agent-plan-delegate/)
The two built-in agent capabilities in a small multi-agent system:
- **plan** — an agent records an ordered plan in its store-backed memory before doing multi-step work
- **delegate** — an agent hands a subtask to another agent (over RPC if it's registered, else to an ephemeral sub-agent)
#### [agent-wrap-tool](./agent-wrap-tool/)
Middleware around an agent's tool execution with `AgentWrapTool`, the tool-side analogue of client/server wrappers:
- **observe** — time every tool call and record per-tool metrics, correlated by call ID
- **retry** — re-run a call whose result is an error, recovering from a transient failure before the model sees it
#### [agent-durable](./agent-durable/)
Durable agent runs that can be checkpointed and resumed, useful once your first
agent needs predictable recovery behavior.
#### [agent-human-input](./agent-human-input/)
Human-in-the-loop agent interaction for decisions that need an explicit person
before the run can continue.
#### [agent-ollama](./agent-ollama/)
Local-model agent wiring for developers experimenting with Ollama-backed model
calls.
### 3. Workflows — coordinate longer-running work
#### [support](./support/)
A maintained 0-to-hero reference path in one runnable file:
- **scaffold** typed `customers`, `tickets`, and `notify` services
- **run/chat** with a support agent that uses those services as tools
- **inspect** the event-driven `intake` flow and approval gate
- **CI** keeps the deterministic mock-model journey runnable with `go test ./examples/support`
#### [flow-durable](./flow-durable/)
A workflow as ordered, checkpointed steps that survives a crash and resumes where it stopped:
- **steps** — a flow is a task with stages (`reserve → charge → confirm`), not just one LLM turn
- **Checkpoint** — each step is persisted; on `Resume`, completed steps are not re-run (no duplicate side effects)
#### [flow-loop](./flow-loop/)
A looping flow example for repeated workflow steps.
### 4. MCP and agent integration examples
### MCP Examples
See the [mcp/](./mcp/) directory for AI agent integration examples:
- **[hello](./mcp/hello/)** - Minimal MCP service (start here)
- **[crud](./mcp/crud/)** - CRUD contact book with full agent documentation
- **[workflow](./mcp/workflow/)** - Cross-service orchestration via AI agents
- **[documented](./mcp/documented/)** - All MCP features with auth scopes
- **[platform](./mcp/platform/)** - Platform-oriented MCP service example
## Other examples
### [agent-demo](./agent-demo/)
Multi-service project management app (Projects, Tasks, Team) with seed data and agent playground integration.
### [auth](./auth/)
Authentication and authorization example.
### [agent-plan-delegate](./agent-plan-delegate/)
The two built-in agent capabilities in a small multi-agent system:
- **plan** — an agent records an ordered plan in its store-backed memory before doing multi-step work
- **delegate** — an agent hands a subtask to another agent (over RPC if it's registered, else to an ephemeral sub-agent)
### [graceful-stop](./graceful-stop/)
Graceful shutdown behavior for long-running services.
### [agent-wrap-tool](./agent-wrap-tool/)
Middleware around an agent's tool execution with `AgentWrapTool`, the tool-side analogue of client/server wrappers:
- **observe** — time every tool call and record per-tool metrics, correlated by call ID
- **retry** — re-run a call whose result is an error, recovering from a transient failure before the model sees it
### [grpc-interop](./grpc-interop/)
gRPC interoperability example.
### [flow-durable](./flow-durable/)
A workflow as ordered, checkpointed steps that survives a crash and resumes where it stopped:
- **steps** — a flow is a task with stages (`reserve → charge → confirm`), not just one LLM turn
- **Checkpoint** — each step is persisted; on `Resume`, completed steps are not re-run (no duplicate side effects)
### [support](./support/)
A maintained 0-to-hero reference path in one runnable file:
- **scaffold** typed `customers`, `tickets`, and `notify` services
- **run/chat** with a support agent that uses those services as tools
- **inspect** the event-driven `intake` flow and approval gate
- **CI** keeps the deterministic mock-model journey runnable with `go test ./examples/support`
## Coming Soon
@@ -158,5 +106,6 @@ To add a new example:
1. Create a new directory
2. Add a descriptive README.md
3. Include working code with comments
4. Add to this index under the lifecycle stage it supports
4. Add to this index
5. Ensure it runs with `go run .`
+9 -11
View File
@@ -22,7 +22,7 @@ Actions instead of subagents. Each role is a workflow:
| **Planner** | `loop-planner.yml`*Loop: Planner* | Tracks live state, prioritizes the roadmap + an internal scan, and maintains the ranked queue in [`.github/loop/PRIORITIES.md`](../../.github/loop/PRIORITIES.md). Decides *what*. |
| **Generator** | `loop-builder.yml`*Loop: Builder (Generator)* | Builds the top open queue item as a single-concern PR (via Codex) and self-merges on green CI. Does the work. |
| **Evaluator** | `harness.yml`*Harness (E2E)*, plus the CI gate (`tests.yaml`, `lint.yaml`) | Grades every change: the mock harness + unit/lint on each push/PR, and real-model conformance hourly. A *separate* grader — never the generator judging itself. |
| **Evaluator → feedback** | `loop-triage.yml`*Loop: Triage (Evaluator feedback)* | When a gate workflow (Lint, Run Tests, or the harness) fails on a non-PR run, root-causes, dedupes, and files scoped fix issues back into the planner's queue. The hill-climbing feedback path. |
| **Evaluator → feedback** | `loop-triage.yml`*Loop: Triage (Evaluator feedback)* | On harness failure, root-causes, dedupes, and files scoped fix issues back into the planner's queue. The hill-climbing feedback path. |
| **Coherence** | `loop-coherence.yml`*Loop: Coherence* | Keeps README/website/docs/blog aligned with the North Star, keeps `CHANGELOG.md` living (reconciling `[Unreleased]` against merged PRs and rolling it into version headings as tags cut), and drafts the changelog blog post. |
| **Release** | `loop-release.yml`*Loop: Release (daily patch)* | Cuts a daily patch tag when master has new commits, so the *installable* framework tracks the loop's improvements (triggers `release.yml`/goreleaser). Minor/major bumps stay with the human. |
@@ -194,16 +194,14 @@ to redirect. Codex is serial, so these passes queue behind any in-flight increme
## Failure triage (the feedback loop)
The loop also closes on its own failures. `.github/workflows/loop-triage.yml`
fires when a gate workflow — **Lint**, **Run Tests**, or the provider-conformance
**Harness (E2E)** — finishes with `conclusion: failure` on a non-PR run (so a red
lint or test on `master`, not just a harness failure, becomes a fix issue). It
dispatches Codex to **triage** the failing run: read the logs, root-cause each
distinct failure, **dedupe** against open issues (comment "recurred" rather than
filing a duplicate), and file a scoped `codex`/`enhancement` issue for each genuine,
self-contained defect — which the increment loop then builds and the next run
verifies. Genuine transient flakes (live-model latency, provider outages) are
ignored; anything needing a breaking or architectural change is escalated as
`needs-human` instead of auto-built. This is
fires when the live provider-conformance harness finishes with `conclusion:
failure` (scheduled/manual runs only), and dispatches Codex to **triage** the
failing run: read the logs, root-cause each distinct failure, **dedupe** against
open issues (comment "recurred" rather than filing a duplicate), and file a scoped
`codex`/`enhancement` issue for each genuine, self-contained defect — which the
increment loop then builds and the next harness run verifies. Transient flakes
(live-model latency, provider outages) are ignored; anything needing a breaking or
architectural change is escalated as `needs-human` instead of auto-built. This is
the hill-climbing layer: CI/harness failures become fixes with no human in the
middle, short of a decision that's genuinely the human's.
+14 -71
View File
@@ -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")
}
}
+14 -115
View File
@@ -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 -144
View File
@@ -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) {
+24 -161
View File
@@ -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,44 +109,26 @@ 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.
// @example {"to": "buyer@acme.com", "message": "Your order is confirmed"}
func (s *Notify) Send(_ context.Context, req *SendRequest, rsp *SendResponse) error {
if !isBuyerNotification(req) {
to, message := "", ""
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)
key := req.To + "\x00" + req.Message
s.mu.Lock()
if s.seen == nil {
s.seen = make(map[string]struct{})
}
for _, key := range keys {
if _, ok := s.seen[key]; ok {
s.mu.Unlock()
fmt.Printf(" \033[35m[notify]\033[0m 📨 duplicate suppressed to=%s %q\n", req.To, req.Message)
rsp.Sent = true
return nil
}
}
for _, key := range keys {
s.seen[key] = struct{}{}
if _, ok := s.seen[key]; ok {
s.mu.Unlock()
fmt.Printf(" \033[35m[notify]\033[0m 📨 duplicate suppressed to=%s %q\n", req.To, req.Message)
rsp.Sent = true
return nil
}
s.seen[key] = struct{}{}
s.mu.Unlock()
atomic.AddInt64(&s.sent, 1)
@@ -158,110 +137,18 @@ func (s *Notify) Send(_ context.Context, req *SendRequest, rsp *SendResponse) er
return nil
}
func isBuyerNotification(req *SendRequest) bool {
if req == nil {
return false
}
return canonicalBuyerRecipient(req.To) != ""
}
func (s *Notify) recordRejected(to, message string) {
s.mu.Lock()
defer s.mu.Unlock()
if strings.TrimSpace(to) == "" && strings.TrimSpace(message) == "" {
s.lastRejected = nil
return
}
s.lastRejected = &SendRequest{To: to, Message: message}
}
func (s *Notify) rejectedSummary() string {
s.mu.Lock()
defer s.mu.Unlock()
if s.lastRejected == nil {
return "no rejected notify call observed"
}
return fmt.Sprintf("last notify args to=%q message=%q", s.lastRejected.To, s.lastRejected.Message)
}
func canonicalBuyerRecipient(to string) string {
recipient := strings.ToLower(strings.TrimSpace(to))
switch recipient {
case "buyer", "buyer@acme.com":
return "buyer@acme.com"
}
if strings.HasPrefix(recipient, "buyer-of-order-") && len(recipient) > len("buyer-of-order-") {
return "buyer@acme.com"
}
for _, field := range strings.FieldsFunc(recipient, func(r rune) bool {
switch r {
case ' ', '\t', '\n', '\r', ',', ';', ':', '/', '\\', '(', ')', '[', ']', '{', '}':
return true
default:
return false
}
}) {
switch field {
case "buyer", "buyer@acme.com":
return "buyer@acme.com"
}
}
return ""
}
func notificationDedupeKeys(req *SendRequest) []string {
recipient := canonicalBuyerRecipient(req.To)
if recipient == "" {
recipient = strings.TrimSpace(req.To)
}
keys := []string{recipient + "\x00" + req.Message}
message := strings.ToLower(req.Message)
if strings.Contains(message, "confirm") {
// Live models occasionally emit equivalent confirmation copy more than
// once while a resumed checkout is completing (for example, a concise
// "order-1 confirmed" followed by a fuller buyer-facing sentence). The
// harness has one checkout order, so treat confirmation messages to the
// same buyer as the same side effect while preserving exact-message
// idempotency for all other notifications.
keys = append(keys, recipient+"\x00confirmation")
}
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 +158,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
}
// ---------------------------------------------------------------------------
@@ -309,10 +190,6 @@ func (m *mockModel) Stream(context.Context, *ai.Request, ...ai.GenerateOption) (
}
func (m *mockModel) Generate(ctx context.Context, req *ai.Request, _ ...ai.GenerateOption) (*ai.Response, error) {
if strings.Contains(strings.ToLower(req.Prompt), "a2a reachability probe") {
return &ai.Response{Answer: "concierge reachable"}, nil
}
// The concierge is asked to notify the buyer. Find the notify tool and call it.
for _, t := range req.Tools {
if strings.Contains(t.Name, "Send") && m.opts.ToolHandler != nil {
@@ -344,19 +221,10 @@ func check(cond bool, format string, args ...any) {
// a2aReachable calls the named agent through the gateway using the A2A
// client — exercising both directions of the protocol — and reports
// whether the agent replied. The probe is intentionally side-effect-free:
// the checkout flow already proved notify tool execution, and reachability
// should not depend on a live model deciding to send another notification.
func a2aReachable(ctx context.Context, base, agent string) error {
probe := "A2A reachability probe only. Reply with the words concierge reachable. Do not call tools or send notifications."
reply, err := a2a.NewClient(base+"/agents/"+agent).Send(ctx, probe)
if err != nil {
return err
}
if strings.TrimSpace(reply) == "" {
return fmt.Errorf("empty A2A reply")
}
return nil
// whether the agent replied.
func a2aReachable(base, agent string) bool {
reply, err := a2a.NewClient(base+"/agents/"+agent).Send(context.Background(), "notify the buyer")
return err == nil && reply != ""
}
func providerKey(provider string) string {
@@ -454,7 +322,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 {
@@ -532,12 +400,7 @@ func runUniverse(provider string) int {
// translated to its Agent.Chat RPC.
gw := httptest.NewServer(a2a.New(a2a.Options{Registry: reg, Client: cl, BaseURL: "http://gw"}).Handler())
defer gw.Close()
beforeA2A := atomic.LoadInt64(&ntf.sent)
reachCtx, cancelReach := context.WithTimeout(ctx, 10*time.Second)
reachErr := a2aReachable(reachCtx, gw.URL, "concierge")
cancelReach()
check(reachErr == nil, "concierge agent reachable over the A2A gateway: %v", reachErr)
check(atomic.LoadInt64(&ntf.sent) == beforeA2A, "A2A reachability probe did not send extra buyer notifications")
check(a2aReachable(gw.URL, "concierge"), "concierge agent reachable over the A2A gateway")
fmt.Println("\n\033[1m> shutting down the universe\033[0m")
// defers stop the agent and flow (deregistering them).
-166
View File
@@ -68,169 +68,3 @@ func TestNotifyStepCompletesAfterObservedSideEffectTimeout(t *testing.T) {
t.Fatalf("notifications sent after duplicate = %d, want 1", got)
}
}
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()
for _, req := range []*SendRequest{
{To: "buyer@acme.com", Message: "Your order order-1 has been confirmed."},
{To: "buyer@acme.com", Message: "order-1 confirmed"},
} {
var rsp SendResponse
if err := ntf.Send(ctx, req, &rsp); err != nil {
t.Fatalf("send notification %q: %v", req.Message, err)
}
if !rsp.Sent {
t.Fatalf("send notification %q did not report sent", req.Message)
}
}
if got := atomic.LoadInt64(&ntf.sent); got != 1 {
t.Fatalf("equivalent confirmation notifications sent = %d, want 1", got)
}
}
func TestNotifyAcceptsBuyerAlias(t *testing.T) {
ntf := new(Notify)
ctx := context.Background()
var rsp SendResponse
if err := ntf.Send(ctx, &SendRequest{
To: "buyer",
Message: "Your order order-1 has been confirmed.",
}, &rsp); err != nil {
t.Fatalf("send buyer alias notification: %v", err)
}
if !rsp.Sent {
t.Fatal("buyer alias notification did not report sent")
}
if got := atomic.LoadInt64(&ntf.sent); got != 1 {
t.Fatalf("buyer alias notifications sent = %d, want 1", got)
}
if err := ntf.Send(ctx, &SendRequest{
To: "buyer@acme.com",
Message: "order-1 confirmed",
}, &rsp); err != nil {
t.Fatalf("send canonical buyer notification: %v", err)
}
if !rsp.Sent {
t.Fatal("canonical buyer notification did not report sent")
}
if got := atomic.LoadInt64(&ntf.sent); got != 1 {
t.Fatalf("alias/canonical confirmation notifications sent = %d, want 1", got)
}
}
func TestNotifyIgnoresNonBuyerRecipients(t *testing.T) {
ntf := new(Notify)
ctx := context.Background()
var rsp SendResponse
if err := ntf.Send(ctx, &SendRequest{
To: "order-1",
Message: "order-1 confirmed",
}, &rsp); err != nil {
t.Fatalf("send non-buyer notification: %v", err)
}
if rsp.Sent {
t.Fatal("non-buyer notification reported sent")
}
if got := atomic.LoadInt64(&ntf.sent); got != 0 {
t.Fatalf("non-buyer notifications sent = %d, want 0", got)
}
if err := ntf.Send(ctx, &SendRequest{
To: "buyer@acme.com",
Message: "Your order order-1 has been confirmed.",
}, &rsp); err != nil {
t.Fatalf("send buyer notification: %v", err)
}
if !rsp.Sent {
t.Fatal("buyer notification did not report sent")
}
if got := atomic.LoadInt64(&ntf.sent); got != 1 {
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)
}
}
@@ -45,7 +45,6 @@ func TestGuidesNavigationLeadsWithDoing(t *testing.T) {
nav := readFile(t, filepath.Join(root, "internal", "website", "_data", "navigation.yml"))
orderedGuides := []string{
"/docs/guides/no-secret-first-agent.html",
"/docs/guides/your-first-agent.html",
"/docs/guides/zero-to-hero.html",
"/docs/guides/plan-delegate.html",
@@ -74,37 +73,6 @@ func TestGuidesNavigationLeadsWithDoing(t *testing.T) {
}
}
func TestNoSecretFirstAgentTranscript(t *testing.T) {
root := filepath.Clean(filepath.Join("..", "..", ".."))
guide := readFile(t, filepath.Join(root, "internal", "website", "docs", "guides", "no-secret-first-agent.md"))
for _, want := range []string{
"go run ./examples/support",
"go test ./examples/support -run TestRunSupportMockSmoke -count=1",
"make harness",
"micro agent preflight",
"micro run",
"micro chat assistant",
"micro inspect agent assistant",
"go test ./cmd/micro -run TestFirstAgentWalkthroughCLIBoundaries -count=1",
"No-secret first-agent transcript",
} {
if !strings.Contains(guide, want) {
t.Fatalf("no-secret first-agent transcript missing %q", want)
}
}
readme := readFile(t, filepath.Join(root, "README.md"))
if !strings.Contains(readme, "internal/website/docs/guides/no-secret-first-agent.md") {
t.Fatal("README does not point to the no-secret first-agent transcript")
}
firstAgent := readFile(t, filepath.Join(root, "internal", "website", "docs", "guides", "your-first-agent.md"))
if !strings.Contains(firstAgent, "no-secret-first-agent.html") {
t.Fatal("Your First Agent guide does not point to the no-secret transcript")
}
}
func readFile(t *testing.T, name string) string {
t.Helper()
data, err := os.ReadFile(name)
-2
View File
@@ -5,8 +5,6 @@ core:
url: /docs/getting-started.html
- title: AI Integration
url: /docs/ai-integration.html
- title: No-secret First Agent
url: /docs/guides/no-secret-first-agent.html
- title: Your First Agent
url: /docs/guides/your-first-agent.html
- title: 0→hero Reference
@@ -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 |
@@ -1,94 +0,0 @@
---
layout: default
---
# No-secret first-agent transcript
This is the fastest first-agent success path when you do not have a provider key
handy. It starts from the maintained `examples/support` app and uses the
repository harness that CI already runs: real Go Micro services, registry,
broker, client, store, agent loop, flow handoff, and guardrail code with only the
LLM provider mocked.
Use it before the live-provider [Your First Agent](your-first-agent.html)
walkthrough when you want to see the services → agents → workflows lifecycle run
end to end with no secrets.
## What this proves
- **Services** expose typed `customers`, `tickets`, and `notify` endpoints.
- **The `support` agent** discovers those endpoints as tools and uses them to
triage a ticket.
- **The `intake` flow** turns a `ticket.created` event into an agent run.
- **The approval gate** intercepts the customer email action before the tool
executes.
## Transcript
From a fresh clone of the repository:
```sh
git clone https://github.com/micro/go-micro.git
cd go-micro
go run ./examples/support
```
The default provider is `mock`, so the command does not need `ANTHROPIC_API_KEY`,
`OPENAI_API_KEY`, or any other secret. A healthy run prints the event, service
calls, guardrail decision, and final support-agent reply in one terminal:
```text
> event: events.ticket.created {"id":"ticket-1","customer":"alice@acme.com",...}
[customers] looked up Alice (pro plan)
[tickets] ticket-1 → priority=high status=in_progress
▣ approval gate notify_NotifyService_Send(alice@acme.com) — approved
[notify] 📨 to=alice@acme.com: "Hi Alice — thanks for reaching out..."
support agent: Hi Alice — thanks for reaching out...
✓ ticket triaged and the customer was replied to — triggered by an event
```
That single run is the no-secret version of the first-agent loop: a service
capability exists, an agent calls it as a tool, and workflow infrastructure can
trigger and inspect the work.
## CI-backed check
Run the same deterministic path as a focused test:
```sh
go test ./examples/support -run TestRunSupportMockSmoke -count=1
```
For the broader no-secret contract that also checks scaffold, chat/inspect CLI
boundaries, flow history, deploy dry-run, and mock provider conformance, run:
```sh
make harness
```
## Equivalent scaffold → run → chat → inspect path
When you are ready to build the smaller live-agent version yourself, follow
[Your First Agent](your-first-agent.html). The command shape is the same, but a
live `micro chat` turn needs a provider key because the model is no longer
mocked:
```sh
micro agent preflight
micro run
micro chat assistant
micro inspect agent assistant
```
CI keeps those CLI boundaries present with:
```sh
go test ./cmd/micro -run TestFirstAgentWalkthroughCLIBoundaries -count=1
```
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"
@@ -13,13 +13,13 @@ run on events or schedules.
## Runnable reference first
If you want to run the lifecycle before copying code, start with the [no-secret first-agent transcript](no-secret-first-agent.html) or run the maintained support-desk example from the repository root:
If you want to run the lifecycle before copying code, start with the maintained support-desk example from the repository root:
```sh
go run ./examples/support
```
It uses a deterministic mock model by default, so it needs no provider key, and it exercises the same shape this guide teaches: services become tools, an agent uses them, and a flow can trigger the work. Use the transcript for expected output, then use this guide when you are ready to build the smaller 0→1 version yourself.
It uses a deterministic mock model by default, so it needs no provider key, and it exercises the same shape this guide teaches: services become tools, an agent uses them, and a flow can trigger the work. Use this guide when you are ready to build the smaller 0→1 version yourself.
## What you'll build
@@ -219,7 +219,7 @@ agent for judgment, tool use, and handoffs when the path is not known up front.
- Read the [0→hero reference path](zero-to-hero.html) for the CI-verified
lifecycle contract.
- Run the [no-secret first-agent transcript](no-secret-first-agent.html) or [`examples/support`](https://github.com/micro/go-micro/tree/master/examples/support) for the no-secret support-desk lifecycle.
- Run [`examples/support`](https://github.com/micro/go-micro/tree/master/examples/support) for the no-secret 0→hero support-desk lifecycle.
- Run [`examples/agent-plan-delegate`](https://github.com/micro/go-micro/tree/master/examples/agent-plan-delegate)
to see planning and delegation across agents.
- Read [Debugging your agent](debugging-agents.html) when a chat turn does not call the tool you expected, loops, refuses a call, loses memory, or fails after a flow handoff.
+1 -3
View File
@@ -16,7 +16,7 @@ It's built on a pluggable architecture of Go interfaces: service discovery, clie
## Learn More
Start with [Getting Started](getting-started.html) for install and the first local service. Then follow the first-agent on-ramp: [No-secret first-agent transcript](guides/no-secret-first-agent.html) to run a mock-model support agent, [Your First Agent](guides/your-first-agent.html) to build and chat with a service-backed agent, [Debugging your agent](guides/debugging-agents.html) to inspect runs and memory, and the [0→hero reference path](guides/zero-to-hero.html) to walk the full scaffold → run → chat → inspect → deploy dry-run lifecycle covered by CI.
Start with [Getting Started](getting-started.html) for install and the first local service. Then follow the first-agent on-ramp: [Your First Agent](guides/your-first-agent.html) to build and chat with a service-backed agent, [Debugging your agent](guides/debugging-agents.html) to inspect runs and memory, and the [0→hero reference path](guides/zero-to-hero.html) to walk the full scaffold → run → chat → inspect → deploy dry-run lifecycle covered by CI.
Otherwise continue to read the docs for more information about the framework.
@@ -24,7 +24,6 @@ Otherwise continue to read the docs for more information about the framework.
- [Getting Started](getting-started.html)
- [0→hero Reference](guides/zero-to-hero.html) - Walk scaffold → run → chat → inspect → deploy dry-run with CI-backed commands
- [No-secret first-agent transcript](guides/no-secret-first-agent.html) - Run the first useful agent path without a provider key
- [Your First Agent](guides/your-first-agent.html) - Build a service-backed agent end to end
- [MCP & AI Agents](mcp.html) - Turn services into AI-callable tools with the Model Context Protocol
- [CLI & Gateway Guide](guides/cli-gateway.html) - Development vs Production modes
@@ -48,7 +47,6 @@ Otherwise continue to read the docs for more information about the framework.
## AI & Agents
- [0→hero Reference](guides/zero-to-hero.html) - Walk scaffold → run → chat → inspect → deploy dry-run with CI-backed commands
- [No-secret first-agent transcript](guides/no-secret-first-agent.html) - Run the first useful agent path without a provider key
- [Your First Agent](guides/your-first-agent.html) - Build a service-backed agent end to end
- [Building AI-Native Services](guides/ai-native-services.html) - End-to-end tutorial for MCP-enabled services
- [MCP Security Guide](guides/mcp-security.html) - Auth, scopes, rate limiting, and audit logging
+54 -39
View File
@@ -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)
-109
View File
@@ -1,109 +0,0 @@
package x402
import (
"crypto/ed25519"
"crypto/rand"
"encoding/base64"
"encoding/hex"
"encoding/json"
"fmt"
"net/http"
"strings"
"time"
)
// CDPFacilitatorURL is Coinbase Developer Platform's hosted x402 facilitator,
// which can settle real payments on Base mainnet (the open x402.org facilitator
// is testnet-only).
const CDPFacilitatorURL = "https://api.cdp.coinbase.com/platform/v2/x402"
// CDP returns a Facilitator (and Settler) backed by Coinbase's hosted
// facilitator, authenticating each verify/settle call with a short-lived
// Ed25519 Bearer JWT minted from a CDP Secret API Key. keyID and keySecret are
// the CDP API Key ID and base64 Ed25519 secret; the secret is used only to sign
// the JWT (stdlib crypto — no chain code, no external dependency).
//
// cfg.Facilitator = x402.CDP(os.Getenv("CDP_API_KEY_ID"), os.Getenv("CDP_API_KEY_SECRET"))
func CDP(keyID, keySecret string) *HTTPFacilitator {
return &HTTPFacilitator{
URL: CDPFacilitatorURL,
Authorize: cdpAuthorizer(keyID, keySecret),
}
}
// cdpAuthorizer returns an Authorize hook that attaches a CDP Bearer JWT bound
// to the request's method and URL.
func cdpAuthorizer(keyID, keySecret string) func(*http.Request) error {
return func(r *http.Request) error {
tok, err := cdpBearer(keyID, keySecret, r.Method, r.URL.Host, r.URL.Path)
if err != nil {
return err
}
r.Header.Set("Authorization", "Bearer "+tok)
return nil
}
}
// cdpBearer builds a CDP Bearer JWT (EdDSA / Ed25519) authorizing a single REST
// call, per CDP's authentication spec: the token binds to "METHOD host/path"
// and is valid for two minutes.
func cdpBearer(keyID, keySecret, method, host, path string) (string, error) {
if keyID == "" || keySecret == "" {
return "", fmt.Errorf("CDP API key id and secret are required")
}
key, err := ed25519KeyFromSecret(keySecret)
if err != nil {
return "", err
}
nonce := make([]byte, 16)
if _, err := rand.Read(nonce); err != nil {
return "", err
}
header := map[string]any{
"alg": "EdDSA",
"typ": "JWT",
"kid": keyID,
"nonce": hex.EncodeToString(nonce),
}
now := time.Now().Unix()
claims := map[string]any{
"sub": keyID,
"iss": "cdp",
"aud": []string{"cdp_service"},
"nbf": now,
"exp": now + 120,
"uri": method + " " + host + path,
}
hb, _ := json.Marshal(header)
cb, _ := json.Marshal(claims)
signing := b64url(hb) + "." + b64url(cb)
sig := ed25519.Sign(key, []byte(signing))
return signing + "." + b64url(sig), nil
}
// ed25519KeyFromSecret decodes a CDP Ed25519 secret. CDP secrets are base64 and
// decode to 64 bytes (32-byte seed + 32-byte public key) — Go's PrivateKey
// layout; a bare 32-byte seed is also accepted.
func ed25519KeyFromSecret(secret string) (ed25519.PrivateKey, error) {
secret = strings.TrimSpace(secret)
if strings.Contains(secret, "BEGIN") {
return nil, fmt.Errorf("CDP secret looks like a PEM/EC key; x402 bearer auth needs an Ed25519 Secret API Key")
}
raw, err := base64.StdEncoding.DecodeString(secret)
if err != nil {
if raw, err = base64.RawURLEncoding.DecodeString(secret); err != nil {
return nil, fmt.Errorf("CDP secret is not valid base64: %w", err)
}
}
switch len(raw) {
case ed25519.PrivateKeySize: // 64: seed + public key
return ed25519.PrivateKey(raw), nil
case ed25519.SeedSize: // 32: seed only
return ed25519.NewKeyFromSeed(raw), nil
default:
return nil, fmt.Errorf("CDP secret decoded to %d bytes; expected 32 or 64 (Ed25519)", len(raw))
}
}
func b64url(b []byte) string { return base64.RawURLEncoding.EncodeToString(b) }
-38
View File
@@ -1,38 +0,0 @@
package x402
import "strings"
// assetInfo describes a known stablecoin: its contract address and EIP-712
// domain (name, version) used to build a transfer-authorization signature.
type assetInfo struct {
Address string
Name string
Version string
}
// knownAssets maps a CAIP-2 network to its default stablecoin (USDC). Used to
// fill in the asset and its EIP-712 domain when the operator does not specify
// them, so a client can sign without hand-configured token metadata.
var knownAssets = map[string]assetInfo{
"eip155:8453": {"0x833589fCD6eDb6E08f4c7C32D4f71b54bdA02913", "USD Coin", "2"}, // Base mainnet
"eip155:84532": {"0x036CbD53842c5426634e7929541eC2318f3dCF7e", "USDC", "2"}, // Base Sepolia
}
// NormalizeNetwork maps common short chain names to their CAIP-2 identifiers,
// which hosted facilitators (Coinbase CDP) use, and passes anything else
// through unchanged. Empty defaults to Base mainnet.
func NormalizeNetwork(n string) string {
switch strings.ToLower(strings.TrimSpace(n)) {
case "", "base", "eip155:8453":
return "eip155:8453"
case "base-sepolia", "eip155:84532":
return "eip155:84532"
default:
return n
}
}
func defaultAsset(network string) (assetInfo, bool) {
a, ok := knownAssets[network]
return a, ok
}
+42 -189
View File
@@ -15,11 +15,6 @@
// })
// mux.Handle("/paid", pay(handler))
//
// For real settlement on Base mainnet, point it at the Coinbase CDP
// facilitator, which requires an authenticated request:
//
// cfg.Facilitator = x402.CDP(os.Getenv("CDP_API_KEY_ID"), os.Getenv("CDP_API_KEY_SECRET"))
//
// x402 is governed by the x402 Foundation (Linux Foundation). See
// https://x402.org and https://docs.cdp.coinbase.com/x402.
package x402
@@ -27,41 +22,32 @@ package x402
import (
"bytes"
"context"
"encoding/base64"
"encoding/json"
"fmt"
"net/http"
"os"
"strings"
)
// Version is the x402 protocol version this package speaks.
const Version = 1
// Header names defined by the protocol. Version 1 uses X-PAYMENT /
// X-PAYMENT-RESPONSE; version 2 renamed them to PAYMENT-SIGNATURE /
// PAYMENT-RESPONSE. We accept either request header and emit both response
// headers so any conformant client interoperates.
// Header names defined by the protocol.
const (
PaymentHeader = "X-PAYMENT" // request: the client's payment payload
PaymentHeaderV2 = "PAYMENT-SIGNATURE" // request: v2 alias
PaymentResponseHeader = "X-PAYMENT-RESPONSE" // response: settlement details
PaymentResponseHeaderV2 = "PAYMENT-RESPONSE" // response: v2 alias
PaymentHeader = "X-PAYMENT" // request: the client's payment payload
PaymentResponseHeader = "X-PAYMENT-RESPONSE" // response: settlement details
)
// Requirements describes what a client must pay to access a resource —
// the body of a 402 response (one entry of "accepts").
type Requirements struct {
Scheme string `json:"scheme"` // payment scheme, e.g. "exact"
Network string `json:"network"` // chain, e.g. "base", "eip155:8453"
MaxAmountRequired string `json:"maxAmountRequired"` // amount in the asset's smallest unit
Resource string `json:"resource"` // the resource being paid for
Description string `json:"description,omitempty"` // human/agent-readable description
MimeType string `json:"mimeType,omitempty"` // response mime type
PayTo string `json:"payTo"` // receiving address
Asset string `json:"asset,omitempty"` // token contract/mint (default: network USDC)
MaxTimeoutSeconds int `json:"maxTimeoutSeconds,omitempty"` // how long the client has to pay
Extra map[string]string `json:"extra,omitempty"` // scheme extras, e.g. EIP-712 domain {name, version}
Scheme string `json:"scheme"` // payment scheme, e.g. "exact"
Network string `json:"network"` // chain, e.g. "base", "solana"
MaxAmountRequired string `json:"maxAmountRequired"` // amount in the asset's smallest unit
Resource string `json:"resource"` // the resource being paid for
Description string `json:"description,omitempty"` // human/agent-readable description
PayTo string `json:"payTo"` // receiving address
Asset string `json:"asset,omitempty"` // token contract/mint (default: network USDC)
MaxTimeoutSeconds int `json:"maxTimeoutSeconds,omitempty"` // how long the client has to pay
}
// challenge is the JSON body returned with a 402 response.
@@ -71,7 +57,7 @@ type challenge struct {
Error string `json:"error,omitempty"`
}
// Result is the outcome of verifying (or settling) a payment.
// Result is the outcome of verifying a payment.
type Result struct {
Valid bool // whether the payment satisfies the requirements
Payer string // the paying address, if known
@@ -79,29 +65,20 @@ type Result struct {
Settlement string // settlement reference (e.g. tx hash), set into X-PAYMENT-RESPONSE
}
// Facilitator verifies a payment a client presented against the stated
// requirements. Implementations talk to a chain or a hosted facilitator; the
// gateway stays chain-agnostic, so a Base facilitator and a Solana facilitator
// are just different implementations behind this interface.
// Facilitator verifies (and optionally settles) a payment a client
// presented against the stated requirements. Implementations talk to a
// chain or a hosted facilitator; the gateway stays chain-agnostic, so a
// Base facilitator and a Solana facilitator are just different
// implementations behind this interface.
type Facilitator interface {
Verify(ctx context.Context, payment string, req Requirements) (Result, error)
}
// Settler is an optional capability: a Facilitator that also settles a verified
// payment on-chain (captures the funds) and returns a settlement reference.
// Require calls Settle after a successful Verify when the facilitator
// implements it — for the "exact" scheme, verify authorizes and settle
// captures, so without settlement no funds actually move.
type Settler interface {
Settle(ctx context.Context, payment string, req Requirements) (Result, error)
}
// Config configures payment enforcement for a set of routes or tools.
type Config struct {
// PayTo is the address payments are sent to. Required.
PayTo string `json:"payTo"`
// Network is the chain to settle on (default "base"). Accepts short
// names ("base", "base-sepolia") or CAIP-2 ids ("eip155:8453").
// Network is the chain to settle on (default "base").
Network string `json:"network,omitempty"`
// Asset is the token contract/mint (default: the network's USDC).
Asset string `json:"asset,omitempty"`
@@ -115,13 +92,8 @@ type Config struct {
Amounts map[string]string `json:"amounts,omitempty"`
// Description is shown to the paying client/agent.
Description string `json:"description,omitempty"`
// Extra carries scheme-specific data echoed in the requirement's "extra".
// For the EVM "exact" scheme this is the asset's EIP-712 domain, e.g.
// {"name":"USD Coin","version":"2"}; when empty it is filled in for known
// assets so clients can build a valid transfer signature.
Extra map[string]string `json:"extra,omitempty"`
// Facilitator verifies (and, if it implements Settler, settles) payments.
// Defaults to an HTTPFacilitator pointed at FacilitatorURL.
// Facilitator verifies payments. Defaults to an HTTPFacilitator
// pointed at FacilitatorURL.
Facilitator Facilitator `json:"-"`
// FacilitatorURL is the verify/settle endpoint used when Facilitator
// is nil (e.g. Coinbase CDP or Alchemy).
@@ -152,62 +124,35 @@ func (c Config) facilitator() Facilitator {
}
func (c Config) requirements(amount, resource string) Requirements {
net := c.network()
asset := c.Asset
extra := c.Extra
// Fill the asset and its EIP-712 domain for known networks so a client
// can sign without the operator hand-configuring token metadata. Keyed on
// the CAIP-2 form so both "base" and "eip155:8453" resolve.
if def, ok := defaultAsset(NormalizeNetwork(net)); ok {
if asset == "" {
asset = def.Address
}
if extra == nil && strings.EqualFold(asset, def.Address) {
extra = map[string]string{"name": def.Name, "version": def.Version}
}
}
return Requirements{
Scheme: "exact",
Network: net,
Network: c.network(),
MaxAmountRequired: amount,
Resource: resource,
Description: c.Description,
MimeType: "application/json",
PayTo: c.PayTo,
Asset: asset,
Asset: c.Asset,
MaxTimeoutSeconds: 60,
Extra: extra,
}
}
// Payment returns the client's payment payload from either the v1 or v2
// request header, or "" if none is present.
func Payment(r *http.Request) string {
if p := r.Header.Get(PaymentHeader); p != "" {
return p
}
return r.Header.Get(PaymentHeaderV2)
}
// Require enforces payment of amount for a single request. It returns
// true if the request may proceed — the amount is free ("" or "0"), or a
// valid payment was presented (and settled, when the facilitator supports
// it) — and false once it has written a 402 challenge, in which case the
// caller must stop. resource names what is being paid for (a tool name or
// URL path).
// valid payment was presented and false once it has written a 402
// challenge, in which case the caller must stop. resource names what is
// being paid for (a tool name or URL path).
func (c Config) Require(w http.ResponseWriter, r *http.Request, amount, resource string) bool {
if amount == "" || amount == "0" {
return true // free
}
req := c.requirements(amount, resource)
payment := Payment(r)
payment := r.Header.Get(PaymentHeader)
if payment == "" {
writeChallenge(w, req, "payment required")
return false
}
fac := c.facilitator()
res, err := fac.Verify(r.Context(), payment, req)
res, err := c.facilitator().Verify(r.Context(), payment, req)
if err != nil {
writeChallenge(w, req, "payment verification failed: "+err.Error())
return false
@@ -220,28 +165,8 @@ func (c Config) Require(w http.ResponseWriter, r *http.Request, amount, resource
writeChallenge(w, req, reason)
return false
}
// Capture the funds when the facilitator can settle. Verify alone only
// authorizes the "exact" transfer; settlement broadcasts it.
if s, ok := fac.(Settler); ok {
sres, err := s.Settle(r.Context(), payment, req)
if err != nil {
writeChallenge(w, req, "payment settlement failed: "+err.Error())
return false
}
if !sres.Valid {
reason := sres.Reason
if reason == "" {
reason = "settlement rejected"
}
writeChallenge(w, req, reason)
return false
}
if sres.Settlement != "" {
res.Settlement = sres.Settlement
}
}
if res.Settlement != "" {
setSettlementHeaders(w, res.Settlement)
w.Header().Set(PaymentResponseHeader, res.Settlement)
}
return true
}
@@ -259,11 +184,6 @@ func Middleware(cfg Config) func(http.Handler) http.Handler {
}
}
func setSettlementHeaders(w http.ResponseWriter, settlement string) {
w.Header().Set(PaymentResponseHeader, settlement)
w.Header().Set(PaymentResponseHeaderV2, settlement)
}
func writeChallenge(w http.ResponseWriter, req Requirements, reason string) {
w.Header().Set("Content-Type", "application/json")
w.WriteHeader(http.StatusPaymentRequired) // 402
@@ -291,111 +211,44 @@ func LoadConfig(path string) (*Config, error) {
return &c, nil
}
// HTTPFacilitator verifies and settles payments by POSTing to an x402
// facilitator's /verify and /settle endpoints (Coinbase CDP, Alchemy, or
// self-hosted). It carries no chain or crypto code itself; when the endpoint
// requires authentication (e.g. CDP), set Authorize to attach credentials —
// see CDP.
// HTTPFacilitator verifies payments by POSTing to an x402 facilitator's
// verify endpoint (Coinbase CDP, Alchemy, or self-hosted). It carries no
// chain or crypto code itself.
type HTTPFacilitator struct {
URL string
Client *http.Client
// Authorize, when set, is called on each facilitator request to attach
// authentication (e.g. a Bearer token). Nil for open facilitators.
Authorize func(*http.Request) error
}
// Verify checks the payment is valid against the requirements.
func (f *HTTPFacilitator) Verify(ctx context.Context, payment string, req Requirements) (Result, error) {
out, err := f.post(ctx, "/verify", payment, req)
if err != nil {
return Result{}, err
}
return Result{Valid: out.IsValid, Reason: firstNonEmpty(out.InvalidReason, out.Error), Payer: out.Payer}, nil
}
// Settle captures a verified payment on-chain and returns the transaction
// reference, satisfying Settler.
func (f *HTTPFacilitator) Settle(ctx context.Context, payment string, req Requirements) (Result, error) {
out, err := f.post(ctx, "/settle", payment, req)
if err != nil {
return Result{}, err
}
// Facilitators report settlement as "success" with a "transaction" ref.
ok := out.Success || out.IsValid
return Result{Valid: ok, Reason: firstNonEmpty(out.ErrorReason, out.InvalidReason, out.Error), Payer: out.Payer, Settlement: out.Transaction}, nil
}
type facilitatorResponse struct {
IsValid bool `json:"isValid"`
Success bool `json:"success"`
InvalidReason string `json:"invalidReason"`
ErrorReason string `json:"errorReason"`
Error string `json:"error"`
Payer string `json:"payer"`
Transaction string `json:"transaction"`
}
func (f *HTTPFacilitator) post(ctx context.Context, path, payment string, req Requirements) (facilitatorResponse, error) {
var out facilitatorResponse
if f.URL == "" {
return out, fmt.Errorf("no facilitator configured")
return Result{}, fmt.Errorf("no facilitator configured")
}
body, _ := json.Marshal(map[string]any{
"x402Version": Version,
"paymentPayload": decodePayment(payment),
"paymentPayload": payment,
"paymentRequirements": req,
})
hreq, err := http.NewRequestWithContext(ctx, http.MethodPost, strings.TrimRight(f.URL, "/")+path, bytes.NewReader(body))
hreq, err := http.NewRequestWithContext(ctx, http.MethodPost, f.URL+"/verify", bytes.NewReader(body))
if err != nil {
return out, err
return Result{}, err
}
hreq.Header.Set("Content-Type", "application/json")
if f.Authorize != nil {
if err := f.Authorize(hreq); err != nil {
return out, fmt.Errorf("authorize: %w", err)
}
}
cl := f.Client
if cl == nil {
cl = http.DefaultClient
}
resp, err := cl.Do(hreq)
if err != nil {
return out, err
return Result{}, err
}
defer resp.Body.Close()
if resp.StatusCode != http.StatusOK {
var buf bytes.Buffer
_, _ = buf.ReadFrom(resp.Body)
return out, fmt.Errorf("facilitator %s returned %d: %s", path, resp.StatusCode, strings.TrimSpace(buf.String()))
var out struct {
IsValid bool `json:"isValid"`
InvalidReason string `json:"invalidReason"`
Payer string `json:"payer"`
}
if err := json.NewDecoder(resp.Body).Decode(&out); err != nil {
return out, err
return Result{}, err
}
return out, nil
}
// decodePayment turns the base64 X-PAYMENT header into the PaymentPayload
// object facilitators expect. If it is not base64 JSON, the raw value is passed
// through unchanged (some facilitators accept the encoded string).
func decodePayment(payment string) any {
payment = strings.TrimSpace(payment)
for _, dec := range []*base64.Encoding{base64.StdEncoding, base64.RawURLEncoding} {
if raw, err := dec.DecodeString(payment); err == nil {
var obj any
if json.Unmarshal(raw, &obj) == nil {
return obj
}
}
}
return payment
}
func firstNonEmpty(vals ...string) string {
for _, v := range vals {
if strings.TrimSpace(v) != "" {
return v
}
}
return ""
return Result{Valid: out.IsValid, Reason: out.InvalidReason, Payer: out.Payer}, nil
}
-162
View File
@@ -1,162 +0,0 @@
package x402
import (
"crypto/ed25519"
"encoding/base64"
"encoding/json"
"net/http"
"net/http/httptest"
"strings"
"testing"
)
// TestVerifyAndSettle checks that Require verifies then settles against an
// HTTP facilitator and surfaces the settlement in both response headers.
func TestVerifyAndSettle(t *testing.T) {
var verifyHit, settleHit bool
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
switch r.URL.Path {
case "/verify":
verifyHit = true
_ = json.NewEncoder(w).Encode(map[string]any{"isValid": true, "payer": "0xabc"})
case "/settle":
settleHit = true
_ = json.NewEncoder(w).Encode(map[string]any{"success": true, "transaction": "0xdeadbeef"})
default:
http.NotFound(w, r)
}
}))
defer srv.Close()
cfg := Config{PayTo: "0xpay", Network: "eip155:8453", FacilitatorURL: srv.URL}
r := httptest.NewRequest(http.MethodGet, "/tool", nil)
r.Header.Set(PaymentHeader, base64.StdEncoding.EncodeToString([]byte(`{"network":"eip155:8453"}`)))
rec := httptest.NewRecorder()
if !cfg.Require(rec, r, "10000", "chat") {
t.Fatalf("Require returned false; body=%s", rec.Body.String())
}
if !verifyHit || !settleHit {
t.Fatalf("expected both verify and settle to be hit: verify=%v settle=%v", verifyHit, settleHit)
}
if got := rec.Header().Get(PaymentResponseHeader); got != "0xdeadbeef" {
t.Errorf("X-PAYMENT-RESPONSE = %q, want 0xdeadbeef", got)
}
if got := rec.Header().Get(PaymentResponseHeaderV2); got != "0xdeadbeef" {
t.Errorf("PAYMENT-RESPONSE = %q, want 0xdeadbeef", got)
}
}
// TestSettlementFailureChallenges checks that a failed settlement blocks the
// request with a fresh 402 rather than letting it through.
func TestSettlementFailureChallenges(t *testing.T) {
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
switch r.URL.Path {
case "/verify":
_ = json.NewEncoder(w).Encode(map[string]any{"isValid": true})
case "/settle":
_ = json.NewEncoder(w).Encode(map[string]any{"success": false, "errorReason": "insufficient_funds"})
}
}))
defer srv.Close()
cfg := Config{PayTo: "0xpay", FacilitatorURL: srv.URL}
r := httptest.NewRequest(http.MethodGet, "/tool", nil)
r.Header.Set(PaymentHeader, "eyJ4IjoxfQ==")
rec := httptest.NewRecorder()
if cfg.Require(rec, r, "10000", "chat") {
t.Fatal("Require should return false when settlement fails")
}
if rec.Code != http.StatusPaymentRequired {
t.Errorf("status = %d, want 402", rec.Code)
}
}
// TestPaymentSignatureHeaderAccepted checks the v2 request header is honored.
func TestPaymentSignatureHeaderAccepted(t *testing.T) {
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
_ = json.NewEncoder(w).Encode(map[string]any{"isValid": true, "success": true, "transaction": "0x1"})
}))
defer srv.Close()
cfg := Config{PayTo: "0xpay", FacilitatorURL: srv.URL}
r := httptest.NewRequest(http.MethodGet, "/tool", nil)
r.Header.Set(PaymentHeaderV2, "eyJ4IjoxfQ==") // PAYMENT-SIGNATURE only
rec := httptest.NewRecorder()
if !cfg.Require(rec, r, "10000", "chat") {
t.Fatalf("Require should honor PAYMENT-SIGNATURE; body=%s", rec.Body.String())
}
}
// TestRequirementsExtraForKnownNetwork checks the EIP-712 domain is filled in.
func TestRequirementsExtraForKnownNetwork(t *testing.T) {
for _, net := range []string{"base", "eip155:8453"} {
req := Config{PayTo: "0xpay", Network: net}.requirements("10000", "chat")
if req.Extra["name"] == "" || req.Extra["version"] == "" {
t.Errorf("network %q: extra not filled: %v", net, req.Extra)
}
if req.Asset == "" {
t.Errorf("network %q: asset not defaulted", net)
}
}
}
// TestDecodePayment checks the base64 header is decoded to an object for the
// facilitator (which expects the payload object, not the raw string).
func TestDecodePayment(t *testing.T) {
enc := base64.StdEncoding.EncodeToString([]byte(`{"network":"eip155:8453","payload":{"x":1}}`))
obj := decodePayment(enc)
m, ok := obj.(map[string]any)
if !ok {
t.Fatalf("decodePayment did not return an object: %T", obj)
}
if m["network"] != "eip155:8453" {
t.Errorf("decoded network = %v", m["network"])
}
// Non-base64 passes through unchanged.
if got := decodePayment("not-base64!"); got != "not-base64!" {
t.Errorf("passthrough failed: %v", got)
}
}
// TestCDPBearer certifies the CDP JWT is well-formed and its signature verifies.
func TestCDPBearer(t *testing.T) {
pub, priv, _ := ed25519.GenerateKey(nil)
secret := base64.StdEncoding.EncodeToString(priv)
tok, err := cdpBearer("key-id", secret, "POST", "api.cdp.coinbase.com", "/platform/v2/x402/verify")
if err != nil {
t.Fatalf("cdpBearer: %v", err)
}
parts := strings.Split(tok, ".")
if len(parts) != 3 {
t.Fatalf("want 3 JWT segments, got %d", len(parts))
}
sig, _ := base64.RawURLEncoding.DecodeString(parts[2])
if !ed25519.Verify(pub, []byte(parts[0]+"."+parts[1]), sig) {
t.Fatal("JWT signature does not verify")
}
var claims map[string]any
cb, _ := base64.RawURLEncoding.DecodeString(parts[1])
_ = json.Unmarshal(cb, &claims)
if claims["iss"] != "cdp" || claims["uri"] != "POST api.cdp.coinbase.com/platform/v2/x402/verify" {
t.Errorf("bad claims: %v", claims)
}
}
// TestCDPAuthorizeAttachesBearer checks CDP() signs facilitator requests.
func TestCDPAuthorizeAttachesBearer(t *testing.T) {
_, priv, _ := ed25519.GenerateKey(nil)
fac := CDP("key-id", base64.StdEncoding.EncodeToString(priv))
req, _ := http.NewRequest(http.MethodPost, "https://api.cdp.coinbase.com/platform/v2/x402/verify", nil)
if err := fac.Authorize(req); err != nil {
t.Fatalf("authorize: %v", err)
}
if !strings.HasPrefix(req.Header.Get("Authorization"), "Bearer ey") {
t.Errorf("missing Bearer JWT: %q", req.Header.Get("Authorization"))
}
}
var _ Settler = (*HTTPFacilitator)(nil)