ai/atlascloud: implement streaming (#3301)

Atlas exposes an OpenAI-compatible SSE chat endpoint; emit content
deltas as they arrive instead of returning ErrStreamingUnsupported.

agent/agent.go

@@ -207,17 +207,6 @@ 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) {
@@ -300,6 +289,10 @@ 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

agent/builtin.go

@@ -20,8 +20,9 @@ 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"
+	toolPlan       = "plan"
+	toolDelegate   = "delegate"
+	toolHumanInput = "request_input"
 )
 
 // builtinTools returns the tool definitions exposed to the model in
@@ -41,6 +42,18 @@ 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,
@@ -78,6 +91,9 @@ 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
@@ -146,6 +162,9 @@ 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)
 		}
@@ -206,6 +225,11 @@ 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 {
@@ -233,6 +257,17 @@ 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

agent/builtin_test.go

@@ -11,15 +11,15 @@ import (
 
 func TestBuiltinTools(t *testing.T) {
 	tools := builtinTools()
-	if len(tools) != 2 {
-		t.Fatalf("builtinTools() = %d tools, want 2", len(tools))
+	if len(tools) != 3 {
+		t.Fatalf("builtinTools() = %d tools, want 3", len(tools))
 	}
 	names := map[string]bool{}
 	for _, tl := range tools {
 		names[tl.Name] = true
 	}
-	if !names[toolPlan] || !names[toolDelegate] {
-		t.Errorf("builtin tools = %v, want plan and delegate", names)
+	if !names[toolPlan] || !names[toolDelegate] || !names[toolHumanInput] {
+		t.Errorf("builtin tools = %v, want plan, request_input, and delegate", names)
 	}
 }
 
@@ -109,8 +109,8 @@ func TestBuiltinsAccessor(t *testing.T) {
 		WithRegistry(registry.NewMemoryRegistry()),
 	)
 
-	if len(tools) != 2 {
-		t.Fatalf("Builtins() returned %d tools, want 2", len(tools))
+	if len(tools) != 3 {
+		t.Fatalf("Builtins() returned %d tools, want 3", len(tools))
 	}
 
 	// A name that isn't a built-in falls through (ok == false).

agent/checkpoint.go

@@ -13,6 +13,7 @@ import (
 const (
 	agentAskStep      = "ask"
 	agentApprovalStep = "approval"
+	agentInputStep    = "input-required"
 )
 
 func (a *agentImpl) newCheckpointRun(runID, message, parentRunID string, existing *flow.Run) flow.Run {
@@ -51,6 +52,17 @@ 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)
@@ -63,6 +75,9 @@ 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
 	}
@@ -83,6 +98,51 @@ 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

agent/checkpoint_test.go

@@ -3,6 +3,7 @@ package agent
 import (
 	"context"
 	"errors"
+	"strings"
 	"testing"
 
 	"go-micro.dev/v6/ai"
@@ -118,6 +119,64 @@ 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")

ai/anthropic/anthropic.go

@@ -77,11 +77,9 @@ 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": 8192,
+		"max_tokens": anthropicMaxTokens(p.opts),
 		"system":     req.SystemPrompt,
-		"messages": []map[string]any{
-			{"role": "user", "content": req.Prompt},
-		},
+		"messages":   threadAnthropicMessages(req),
 	}
 
 	if len(anthropicTools) > 0 {
@@ -101,10 +99,9 @@ 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 := []map[string]any{
-		{"role": "user", "content": req.Prompt},
-		{"role": "assistant", "content": cleanContent(rawContent)},
-	}
+	messages := append(threadAnthropicMessages(req),
+		map[string]any{"role": "assistant", "content": cleanContent(rawContent)},
+	)
 
 	pendingCalls := resp.ToolCalls
 
@@ -127,7 +124,7 @@ func (p *Provider) Generate(ctx context.Context, req *ai.Request, opts ...ai.Gen
 
 		followUpReq := map[string]any{
 			"model":      p.opts.Model,
-			"max_tokens": 8192,
+			"max_tokens": anthropicMaxTokens(p.opts),
 			"system":     req.SystemPrompt,
 			"messages":   messages,
 		}
@@ -270,3 +267,24 @@ 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
+}

ai/atlascloud/atlascloud.go

@@ -20,6 +20,7 @@
 package atlascloud
 
 import (
+	"bufio"
 	"bytes"
 	"context"
 	"encoding/json"
@@ -91,13 +92,21 @@ func (p *Provider) Generate(ctx context.Context, req *ai.Request, opts ...ai.Gen
 
 	messages := []map[string]any{
 		{"role": "system", "content": req.SystemPrompt},
-		{"role": "user", "content": req.Prompt},
+	}
+	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})
 	}
 
 	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
@@ -142,8 +151,97 @@ func (p *Provider) Generate(ctx context.Context, req *ai.Request, opts ...ai.Gen
 	return resp, nil
 }
 
+// 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) {
-	return nil, fmt.Errorf("%w: atlascloud provider", ai.ErrStreamingUnsupported)
+	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})
+	}
+	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)
+	}
+	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, fmt.Errorf("failed to create stream request: %w", err)
+	}
+	httpReq.Header.Set("Content-Type", "application/json")
+	httpReq.Header.Set("Accept", "text/event-stream")
+	httpReq.Header.Set("Authorization", "Bearer "+p.opts.APIKey)
+
+	httpResp, err := http.DefaultClient.Do(httpReq)
+	if err != nil {
+		return nil, fmt.Errorf("stream API request failed: %w", err)
+	}
+	if httpResp.StatusCode != http.StatusOK {
+		defer httpResp.Body.Close()
+		respBody, _ := io.ReadAll(httpResp.Body)
+		return nil, fmt.Errorf("stream API error (%s): %s", httpResp.Status, string(respBody))
+	}
+	return &atlasStream{body: httpResp.Body, scanner: bufio.NewScanner(httpResp.Body)}, nil
+}
+
+type atlasStream struct {
+	body    io.ReadCloser
+	scanner *bufio.Scanner
+	closed  bool
+}
+
+func (s *atlasStream) Recv() (*ai.Response, error) {
+	for s.scanner.Scan() {
+		line := strings.TrimSpace(s.scanner.Text())
+		if line == "" || strings.HasPrefix(line, ":") {
+			continue
+		}
+		if !strings.HasPrefix(line, "data:") {
+			continue
+		}
+		data := strings.TrimSpace(strings.TrimPrefix(line, "data:"))
+		if data == "[DONE]" {
+			return nil, io.EOF
+		}
+		var chunk struct {
+			Choices []struct {
+				Delta struct {
+					Content string `json:"content"`
+				} `json:"delta"`
+			} `json:"choices"`
+		}
+		if err := json.Unmarshal([]byte(data), &chunk); err != nil {
+			return nil, fmt.Errorf("failed to parse stream chunk: %w", err)
+		}
+		if len(chunk.Choices) == 0 || chunk.Choices[0].Delta.Content == "" {
+			continue
+		}
+		return &ai.Response{Reply: chunk.Choices[0].Delta.Content}, nil
+	}
+	if err := s.scanner.Err(); err != nil {
+		return nil, err
+	}
+	return nil, io.EOF
+}
+
+func (s *atlasStream) Close() error {
+	if s.closed {
+		return nil
+	}
+	s.closed = true
+	return s.body.Close()
 }
 
 func (p *Provider) callAPI(ctx context.Context, req map[string]any) (*ai.Response, map[string]any, error) {

ai/openai/openai.go

@@ -85,7 +85,12 @@ func (p *Provider) Generate(ctx context.Context, req *ai.Request, opts ...ai.Gen
 	// Build messages
 	messages := []map[string]any{
 		{"role": "system", "content": req.SystemPrompt},
-		{"role": "user", "content": req.Prompt},
+	}
+	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})
 	}
 
 	// Build initial request
