Compare commits

..

1 Commits

Author SHA1 Message Date
Codex d46a49b50c Update architect priorities after A2A handoffs
Harness (E2E) / Harnesses (mock LLM) (push) Waiting to run
Harness (E2E) / Provider harnesses (live LLM conformance) (push) Waiting to run
Lint / golangci-lint (push) Waiting to run
Run Tests / Unit Tests (push) Waiting to run
Run Tests / Etcd Integration Tests (push) Waiting to run
2026-06-28 17:33:47 +00:00
17 changed files with 63 additions and 521 deletions
+11 -4
View File
@@ -207,6 +207,17 @@ func (a *agentImpl) Stream(ctx context.Context, message string) (ai.Stream, erro
})
}
// Resume returns the response for a checkpointed agent run. Completed runs are
// returned from the checkpoint without calling the model or replaying tool
// calls; failed or in-progress runs continue from the saved input message.
func Resume(ctx context.Context, ag Agent, runID string) (*Response, error) {
a, ok := ag.(*agentImpl)
if !ok {
return nil, fmt.Errorf("agent resume: unsupported agent implementation %T", ag)
}
return a.resume(ctx, runID)
}
// Pending returns checkpointed agent runs that have not completed. It mirrors
// flow.Pending for startup recovery loops that drain durable agent work.
func Pending(ctx context.Context, ag Agent) ([]flow.Run, error) {
@@ -289,10 +300,6 @@ func (a *agentImpl) askLocked(ctx context.Context, runID, message, parentRunID s
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
+2 -37
View File
@@ -20,9 +20,8 @@ import (
// the discovered service tools. There is no separate harness or graph:
// the LLM calls them like any other tool.
const (
toolPlan = "plan"
toolDelegate = "delegate"
toolHumanInput = "request_input"
toolPlan = "plan"
toolDelegate = "delegate"
)
// builtinTools returns the tool definitions exposed to the model in
@@ -42,18 +41,6 @@ func builtinTools() []ai.Tool {
},
},
},
{
Name: toolHumanInput,
OriginalName: toolHumanInput,
Description: "Pause this agent run when you need missing information, a decision, or other human input before you can continue. " +
"The run is checkpointed as input-required and can be resumed with the human response without losing completed tool history.",
Properties: map[string]any{
"prompt": map[string]any{
"type": "string",
"description": "The specific question, decision, or instruction needed from the human operator.",
},
},
},
{
Name: toolDelegate,
OriginalName: toolDelegate,
@@ -91,9 +78,6 @@ func Builtins(opts ...Option) (tools []ai.Tool, handle func(name string, input m
case toolPlan:
r := a.handlePlan(ai.ToolCall{Name: name, Input: input})
return r.Value, r.Content, true
case toolHumanInput:
r := a.handleHumanInput(ai.ToolCall{Name: name, Input: input})
return r.Value, r.Content, true
case toolDelegate:
r := a.handleDelegate(context.Background(), ai.ToolCall{Name: name, Input: input})
return r.Value, r.Content, true
@@ -162,9 +146,6 @@ func (a *agentImpl) baseHandler() ai.ToolHandler {
return ai.ToolResult{ID: call.ID, Value: out, Content: out}
}
}
if call.Name == toolHumanInput {
return a.handleHumanInput(call)
}
if call.Name == toolDelegate {
return a.handleDelegate(ctx, call)
}
@@ -225,11 +206,6 @@ type approvalPause struct {
Message string
}
type inputPause struct {
OriginalMessage string `json:"original_message"`
Prompt string `json:"prompt"`
}
func (a *agentImpl) approveWrap(next ai.ToolHandler) ai.ToolHandler {
return func(ctx context.Context, call ai.ToolCall) ai.ToolResult {
if a.opts.Approve != nil {
@@ -257,17 +233,6 @@ func (a *agentImpl) handlePlan(call ai.ToolCall) ai.ToolResult {
return ai.ToolResult{ID: call.ID, Value: call.Input, Content: string(data)}
}
// 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)
prompt = strings.TrimSpace(prompt)
if prompt == "" {
prompt = "human input required"
}
a.pause = &approvalPause{Tool: toolHumanInput, Message: prompt}
return refused(call.ID, ai.RefusedApproval, "input-required: "+prompt)
}
// handleDelegate hands a subtask to another agent. Delegate-first:
// if 'to' names a registered agent, it is called via RPC. Otherwise an
// ephemeral sub-agent is created with a fresh, isolated context, asked
+6 -6
View File
@@ -11,15 +11,15 @@ import (
func TestBuiltinTools(t *testing.T) {
tools := builtinTools()
if len(tools) != 3 {
t.Fatalf("builtinTools() = %d tools, want 3", len(tools))
if len(tools) != 2 {
t.Fatalf("builtinTools() = %d tools, want 2", len(tools))
}
names := map[string]bool{}
for _, tl := range tools {
names[tl.Name] = true
}
if !names[toolPlan] || !names[toolDelegate] || !names[toolHumanInput] {
t.Errorf("builtin tools = %v, want plan, request_input, and delegate", names)
if !names[toolPlan] || !names[toolDelegate] {
t.Errorf("builtin tools = %v, want plan and delegate", names)
}
}
@@ -109,8 +109,8 @@ func TestBuiltinsAccessor(t *testing.T) {
WithRegistry(registry.NewMemoryRegistry()),
)
if len(tools) != 3 {
t.Fatalf("Builtins() returned %d tools, want 3", len(tools))
if len(tools) != 2 {
t.Fatalf("Builtins() returned %d tools, want 2", len(tools))
}
// A name that isn't a built-in falls through (ok == false).
-60
View File
@@ -13,7 +13,6 @@ import (
const (
agentAskStep = "ask"
agentApprovalStep = "approval"
agentInputStep = "input-required"
)
func (a *agentImpl) newCheckpointRun(runID, message, parentRunID string, existing *flow.Run) flow.Run {
@@ -52,17 +51,6 @@ func (a *agentImpl) saveRun(ctx context.Context, run flow.Run) error {
return nil
}
// Resume returns the response for a checkpointed agent run. Completed runs are
// returned from the checkpoint without calling the model or replaying tool
// calls; failed or in-progress runs continue from the saved input message.
func Resume(ctx context.Context, ag Agent, runID string) (*Response, error) {
a, ok := ag.(*agentImpl)
if !ok {
return nil, fmt.Errorf("agent resume: unsupported agent implementation %T", ag)
}
return a.resume(ctx, runID)
}
func (a *agentImpl) resume(ctx context.Context, runID string) (*Response, error) {
if a.opts.Checkpoint == nil {
return nil, fmt.Errorf("agent %s has no checkpoint configured", a.opts.Name)
@@ -75,9 +63,6 @@ func (a *agentImpl) resume(ctx context.Context, runID string) (*Response, error)
return nil, fmt.Errorf("agent run %s not found", runID)
}
if run.Status == "paused" {
if run.State.Stage == agentInputStep {
return nil, fmt.Errorf("agent run %s is input-required; resume with ResumeInput", runID)
}
run.Status = "running"
run.State.Stage = agentAskStep
}
@@ -98,51 +83,6 @@ func (a *agentImpl) resume(ctx context.Context, runID string) (*Response, error)
return a.askLocked(ctx, run.ID, message, parentID, &run)
}
// ResumeInput resumes a checkpointed agent run that paused via the built-in
// request_input tool. The supplied input is appended to the original request so
// the same run can continue with durable checkpoint and completed tool history.
func ResumeInput(ctx context.Context, ag Agent, runID, input string) (*Response, error) {
a, ok := ag.(*agentImpl)
if !ok {
return nil, fmt.Errorf("agent resume input: unsupported agent implementation %T", ag)
}
return a.resumeInput(ctx, runID, input)
}
func (a *agentImpl) resumeInput(ctx context.Context, runID, input string) (*Response, error) {
if a.opts.Checkpoint == nil {
return nil, fmt.Errorf("agent %s has no checkpoint configured", a.opts.Name)
}
run, ok, err := a.opts.Checkpoint.Load(ctx, runID)
if err != nil {
return nil, err
}
if !ok {
return nil, fmt.Errorf("agent run %s not found", runID)
}
if run.Status != "paused" || run.State.Stage != agentInputStep {
return nil, fmt.Errorf("agent run %s is not waiting for human input", runID)
}
var p inputPause
if err := run.State.Scan(&p); err != nil {
return nil, fmt.Errorf("agent run %s input state decode: %w", runID, err)
}
message := p.OriginalMessage
if message == "" {
message = string(run.State.Data)
}
message += "\n\nHuman input: " + input
run.Status = "running"
run.State.Stage = agentAskStep
run.State.Data = []byte(message)
a.mu.Lock()
defer a.mu.Unlock()
if a.model == nil {
a.setup()
}
return a.askLocked(ctx, run.ID, message, run.ParentID, &run)
}
func (a *agentImpl) pending(ctx context.Context) ([]flow.Run, error) {
if a.opts.Checkpoint == nil {
return nil, nil
-59
View File
@@ -3,7 +3,6 @@ package agent
import (
"context"
"errors"
"strings"
"testing"
"go-micro.dev/v6/ai"
@@ -119,64 +118,6 @@ func TestPendingReturnsUnfinishedAgentRuns(t *testing.T) {
}
}
func TestHumanInputPauseResumesSameRunWithInput(t *testing.T) {
ctx := context.Background()
cp := flow.StoreCheckpoint(store.NewStore(), "input-agent")
calls := 0
fakeGen = func(ctx context.Context, opts ai.Options, req *ai.Request) (*ai.Response, error) {
calls++
if calls == 1 {
if opts.ToolHandler != nil {
opts.ToolHandler(ctx, ai.ToolCall{ID: "input-1", Name: toolHumanInput, Input: map[string]any{"prompt": "Which region should I deploy to?"}})
}
return &ai.Response{Reply: "waiting"}, nil
}
if !strings.Contains(req.Prompt, "Human input: us-east-1") {
t.Fatalf("resumed prompt = %q, want human input", req.Prompt)
}
return &ai.Response{Reply: "deploying to us-east-1"}, nil
}
defer func() { fakeGen = nil }()
a := newTestAgent(Name("input-agent"), WithCheckpoint(cp))
_, err := a.Ask(ctx, "deploy the service")
if err == nil {
t.Fatal("Ask succeeded, want input-required pause")
}
runs, err := Pending(ctx, a)
if err != nil {
t.Fatalf("Pending: %v", err)
}
if len(runs) != 1 || runs[0].Status != "paused" || runs[0].State.Stage != agentInputStep {
t.Fatalf("paused runs = %#v, want one input-required run", runs)
}
var pause inputPause
if err := runs[0].State.Scan(&pause); err != nil {
t.Fatalf("Scan pause: %v", err)
}
if pause.OriginalMessage != "deploy the service" || pause.Prompt != "Which region should I deploy to?" {
t.Fatalf("pause = %#v", pause)
}
if _, err := Resume(ctx, a, runs[0].ID); err == nil || !strings.Contains(err.Error(), "ResumeInput") {
t.Fatalf("Resume input-required err = %v, want guidance", err)
}
resp, err := ResumeInput(ctx, a, runs[0].ID, "us-east-1")
if err != nil {
t.Fatalf("ResumeInput: %v", err)
}
if resp.RunID != runs[0].ID || resp.Reply != "deploying to us-east-1" {
t.Fatalf("response = %#v", resp)
}
loaded, ok, err := cp.Load(ctx, runs[0].ID)
if err != nil || !ok {
t.Fatalf("Load resumed run ok=%v err=%v", ok, err)
}
if loaded.Status != "done" {
t.Fatalf("resumed run status = %q, want done", loaded.Status)
}
}
func TestApprovalDenialPausesCheckpointedRunAndResumeContinues(t *testing.T) {
ctx := context.Background()
cp := flow.StoreCheckpoint(store.NewStore(), "approval-agent")
+8 -34
View File
@@ -3,7 +3,6 @@ package agent
import (
"encoding/json"
"fmt"
"sort"
"strings"
"sync"
@@ -124,31 +123,17 @@ func (m *storeMemory) Recall(query string, limit int) []ai.Message {
limit = 5
}
terms := recallTerms(query)
type match struct {
msg ai.Message
score int
index int
}
matches := make([]match, 0, len(m.archive))
for i := len(m.archive) - 1; i >= 0; i-- {
var out []ai.Message
for i := len(m.archive) - 1; i >= 0 && len(out) < limit; i-- {
msg := m.archive[i]
if score := recallScore(msg, terms); score > 0 {
matches = append(matches, match{msg: msg, score: score, index: i})
text := strings.ToLower(fmt.Sprint(msg.Content))
for _, term := range terms {
if strings.Contains(text, term) {
out = append(out, msg)
break
}
}
}
sort.SliceStable(matches, func(i, j int) bool {
if matches[i].score != matches[j].score {
return matches[i].score > matches[j].score
}
return matches[i].index > matches[j].index
})
if len(matches) > limit {
matches = matches[:limit]
}
out := make([]ai.Message, 0, len(matches))
for _, match := range matches {
out = append(out, match.msg)
}
return out
}
@@ -243,17 +228,6 @@ func compactText(s string, max int) string {
return s
}
func recallScore(msg ai.Message, terms []string) int {
text := strings.ToLower(fmt.Sprint(msg.Content))
score := 0
for _, term := range terms {
if strings.Contains(text, term) {
score++
}
}
return score
}
func recallTerms(query string) []string {
seen := map[string]bool{}
var terms []string
-40
View File
@@ -62,46 +62,6 @@ func TestWithMemoryUsed(t *testing.T) {
}
}
func TestCompactingMemoryRecallRanksSpecificMatches(t *testing.T) {
m := NewCompactingMemory(store.NewMemoryStore(), "agent/rank/history", 3, 1).(MemoryRecall)
writer := m.(Memory)
writer.Add("user", "alpha budget is 42")
writer.Add("assistant", "noted")
writer.Add("user", "beta budget is 7")
writer.Add("assistant", "noted")
writer.Add("user", "alpha owner is sam")
recalled := m.Recall("alpha budget", 2)
if len(recalled) == 0 {
t.Fatal("expected recalled messages")
}
if got := recalled[0].Content.(string); !strings.Contains(got, "alpha budget is 42") {
t.Fatalf("top recall = %q, want alpha budget match", got)
}
}
func TestCompactingMemoryArchivePersistsAndReloads(t *testing.T) {
st := store.NewMemoryStore()
m := NewCompactingMemory(st, "agent/reload/history", 3, 1)
m.Add("user", "alpha budget is 42")
m.Add("assistant", "noted")
m.Add("user", "beta budget is 7")
m.Add("assistant", "noted")
reloaded := NewCompactingMemory(st, "agent/reload/history", 3, 1)
recall, ok := reloaded.(MemoryRecall)
if !ok {
t.Fatal("compacting memory should support recall")
}
recalled := recall.Recall("alpha budget", 1)
if len(recalled) != 1 {
t.Fatalf("recalled %d messages, want 1", len(recalled))
}
if got := recalled[0].Content.(string); !strings.Contains(got, "alpha budget is 42") {
t.Fatalf("reloaded recall = %q, want alpha budget", got)
}
}
// A custom tool is offered to the model and dispatched to its handler.
func TestWithToolExposedAndDispatched(t *testing.T) {
var got map[string]any
+9 -28
View File
@@ -77,9 +77,11 @@ func (p *Provider) Generate(ctx context.Context, req *ai.Request, opts ...ai.Gen
// Build initial request
apiReq := map[string]any{
"model": p.opts.Model,
"max_tokens": anthropicMaxTokens(p.opts),
"max_tokens": 8192,
"system": req.SystemPrompt,
"messages": threadAnthropicMessages(req),
"messages": []map[string]any{
{"role": "user", "content": req.Prompt},
},
}
if len(anthropicTools) > 0 {
@@ -99,9 +101,10 @@ func (p *Provider) Generate(ctx context.Context, req *ai.Request, opts ...ai.Gen
// Tool execution loop: execute tools, send results back, repeat
// until the model responds with text only (no more tool calls)
messages := append(threadAnthropicMessages(req),
map[string]any{"role": "assistant", "content": cleanContent(rawContent)},
)
messages := []map[string]any{
{"role": "user", "content": req.Prompt},
{"role": "assistant", "content": cleanContent(rawContent)},
}
pendingCalls := resp.ToolCalls
@@ -124,7 +127,7 @@ func (p *Provider) Generate(ctx context.Context, req *ai.Request, opts ...ai.Gen
followUpReq := map[string]any{
"model": p.opts.Model,
"max_tokens": anthropicMaxTokens(p.opts),
"max_tokens": 8192,
"system": req.SystemPrompt,
"messages": messages,
}
@@ -267,25 +270,3 @@ func cleanContent(raw any) any {
}
return cleaned
}
// threadAnthropicMessages builds the Anthropic messages array from the
// conversation history (req.Messages) followed by the current prompt. The
// system prompt is sent separately via the top-level "system" field.
func threadAnthropicMessages(req *ai.Request) []map[string]any {
msgs := make([]map[string]any, 0, len(req.Messages)+1)
for _, m := range req.Messages {
msgs = append(msgs, map[string]any{"role": m.Role, "content": m.Content})
}
if req.Prompt != "" {
msgs = append(msgs, map[string]any{"role": "user", "content": req.Prompt})
}
return msgs
}
func anthropicMaxTokens(o ai.Options) int {
if o.MaxTokens > 0 {
return o.MaxTokens
}
return 8192
}
+1 -9
View File
@@ -91,21 +91,13 @@ func (p *Provider) Generate(ctx context.Context, req *ai.Request, opts ...ai.Gen
messages := []map[string]any{
{"role": "system", "content": req.SystemPrompt},
}
for _, m := range req.Messages {
messages = append(messages, map[string]any{"role": m.Role, "content": m.Content})
}
if req.Prompt != "" {
messages = append(messages, map[string]any{"role": "user", "content": req.Prompt})
{"role": "user", "content": req.Prompt},
}
apiReq := map[string]any{
"model": p.opts.Model,
"messages": messages,
}
if p.opts.MaxTokens > 0 {
apiReq["max_tokens"] = p.opts.MaxTokens
}
if len(tools) > 0 {
apiReq["tools"] = tools
+2 -18
View File
@@ -85,12 +85,7 @@ func (p *Provider) Generate(ctx context.Context, req *ai.Request, opts ...ai.Gen
// Build messages
messages := []map[string]any{
{"role": "system", "content": req.SystemPrompt},
}
for _, m := range req.Messages {
messages = append(messages, map[string]any{"role": m.Role, "content": m.Content})
}
if req.Prompt != "" {
messages = append(messages, map[string]any{"role": "user", "content": req.Prompt})
{"role": "user", "content": req.Prompt},
}
// Build initial request
@@ -98,9 +93,6 @@ func (p *Provider) Generate(ctx context.Context, req *ai.Request, opts ...ai.Gen
"model": p.opts.Model,
"messages": messages,
}
if p.opts.MaxTokens > 0 {
apiReq["max_tokens"] = p.opts.MaxTokens
}
if len(openaiTools) > 0 {
apiReq["tools"] = openaiTools
@@ -154,21 +146,13 @@ func (p *Provider) Generate(ctx context.Context, req *ai.Request, opts ...ai.Gen
func (p *Provider) Stream(ctx context.Context, req *ai.Request, opts ...ai.GenerateOption) (ai.Stream, error) {
messages := []map[string]any{
{"role": "system", "content": req.SystemPrompt},
}
for _, m := range req.Messages {
messages = append(messages, map[string]any{"role": m.Role, "content": m.Content})
}
if req.Prompt != "" {
messages = append(messages, map[string]any{"role": "user", "content": req.Prompt})
{"role": "user", "content": req.Prompt},
}
apiReq := map[string]any{
"model": p.opts.Model,
"messages": messages,
"stream": true,
}
if p.opts.MaxTokens > 0 {
apiReq["max_tokens"] = p.opts.MaxTokens
}
reqBody, err := json.Marshal(apiReq)
if err != nil {
return nil, fmt.Errorf("failed to marshal stream request: %w", err)
-10
View File
@@ -16,8 +16,6 @@ type Options struct {
BaseURL string
// ToolHandler handles tool calls (optional, for automatic tool execution)
ToolHandler ToolHandler
// MaxTokens caps the length of the response (0 = provider default)
MaxTokens int
}
// GenerateOptions for generate call
@@ -93,11 +91,3 @@ func WithTools(t *Tools) Option {
}
}
}
// WithMaxTokens caps the number of tokens in the response. 0 leaves the
// provider default in place.
func WithMaxTokens(n int) Option {
return func(o *Options) {
o.MaxTokens = n
}
}
+3 -16
View File
@@ -42,28 +42,14 @@ func Deploy(c *cli.Context) error {
return showDeployHelp()
}
target, remotePath := resolveDeployTarget(c, target, cfg)
return deploySSH(c, target, cfg, remotePath)
}
func resolveDeployTarget(c *cli.Context, target string, cfg *config.Config) (string, string) {
remotePath := c.String("path")
if remotePath == "" {
remotePath = defaultRemotePath
}
// Check if target is a named target from config
if cfg != nil {
if dt, ok := cfg.Deploy[target]; ok {
target = dt.SSH
if dt.Path != "" && !c.IsSet("path") {
remotePath = dt.Path
}
}
}
return target, remotePath
return deploySSH(c, target, cfg)
}
func showDeployHelp() error {
@@ -96,7 +82,7 @@ func showDeployTargets(cfg *config.Config) error {
return fmt.Errorf("%s", sb.String())
}
func deploySSH(c *cli.Context, target string, cfg *config.Config, remotePath string) error {
func deploySSH(c *cli.Context, target string, cfg *config.Config) error {
dir := c.Args().Get(1)
if dir == "" {
dir = "."
@@ -112,6 +98,7 @@ func deploySSH(c *cli.Context, target string, cfg *config.Config, remotePath str
cfg, _ = config.Load(absDir)
}
remotePath := c.String("path")
if remotePath == "" {
remotePath = defaultRemotePath
}
-120
View File
@@ -1,120 +0,0 @@
package deploy
import (
"flag"
"os"
"strings"
"testing"
"github.com/urfave/cli/v2"
"go-micro.dev/v6/cmd/micro/run/config"
)
func newDeployTestContext(t *testing.T, args ...string) *cli.Context {
t.Helper()
set := flag.NewFlagSet("deploy", flag.ContinueOnError)
set.String("path", defaultRemotePath, "")
set.String("ssh", "", "")
set.String("service", "", "")
set.Bool("build", false, "")
if err := set.Parse(args); err != nil {
t.Fatalf("parse flags: %v", err)
}
return cli.NewContext(cli.NewApp(), set, nil)
}
func TestDeployNoTargetExplainsInitAndDeployHandoff(t *testing.T) {
err := showDeployHelp()
if err == nil {
t.Fatal("expected missing target guidance")
}
msg := err.Error()
for _, want := range []string{
"no deployment target specified",
"sudo micro init --server",
"micro deploy user@your-server",
"deploy prod",
} {
if !strings.Contains(msg, want) {
t.Fatalf("missing %q in guidance:\n%s", want, msg)
}
}
}
func TestDeployListsConfiguredTargetsWhenNoTargetProvided(t *testing.T) {
err := showDeployTargets(&config.Config{Deploy: map[string]*config.DeployTarget{
"prod": {Name: "prod", SSH: "deploy@prod.example.com"},
"staging": {Name: "staging", SSH: "deploy@staging.example.com"},
}})
if err == nil {
t.Fatal("expected configured target guidance")
}
msg := err.Error()
for _, want := range []string{
"Available deploy targets:",
"prod -> deploy@prod.example.com",
"staging -> deploy@staging.example.com",
"micro deploy <target>",
} {
if !strings.Contains(msg, want) {
t.Fatalf("missing %q in configured target guidance:\n%s", want, msg)
}
}
}
func TestResolveDeployTargetUsesConfigTargetAndPath(t *testing.T) {
ctx := newDeployTestContext(t, "prod")
cfg := &config.Config{Deploy: map[string]*config.DeployTarget{
"prod": {Name: "prod", SSH: "deploy@prod.example.com", Path: "/srv/micro"},
}}
target, remotePath := resolveDeployTarget(ctx, ctx.Args().First(), cfg)
if target != "deploy@prod.example.com" {
t.Fatalf("target = %q, want configured SSH", target)
}
if remotePath != "/srv/micro" {
t.Fatalf("remotePath = %q, want configured path", remotePath)
}
}
func TestResolveDeployTargetAllowsCLIPathOverride(t *testing.T) {
ctx := newDeployTestContext(t, "--path", "/tmp/micro", "prod")
cfg := &config.Config{Deploy: map[string]*config.DeployTarget{
"prod": {Name: "prod", SSH: "deploy@prod.example.com", Path: "/srv/micro"},
}}
target, remotePath := resolveDeployTarget(ctx, ctx.Args().First(), cfg)
if target != "deploy@prod.example.com" {
t.Fatalf("target = %q, want configured SSH", target)
}
if remotePath != "/tmp/micro" {
t.Fatalf("remotePath = %q, want CLI override", remotePath)
}
}
func TestDeployConfigParserSupportsDeployTargets(t *testing.T) {
dir := t.TempDir()
path := dir + "/micro.mu"
content := `service api
path ./api
deploy prod
ssh deploy@prod.example.com
path /srv/micro
`
if err := os.WriteFile(path, []byte(content), 0644); err != nil {
t.Fatalf("write config: %v", err)
}
cfg, err := config.ParseMu(path)
if err != nil {
t.Fatalf("parse config: %v", err)
}
prod := cfg.Deploy["prod"]
if prod == nil {
t.Fatal("missing prod deploy target")
}
if prod.SSH != "deploy@prod.example.com" || prod.Path != "/srv/micro" {
t.Fatalf("deploy target = %#v", prod)
}
}
-36
View File
@@ -1,36 +0,0 @@
# Agent Human Input Pause/Resume
Agents can pause a durable run when the model needs a human decision before it
can continue. This keeps the services → agents → workflows lifecycle in one
runtime: services expose tools, the agent decides it needs operator input, and
the same checkpointed run resumes once that input arrives.
## Pattern
```go
cp := flow.StoreCheckpoint(nil, "deploy-agent")
ag := agent.New(
agent.Name("deploy-agent"),
agent.WithCheckpoint(cp),
)
resp, err := ag.Ask(ctx, "Deploy the service")
if err != nil {
// If the model called the built-in request_input tool, the run is saved as
// paused/input-required instead of losing state or completing early.
pending, _ := agent.Pending(ctx, ag)
runID := pending[0].ID
// Later, after an operator supplies the missing answer, the same run ID
// continues with the original prompt, human input, memory, and completed
// tool history intact.
resp, err = agent.ResumeInput(ctx, ag, runID, "Deploy to us-east-1")
}
_ = resp
```
The model sees a built-in `request_input` tool with a `prompt` argument. When it
calls that tool, Go Micro persists the run with status `paused` and stage
`input-required`. Plain `agent.Resume` continues to support completed, failed,
and approval-paused runs; input-required runs are resumed with
`agent.ResumeInput` so the human response is explicit.
+19 -31
View File
@@ -212,39 +212,37 @@ func waitFor(reg registry.Registry, names ...string) {
}
}
func runSupport(provider string) error {
func main() {
provider := flag.String("provider", "mock", "LLM provider: mock (default), anthropic, openai, ...")
flag.Parse()
apiKey := ""
if provider == "mock" {
if *provider == "mock" {
ai.Register("mock", newMock)
} else if apiKey = providerKey(provider); apiKey == "" {
return fmt.Errorf("no API key for provider %q — set MICRO_AI_API_KEY or the provider's key env", provider)
} else if apiKey = providerKey(*provider); apiKey == "" {
fmt.Printf("no API key for provider %q — set MICRO_AI_API_KEY or the provider's key env\n", *provider)
os.Exit(1)
}
fmt.Printf("\n\033[1mSupport desk (provider: %s)\033[0m\n\n", provider)
fmt.Printf("\n\033[1mSupport desk (provider: %s)\033[0m\n\n", *provider)
// Shared in-memory infrastructure so the demo runs in one process.
reg := registry.NewMemoryRegistry()
br := broker.NewMemoryBroker()
if err := br.Connect(); err != nil {
return fmt.Errorf("broker connect: %w", err)
fmt.Println("broker connect:", err)
os.Exit(1)
}
cl := client.NewClient(client.Registry(reg), client.Selector(selector.NewSelector(selector.Registry(reg))))
// Services.
tickets := new(TicketService)
notify := new(NotifyService)
var services []service.Service
for name, h := range map[string]any{"customers": new(CustomerService), "tickets": tickets, "notify": notify} {
svc := service.New(service.Name(name), service.Address("127.0.0.1:0"), service.Registry(reg), service.Client(cl), service.HandleSignal(false))
svc := service.New(service.Name(name), service.Registry(reg), service.Client(cl))
_ = svc.Handle(h)
services = append(services, svc)
go svc.Run()
}
defer func() {
for _, svc := range services {
_ = svc.Server().Stop()
}
}()
// The support agent manages the three services. The approval gate is
// the human-in-the-loop: it can read and triage freely, but emailing a
@@ -252,11 +250,10 @@ func runSupport(provider string) error {
// it for a person or a policy; here we approve and log.
support := agent.New(
agent.Name("support"),
agent.Address("127.0.0.1:0"),
agent.Services("customers", "tickets", "notify"),
agent.Prompt("You are a support agent. For each ticket, look up the customer, set an "+
"appropriate priority, and reply to them. Escalate billing issues."),
agent.Provider(provider), agent.APIKey(apiKey),
agent.Provider(*provider), agent.APIKey(apiKey),
agent.ApproveTool(func(tool string, input map[string]any) (bool, string) {
if strings.Contains(tool, "Send") {
fmt.Printf(" \033[33m▣ approval gate\033[0m %s(%v) — approved\n", tool, input["to"])
@@ -278,7 +275,8 @@ func runSupport(provider string) error {
flow.Prompt("A new support ticket arrived: {{.Data}}. Handle it."),
)
if err := intake.Register(reg, br, cl); err != nil {
return fmt.Errorf("flow register: %w", err)
fmt.Println("flow register:", err)
os.Exit(1)
}
defer intake.Stop()
@@ -289,7 +287,8 @@ func runSupport(provider string) error {
fmt.Println("\033[1m> event:\033[0m events.ticket.created", string(body))
fmt.Println()
if err := br.Publish("events.ticket.created", &broker.Message{Body: body}); err != nil {
return fmt.Errorf("publish: %w", err)
fmt.Println("publish:", err)
os.Exit(1)
}
// Wait for the agent to act.
@@ -306,18 +305,7 @@ func runSupport(provider string) error {
}
if notify.sent >= 1 {
fmt.Println("\n\033[32m✓ ticket triaged and the customer was replied to — triggered by an event\033[0m")
return nil
}
fmt.Println("\n\033[31m✗ the agent did not complete the triage\033[0m")
return fmt.Errorf("support agent did not complete triage")
}
func main() {
provider := flag.String("provider", "mock", "LLM provider: mock (default), anthropic, openai, ...")
flag.Parse()
if err := runSupport(*provider); err != nil {
fmt.Println(err)
os.Exit(1)
} else {
fmt.Println("\n\033[31m✗ the agent did not complete the triage\033[0m")
}
}
-9
View File
@@ -1,9 +0,0 @@
package main
import "testing"
func TestRunSupportMockSmoke(t *testing.T) {
if err := runSupport("mock"); err != nil {
t.Fatalf("support example failed: %v", err)
}
}
+2 -4
View File
@@ -19,11 +19,9 @@ redirect the loop; direction always wins.
items the loop can auto-merge): brand/positioning copy, breaking public-API
changes, architectural rewrites. Those go to the human.
## Developer experience (ranked)
## Later (ranked)
1. **Schedule cross-provider agent conformance** ([#3295](https://github.com/micro/go-micro/issues/3295)) — after the 0→hero reference and provider-focused unit coverage shipped, the highest-value remaining Now-roadmap hardening gap is a scheduled, key-gated provider matrix that proves the same agent/tool workflow keeps working across supported models without blocking contributors who lack secrets.
2. **Harden agent failure and cancellation semantics** ([#3296](https://github.com/micro/go-micro/issues/3296)) — the harness is increasingly durable, streaming, observable, and human-in-the-loop; the next operability seam is making timeouts, cancellation, rate-limit errors, and retry/backoff behavior predictable across agent, AI provider, service-tool, and flow boundaries.
3. **Expose run inspection in the CLI inner loop** ([#3297](https://github.com/micro/go-micro/issues/3297)) — the canon promises scaffold → run → chat → inspect → deploy, and recent work has strengthened scaffold/run/chat/deploy; inspection remains the most visible DX gap for turning agent/flow activity into actionable breadcrumbs during local development.
1. **Add agent memory summarization and retrieval** ([#3273](https://github.com/micro/go-micro/issues/3273)) — with A2A resubscribe/input-required handoffs now shipped, the highest-value remaining roadmap gap is durable memory beyond a fixed context buffer: compact older turns/tool results and retrieve relevant prior facts using the existing agent/store primitives so longer-running agents remain coherent without becoming a separate product layer.
_Seeded by Claude Code from the roadmap + open issues; thereafter maintained by the
architecture-review pass._