diff --git a/cli/azd/extensions/azure.ai.agents/internal/cmd/invoke.go b/cli/azd/extensions/azure.ai.agents/internal/cmd/invoke.go index 198649c0f10..5da590b27db 100644 --- a/cli/azd/extensions/azure.ai.agents/internal/cmd/invoke.go +++ b/cli/azd/extensions/azure.ai.agents/internal/cmd/invoke.go @@ -51,6 +51,7 @@ type invokeFlags struct { resumable bool noWait bool resume bool + steer bool cancel bool } @@ -129,9 +130,10 @@ suppressed in raw mode. Use --resumable with the Responses protocol to start work that continues running in the service if this command disconnects. The command remains attached until the work finishes. Add --no-wait to detach as soon as the service acknowledges the background -work. Use --resume to reconnect to saved work and --cancel to cancel it. In multi-agent -projects, pass the agent name positionally when resuming or cancelling saved work. Resumable -invocation is remote-only, does not support raw output, and cannot be combined with --timeout.`, +work. Use --resume to reconnect to saved work, --steer with input to revise active work +or start the next resumable turn after completion, and --cancel to cancel saved work. In +multi-agent projects, pass the agent name positionally. Resumable operations are remote-only, +do not support raw output, and cannot be combined with --timeout.`, Example: ` # Invoke the remote agent on Foundry (auto-detects agent from azure.yaml) azd ai agent invoke "Hello!" @@ -166,8 +168,9 @@ invocation is remote-only, does not support raw output, and cannot be combined w # Start resumable work and detach after the service acknowledges it azd ai agent invoke --resumable --no-wait "Run the long task" - # Resume or cancel saved resumable work + # Resume, steer, or cancel saved resumable work azd ai agent invoke --resume + azd ai agent invoke "Use the revised requirements" --steer azd ai agent invoke --cancel # Select an agent when reconnecting to or cancelling saved work @@ -272,7 +275,7 @@ invocation is remote-only, does not support raw output, and cannot be combined w ) } } - if flags.resumable || flags.resume || flags.cancel { + if flags.resumable || flags.resume || flags.steer || flags.cancel { if flags.local { return exterrors.Validation( exterrors.CodeInvalidParameter, @@ -345,7 +348,13 @@ invocation is remote-only, does not support raw output, and cannot be combined w "Start resumable work that continues in the service if the command disconnects; remain attached until it finishes", ) cmd.Flags().BoolVar(&flags.noWait, "no-wait", false, "Detach after the service acknowledges the resumable work") - cmd.Flags().BoolVar(&flags.resume, "resume", false, "Resume saved background work") + cmd.Flags().BoolVar(&flags.resume, "resume", false, "Reconnect to saved background work") + cmd.Flags().BoolVar( + &flags.steer, + "steer", + false, + "Revise active work or start the next resumable turn after completion", + ) cmd.Flags().BoolVar(&flags.cancel, "cancel", false, "Cancel the saved current background Response") // Register `raw` as an additional allowed value on the inherited global @@ -382,20 +391,21 @@ func parseInvokeArgs(flags *invokeFlags, args []string) { func validateInvokeOperationFlags(cmd *cobra.Command, flags *invokeFlags) error { hasInput := flags.message != "" || flags.inputFile != "" - continuesOrCancels := flags.resume || flags.cancel + messageFreeOperation := flags.resume || flags.cancel + savedResponseOperation := messageFreeOperation || flags.steer // An invocation selects exactly one operation. - if flags.resume && flags.cancel { + if (flags.resume && (flags.steer || flags.cancel)) || (flags.steer && flags.cancel) { return exterrors.Validation( exterrors.CodeInvalidParameter, - "--resume and --cancel are mutually exclusive", + "--resume, --steer, and --cancel are mutually exclusive", "choose one operation", ) } - if flags.resumable && continuesOrCancels { + if flags.resumable && savedResponseOperation { return exterrors.Validation( exterrors.CodeInvalidParameter, - "--resumable cannot be combined with --resume or --cancel", + "--resumable cannot be combined with --resume, --steer, or --cancel", "choose one operation", ) } @@ -417,14 +427,21 @@ func validateInvokeOperationFlags(cmd *cobra.Command, flags *invokeFlags) error "provide either a message argument or --input-file, not both", ) } - if !hasInput && !continuesOrCancels { + if flags.steer && !hasInput { + return exterrors.Validation( + exterrors.CodeInvalidParameter, + "--steer requires a message argument or --input-file", + "provide revised input to steer the saved Response", + ) + } + if !hasInput && !messageFreeOperation { return exterrors.Validation( exterrors.CodeInvalidParameter, "a message argument or --input-file is required", "provide a message as a positional argument, or use --input-file/-f to send a file", ) } - if continuesOrCancels && hasInput { + if messageFreeOperation && hasInput { return exterrors.Validation( exterrors.CodeInvalidParameter, "--resume and --cancel do not accept a message or --input-file", @@ -433,12 +450,12 @@ func validateInvokeOperationFlags(cmd *cobra.Command, flags *invokeFlags) error } // Operations on saved work own their session, conversation, and timeout. - if continuesOrCancels { + if savedResponseOperation { for _, name := range []string{"session-id", "new-session", "conversation-id", "new-conversation"} { if cmd.Flags().Changed(name) { return exterrors.Validation( exterrors.CodeInvalidParameter, - "--resume and --cancel use the saved session and conversation", + "--resume, --steer, and --cancel use the saved session and conversation", "remove session and conversation overrides", ) } @@ -446,7 +463,7 @@ func validateInvokeOperationFlags(cmd *cobra.Command, flags *invokeFlags) error if cmd.Flags().Changed("timeout") { return exterrors.Validation( exterrors.CodeConflictingArguments, - "--timeout is not supported with --resume or --cancel", + "--timeout is not supported with --resume, --steer, or --cancel", "remove --timeout; attached background work has no overall timeout", ) } @@ -560,7 +577,8 @@ func (a *InvokeAction) Run(ctx context.Context) error { // populated, but a2aRemote never calls applyCustomHeaders — the headers // would be silently dropped, which is the exact silent no-op the guard // intends to prevent. - if (a.flags.resumable || a.flags.resume || a.flags.cancel) && protocol != agent_api.AgentProtocolResponses { + if (a.flags.resumable || a.flags.resume || a.flags.steer || a.flags.cancel) && + protocol != agent_api.AgentProtocolResponses { return exterrors.Validation( exterrors.CodeInvalidParameter, fmt.Sprintf("resumable operations are not supported with the %s protocol", protocol), @@ -595,6 +613,9 @@ func (a *InvokeAction) Run(ctx context.Context) error { case agent_api.AgentProtocolA2A: return a.a2aRemote(ctx) default: + if a.flags.steer { + return a.responsesSteerRemote(ctx) + } if a.flags.resume { return a.responsesResumeRemote(ctx) } @@ -1287,8 +1308,8 @@ func (a *InvokeAction) ensureNoActiveBackgroundResponse( return nil } return fmt.Errorf( - "background Response %s is still active; reconnect with `azd ai agent invoke --resume` or cancel it with "+ - "`azd ai agent invoke --cancel`", + "background Response %s is still active; reconnect with `azd ai agent invoke --resume`, revise it with "+ + "`azd ai agent invoke \"\" --steer`, or cancel it with `azd ai agent invoke --cancel`", record.ResponseID, ) } diff --git a/cli/azd/extensions/azure.ai.agents/internal/cmd/invoke_background.go b/cli/azd/extensions/azure.ai.agents/internal/cmd/invoke_background.go index f2a1c874d3a..3326543e0ca 100644 --- a/cli/azd/extensions/azure.ai.agents/internal/cmd/invoke_background.go +++ b/cli/azd/extensions/azure.ai.agents/internal/cmd/invoke_background.go @@ -322,6 +322,8 @@ func classifyResponseLifecycleHTTPError(cause error, operation string) error { switch operation { case exterrors.OpResumeBackgroundResponse: operationLabel = "resuming background Response" + case exterrors.OpSteerBackgroundResponse: + operationLabel = "steering background Response" case exterrors.OpCancelBackgroundResponse: operationLabel = "cancelling background Response" } diff --git a/cli/azd/extensions/azure.ai.agents/internal/cmd/invoke_background_test.go b/cli/azd/extensions/azure.ai.agents/internal/cmd/invoke_background_test.go index b3225782ba5..0ea34fe8568 100644 --- a/cli/azd/extensions/azure.ai.agents/internal/cmd/invoke_background_test.go +++ b/cli/azd/extensions/azure.ai.agents/internal/cmd/invoke_background_test.go @@ -29,7 +29,7 @@ func TestInvokeCommandLifecycleFlagsRegistered(t *testing.T) { t.Parallel() flags := newInvokeCommand(nil).Flags() - for _, name := range []string{"resumable", "resume", "cancel"} { + for _, name := range []string{"resumable", "resume", "steer", "cancel"} { flag := flags.Lookup(name) require.NotNil(t, flag) assert.Equal(t, "false", flag.DefValue) @@ -838,6 +838,7 @@ func TestClassifyResponseLifecycleHTTPError(t *testing.T) { for _, operation := range []string{ exterrors.OpResumeBackgroundResponse, + exterrors.OpSteerBackgroundResponse, exterrors.OpCancelBackgroundResponse, } { t.Run(operation, func(t *testing.T) { @@ -947,17 +948,24 @@ func TestValidateInvokeOperationFlags(t *testing.T) { flags: invokeFlags{noWait: true, message: "hello"}, wantErr: "--no-wait requires --resumable", }, - {name: "continue accepts empty input", flags: invokeFlags{resume: true}}, + {name: "resume accepts empty input", flags: invokeFlags{resume: true}}, { - name: "continue rejects message", + name: "resume rejects message", flags: invokeFlags{resume: true, message: "hello"}, wantErr: "--resume and --cancel do not accept a message or --input-file", }, { - name: "continue rejects file", + name: "resume rejects file", flags: invokeFlags{resume: true, inputFile: "request.json"}, wantErr: "--resume and --cancel do not accept a message or --input-file", }, + {name: "steer accepts message", flags: invokeFlags{steer: true, message: "hello"}}, + {name: "steer accepts file", flags: invokeFlags{steer: true, inputFile: "request.json"}}, + { + name: "steer requires input", + flags: invokeFlags{steer: true}, + wantErr: "--steer requires a message argument or --input-file", + }, {name: "cancel accepts empty input", flags: invokeFlags{cancel: true}}, { name: "cancel rejects input", @@ -965,19 +973,29 @@ func TestValidateInvokeOperationFlags(t *testing.T) { wantErr: "--resume and --cancel do not accept a message or --input-file", }, { - name: "continue and cancel are exclusive", - flags: invokeFlags{resume: true, cancel: true}, - wantErr: "--resume and --cancel are mutually exclusive", + name: "resume and steer are exclusive", + flags: invokeFlags{resume: true, steer: true, message: "hello"}, + wantErr: "--resume, --steer, and --cancel are mutually exclusive", + }, + { + name: "steer and cancel are exclusive", + flags: invokeFlags{steer: true, cancel: true, message: "hello"}, + wantErr: "--resume, --steer, and --cancel are mutually exclusive", }, { - name: "background and continue are exclusive", + name: "background and resume are exclusive", flags: invokeFlags{resumable: true, resume: true, message: "hello"}, - wantErr: "--resumable cannot be combined with --resume or --cancel", + wantErr: "--resumable cannot be combined with --resume, --steer, or --cancel", + }, + { + name: "background and steer are exclusive", + flags: invokeFlags{resumable: true, steer: true, message: "hello"}, + wantErr: "--resumable cannot be combined with --resume, --steer, or --cancel", }, { name: "background and cancel are exclusive", flags: invokeFlags{resumable: true, cancel: true, message: "hello"}, - wantErr: "--resumable cannot be combined with --resume or --cancel", + wantErr: "--resumable cannot be combined with --resume, --steer, or --cancel", }, { name: "continue rejects session id", @@ -991,6 +1009,12 @@ func TestValidateInvokeOperationFlags(t *testing.T) { changed: map[string]string{"new-session": "true"}, wantErr: "use the saved session and conversation", }, + { + name: "steer rejects conversation id", + flags: invokeFlags{steer: true, message: "hello"}, + changed: map[string]string{"conversation-id": "conv_123"}, + wantErr: "use the saved session and conversation", + }, { name: "cancel rejects conversation id", flags: invokeFlags{cancel: true}, @@ -1004,16 +1028,22 @@ func TestValidateInvokeOperationFlags(t *testing.T) { wantErr: "use the saved session and conversation", }, { - name: "continue rejects timeout", + name: "resume rejects timeout", flags: invokeFlags{resume: true}, changed: map[string]string{"timeout": "1"}, - wantErr: "--timeout is not supported with --resume or --cancel", + wantErr: "--timeout is not supported with --resume, --steer, or --cancel", + }, + { + name: "steer rejects timeout", + flags: invokeFlags{steer: true, message: "hello"}, + changed: map[string]string{"timeout": "1"}, + wantErr: "--timeout is not supported with --resume, --steer, or --cancel", }, { name: "cancel rejects timeout", flags: invokeFlags{cancel: true}, changed: map[string]string{"timeout": "1"}, - wantErr: "--timeout is not supported with --resume or --cancel", + wantErr: "--timeout is not supported with --resume, --steer, or --cancel", }, } @@ -1058,6 +1088,12 @@ func TestParseInvokeArgs(t *testing.T) { args: []string{"agent"}, wantName: "agent", }, + { + name: "single positional with steer is message", + flags: invokeFlags{steer: true}, + args: []string{"revised requirements"}, + wantMessage: "revised requirements", + }, { name: "single positional with cancel is agent", flags: invokeFlags{cancel: true}, diff --git a/cli/azd/extensions/azure.ai.agents/internal/cmd/invoke_response_steer.go b/cli/azd/extensions/azure.ai.agents/internal/cmd/invoke_response_steer.go new file mode 100644 index 00000000000..8645f5b595f --- /dev/null +++ b/cli/azd/extensions/azure.ai.agents/internal/cmd/invoke_response_steer.go @@ -0,0 +1,139 @@ +// Copyright (c) Microsoft Corporation. All rights reserved. +// Licensed under the MIT License. + +package cmd + +import ( + "bytes" + "context" + "encoding/json" + "errors" + "fmt" + "io" + "net/http" + "os" + + "azureaiagent/internal/exterrors" +) + +func (a *InvokeAction) responsesSteerRemote(ctx context.Context) error { + body, bodyLabel, err := a.resolveBody() + if err != nil { + return err + } + rc, store, current, err := a.resolveSavedBackgroundResponse(ctx) + if err != nil { + return err + } + defer rc.azdClient.Close() + + rc.bearerToken, err = a.acquireBearerToken(ctx) + if err != nil { + return err + } + + requestBody, err := buildConversationContinuationRequest(string(body), current) + if err != nil { + return err + } + payload, err := json.Marshal(requestBody) + if err != nil { + return fmt.Errorf("marshal steering request: %w", err) + } + + responseURL := buildResponsesURL(rc.projectEndpoint, rc.name, rc.apiVersion) + req, err := http.NewRequestWithContext(ctx, http.MethodPost, responseURL, bytes.NewReader(payload)) + if err != nil { + return fmt.Errorf("create steering request: %w", err) + } + req.Header.Set("Authorization", "Bearer "+rc.bearerToken) + req.Header.Set("Content-Type", "application/json") + req.Header.Set("Accept", "text/event-stream") + applyCustomHeaders(req, a.clientHeaders) + applyRemoteUserIdentityHeader(req, &a.flags.userIdentityFlags) + + resp, err := backgroundHTTPClient().Do(req) //nolint:gosec // validated Foundry endpoint + if err != nil { + return fmt.Errorf("POST %s failed: %w", responseURL, err) + } + defer resp.Body.Close() + if resp.StatusCode >= 400 { + responseBody, _ := io.ReadAll(resp.Body) + return classifyResponseLifecycleHTTPError( + &responseLifecycleHTTPError{ + method: http.MethodPost, + requestURL: responseURL, + statusCode: resp.StatusCode, + status: resp.Status, + body: responseBody, + }, + exterrors.OpSteerBackgroundResponse, + ) + } + + effectiveSessionID := current.SessionID + if effectiveSessionID == "" { + effectiveSessionID = resp.Header.Get("x-agent-session-id") + } + captureResponseSession(ctx, rc.azdClient, rc.agentKey, current.SessionID, resp, "") + + fmt.Printf("Agent: %s (remote)\n", rc.name) + fmt.Printf("Message: %s\n", bodyLabel) + printSessionStatus("Session: ", effectiveSessionID) + fmt.Printf("Conversation: %s\n\n", current.ConversationID) + + progressPersister := newBackgroundProgressPersister( + store, + rc.agentKey, + effectiveSessionID, + current.ConversationID, + os.Stdout, + ) + streamErr := readResponsesSSE( + ctx, + resp.Body, + os.Stdout, + rc.name, + responsesSSEOptions{ + requireTerminal: true, + onProgress: func(progress responsesStreamProgress) error { + return progressPersister.Apply(ctx, progress) + }, + }, + ) + var flushErr error + if ctx.Err() == nil { + flushErr = progressPersister.Flush(ctx) + } + closeErr := progressPersister.Close() + if streamErr != nil && ctx.Err() == nil && progressPersister.latest.ResponseID != "" && + !isTerminalResponseStatus(progressPersister.latest.Status) && isRetryableBackgroundStreamError(streamErr) && + flushErr == nil && closeErr == nil { + latest := progressPersister.latest + return classifyResponseLifecycleHTTPError( + a.followBackgroundResponse(ctx, rc, store, latest, os.Stdout), + exterrors.OpSteerBackgroundResponse, + ) + } + return errors.Join(streamErr, flushErr, closeErr) +} + +func buildConversationContinuationRequest( + input string, + current *savedBackgroundResponse, +) (map[string]any, error) { + if current.ConversationID == "" { + return nil, fmt.Errorf("saved background Response has no conversation ID for --steer") + } + requestBody := map[string]any{ + "input": input, + "stream": true, + "store": true, + "background": true, + "conversation": map[string]string{"id": current.ConversationID}, + } + if current.SessionID != "" { + requestBody["agent_session_id"] = current.SessionID + } + return requestBody, nil +} diff --git a/cli/azd/extensions/azure.ai.agents/internal/cmd/invoke_response_steer_test.go b/cli/azd/extensions/azure.ai.agents/internal/cmd/invoke_response_steer_test.go new file mode 100644 index 00000000000..df3527f6c18 --- /dev/null +++ b/cli/azd/extensions/azure.ai.agents/internal/cmd/invoke_response_steer_test.go @@ -0,0 +1,61 @@ +// Copyright (c) Microsoft Corporation. All rights reserved. +// Licensed under the MIT License. + +package cmd + +import ( + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func TestBuildConversationContinuationRequestUsesSameShapeForAnyStatus(t *testing.T) { + t.Parallel() + + for _, status := range []string{"queued", "in_progress", "completed", "failed", "incomplete", "cancelled"} { + t.Run(status, func(t *testing.T) { + t.Parallel() + request, err := buildConversationContinuationRequest("revised input", &savedBackgroundResponse{ + ResponseID: "resp_123", + Status: status, + SessionID: "sess_123", + ConversationID: "conv_123", + }) + + require.NoError(t, err) + assert.Equal(t, "revised input", request["input"]) + assert.Equal(t, true, request["stream"]) + assert.Equal(t, true, request["store"]) + assert.Equal(t, true, request["background"]) + assert.Equal(t, "sess_123", request["agent_session_id"]) + assert.Equal(t, map[string]string{"id": "conv_123"}, request["conversation"]) + assert.NotContains(t, request, "previous_response_id") + }) + } +} + +func TestBuildConversationContinuationRequestOmitsEmptySession(t *testing.T) { + t.Parallel() + + request, err := buildConversationContinuationRequest("next", &savedBackgroundResponse{ + ResponseID: "resp_123", + ConversationID: "conv_123", + }) + + require.NoError(t, err) + assert.NotContains(t, request, "agent_session_id") + assert.Equal(t, map[string]string{"id": "conv_123"}, request["conversation"]) +} + +func TestBuildConversationContinuationRequestRequiresConversation(t *testing.T) { + t.Parallel() + + request, err := buildConversationContinuationRequest("next", &savedBackgroundResponse{ + ResponseID: "resp_123", + SessionID: "sess_123", + }) + + require.EqualError(t, err, "saved background Response has no conversation ID for --steer") + assert.Nil(t, request) +} diff --git a/cli/azd/extensions/azure.ai.agents/internal/exterrors/codes.go b/cli/azd/extensions/azure.ai.agents/internal/exterrors/codes.go index 244abed4e73..eccd65e520d 100644 --- a/cli/azd/extensions/azure.ai.agents/internal/exterrors/codes.go +++ b/cli/azd/extensions/azure.ai.agents/internal/exterrors/codes.go @@ -217,6 +217,7 @@ const ( OpPackTeamsApp = "pack_teams_app" OpPublishTeamsApp = "publish_teams_app" OpResumeBackgroundResponse = "resume_background_response" + OpSteerBackgroundResponse = "steer_background_response" OpCancelBackgroundResponse = "cancel_background_response" )