@@ -93,6 +98,9 @@ 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
@@ -146,13 +154,21 @@ 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},
-		{"role": "user", "content": req.Prompt},
+	}
+	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})
 	}
 	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)

ai/options.go

@@ -16,6 +16,8 @@ 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
@@ -91,3 +93,11 @@ 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
+	}
+}

cmd/micro/cli/deploy/deploy.go

@@ -42,14 +42,28 @@ 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 deploySSH(c, target, cfg)
+	return target, remotePath
 }
 
 func showDeployHelp() error {
@@ -82,7 +96,7 @@ func showDeployTargets(cfg *config.Config) error {
 	return fmt.Errorf("%s", sb.String())
 }
 
-func deploySSH(c *cli.Context, target string, cfg *config.Config) error {
+func deploySSH(c *cli.Context, target string, cfg *config.Config, remotePath string) error {
 	dir := c.Args().Get(1)
 	if dir == "" {
 		dir = "."
@@ -98,7 +112,6 @@ func deploySSH(c *cli.Context, target string, cfg *config.Config) error {
 		cfg, _ = config.Load(absDir)
 	}
 
-	remotePath := c.String("path")
 	if remotePath == "" {
 		remotePath = defaultRemotePath
 	}

cmd/micro/cli/deploy/deploy_test.go

@@ -0,0 +1,120 @@
+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)
+	}
+}

examples/agent-human-input/README.md

@@ -0,0 +1,36 @@
+# 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.

examples/support/main.go

@@ -212,37 +212,39 @@ func waitFor(reg registry.Registry, names ...string) {
 	}
 }
 
-func main() {
-	provider := flag.String("provider", "mock", "LLM provider: mock (default), anthropic, openai, ...")
-	flag.Parse()
-
+func runSupport(provider string) error {
 	apiKey := ""
-	if *provider == "mock" {
+	if provider == "mock" {
 		ai.Register("mock", newMock)
-	} 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)
+	} 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)
 	}
 
-	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 {
-		fmt.Println("broker connect:", err)
-		os.Exit(1)
+		return fmt.Errorf("broker connect: %w", err)
 	}
 	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.Registry(reg), service.Client(cl))
+		svc := service.New(service.Name(name), service.Address("127.0.0.1:0"), service.Registry(reg), service.Client(cl), service.HandleSignal(false))
 		_ = 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
@@ -250,10 +252,11 @@ func main() {
 	// 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"])
@@ -275,8 +278,7 @@ func main() {
 		flow.Prompt("A new support ticket arrived: {{.Data}}. Handle it."),
 	)
 	if err := intake.Register(reg, br, cl); err != nil {
-		fmt.Println("flow register:", err)
-		os.Exit(1)
+		return fmt.Errorf("flow register: %w", err)
 	}
 	defer intake.Stop()
 
@@ -287,8 +289,7 @@ func main() {
 	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 {
-		fmt.Println("publish:", err)
-		os.Exit(1)
+		return fmt.Errorf("publish: %w", err)
 	}
 
 	// Wait for the agent to act.
@@ -305,7 +306,18 @@ func main() {
 	}
 	if notify.sent >= 1 {
 		fmt.Println("\n\033[32m✓ ticket triaged and the customer was replied to — triggered by an event\033[0m")
-	} else {
-		fmt.Println("\n\033[31m✗ the agent did not complete the triage\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)
 	}
 }

examples/support/main_test.go

@@ -0,0 +1,9 @@
+package main
+
+import "testing"
+
+func TestRunSupportMockSmoke(t *testing.T) {
+	if err := runSupport("mock"); err != nil {
+		t.Fatalf("support example failed: %v", err)
+	}
+}

internal/docs/PRIORITIES.md

@@ -19,9 +19,11 @@ 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.
 
-## Later (ranked)
+## Developer experience (ranked)
 
-1. **Add human-in-the-loop pause and resume for agent runs** ([#3278](https://github.com/micro/go-micro/issues/3278)) — after memory summarization/retrieval shipped, the highest-value remaining roadmap gap is making long-running agents safer and more operable by letting a run enter an input-required pause, persist enough state to survive the boundary, and resume with human input using the existing agent/store/flow/A2A semantics rather than a separate orchestration layer.
+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.
 
 _Seeded by Claude Code from the roadmap + open issues; thereafter maintained by the
 architecture-review pass._