diff --git a/dotnet/samples/DurableWorkflows/AzureFunctions/01_SequentialWorkflow/README.md b/dotnet/samples/DurableWorkflows/AzureFunctions/01_SequentialWorkflow/README.md index 4ef1c6a..93ed14f 100644 --- a/dotnet/samples/DurableWorkflows/AzureFunctions/01_SequentialWorkflow/README.md +++ b/dotnet/samples/DurableWorkflows/AzureFunctions/01_SequentialWorkflow/README.md @@ -52,10 +52,15 @@ Invoke-RestMethod -Method Post ` The response will confirm the workflow orchestration has started: -```text -Workflow orchestration started for CancelOrder. Orchestration runId: abc123def456 +```json +{ + "runId": "abc123def456", + "message": "Workflow orchestration started for CancelOrder." +} ``` +Workflow run responses use JSON by default. Include `Accept: text/plain` to request the legacy plain-text representation instead. + > **Tip:** You can provide a custom run ID by appending a `runId` query parameter: > > ```bash @@ -68,14 +73,13 @@ Workflow orchestration started for CancelOrder. Orchestration runId: abc123def45 ### Wait for the Workflow Result -By default, the HTTP endpoint returns `202 Accepted` immediately with the run ID. If you want to wait for the workflow to complete and get the result in the response, add the `x-ms-wait-for-response: true` header: +By default, the HTTP endpoint returns `202 Accepted` immediately with the run ID. To wait for the workflow to complete, set `waitForResponse=true` in the query string. The endpoint waits for up to 10 seconds by default; set `timeoutSeconds` to use a timeout from 1 to 200 seconds. If the workflow is still running when the timeout expires, the endpoint returns the same `202 Accepted` response as the default asynchronous invocation. Bash (Linux/macOS/WSL): ```bash -curl -X POST http://localhost:7071/api/workflows/CancelOrder/run \ +curl -X POST "http://localhost:7071/api/workflows/CancelOrder/run?waitForResponse=true&timeoutSeconds=30" \ -H "Content-Type: text/plain" \ - -H "x-ms-wait-for-response: true" \ -d "12345" ``` @@ -83,36 +87,36 @@ PowerShell: ```powershell Invoke-RestMethod -Method Post ` - -Uri http://localhost:7071/api/workflows/CancelOrder/run ` + -Uri "http://localhost:7071/api/workflows/CancelOrder/run?waitForResponse=true&timeoutSeconds=30" ` -ContentType text/plain ` - -Headers @{ "x-ms-wait-for-response" = "true" } ` -Body "12345" ``` -The response will contain the workflow result as plain text (200 OK): +The response is JSON by default: -```text -Cancellation email sent for order 12345 to jerry@example.com. +```json +{ + "runId": "abc123def456", + "workflowStatus": "Completed", + "result": "Cancellation email sent for order 12345 to jerry@example.com." +} ``` -To get the result as JSON, also include the `Accept: application/json` header: +To get only the workflow result as plain text, include the `Accept: text/plain` header: ```bash -curl -X POST http://localhost:7071/api/workflows/CancelOrder/run \ +curl -X POST "http://localhost:7071/api/workflows/CancelOrder/run?waitForResponse=true" \ -H "Content-Type: text/plain" \ - -H "x-ms-wait-for-response: true" \ - -H "Accept: application/json" \ + -H "Accept: text/plain" \ -d "12345" ``` -```json -{ - "runId": "abc123def456", - "workflowStatus": "Completed", - "result": "Cancellation email sent for order 12345 to jerry@example.com." -} +```text +Cancellation email sent for order 12345 to jerry@example.com. ``` +The `x-ms-wait-for-response` header remains supported for backward compatibility. A wait timeout returns the same `202 Accepted` response as the default asynchronous invocation. Client request cancellation instead aborts the HTTP wait without returning a response, but the durable workflow continues; callers that need to recover should supply `runId` up front and query its status later. + In the function app logs, you will see the sequential execution of each executor: ```text diff --git a/dotnet/samples/DurableWorkflows/AzureFunctions/01_SequentialWorkflow/demo.http b/dotnet/samples/DurableWorkflows/AzureFunctions/01_SequentialWorkflow/demo.http index adbafad..95f52ea 100644 --- a/dotnet/samples/DurableWorkflows/AzureFunctions/01_SequentialWorkflow/demo.http +++ b/dotnet/samples/DurableWorkflows/AzureFunctions/01_SequentialWorkflow/demo.http @@ -7,18 +7,16 @@ Content-Type: text/plain 12345 -### Cancel an order and wait for the result -POST {{authority}}/api/workflows/CancelOrder/run +### Cancel an order and wait for the result (plain-text response) +POST {{authority}}/api/workflows/CancelOrder/run?waitForResponse=true&timeoutSeconds=30 Content-Type: text/plain -x-ms-wait-for-response: true +Accept: text/plain 12345 -### Cancel an order and wait for the result (JSON response) -POST {{authority}}/api/workflows/CancelOrder/run +### Cancel an order and wait for the result (default JSON response) +POST {{authority}}/api/workflows/CancelOrder/run?waitForResponse=true Content-Type: text/plain -Accept: application/json -x-ms-wait-for-response: true 12345 @@ -35,9 +33,8 @@ Content-Type: text/plain 12345 ### Get order status and wait for the result -POST {{authority}}/api/workflows/OrderStatus/run +POST {{authority}}/api/workflows/OrderStatus/run?waitForResponse=true Content-Type: text/plain -x-ms-wait-for-response: true 12345 diff --git a/dotnet/src/Microsoft.Agents.AI.Hosting.AzureFunctions/BuiltInFunctions.cs b/dotnet/src/Microsoft.Agents.AI.Hosting.AzureFunctions/BuiltInFunctions.cs index a96932d..f79e074 100644 --- a/dotnet/src/Microsoft.Agents.AI.Hosting.AzureFunctions/BuiltInFunctions.cs +++ b/dotnet/src/Microsoft.Agents.AI.Hosting.AzureFunctions/BuiltInFunctions.cs @@ -2,6 +2,7 @@ using System.Diagnostics.CodeAnalysis; using System.Net; +using System.Net.Http.Headers; using System.Text.Json; using System.Text.Json.Serialization; using Microsoft.Agents.AI.DurableTask; @@ -26,6 +27,28 @@ internal static class BuiltInFunctions private const string WaitForResponseHeaderName = "x-ms-wait-for-response"; + /// + /// Query parameter used by the agent endpoints, which use snake_case names throughout their + /// query string, request body, and response body. + /// + /// + /// Issue https://github.com/microsoft/agent-framework-durable-extension/issues/51 tracks the plan + /// to unify the query parameter naming across all endpoints. + /// + private const string AgentWaitForResponseParameterName = "wait_for_response"; + + /// + /// Query parameter used by the workflow endpoints, which use camelCase names throughout their + /// query string, request body, and response body. + /// + private const string WorkflowWaitForResponseParameterName = "waitForResponse"; + + private const string WorkflowRunIdParameterName = "runId"; + private const int MaxWorkflowRunIdLength = 100; + private const string WaitTimeoutSecondsParameterName = "timeoutSeconds"; + private const int DefaultWaitTimeoutSeconds = 10; + private const int MaxWaitTimeoutSeconds = 200; + private const string SessionIdHeaderName = "x-ms-session-id"; private const string SessionIdParameterName = "session_id"; private const string SessionIdMcpArgumentName = "sessionId"; @@ -73,24 +96,60 @@ public static async Task RunWorkflowOrchestrationHttpTriggerAs if (string.IsNullOrEmpty(inputMessage)) { - return await CreateErrorResponseAsync(req, context, HttpStatusCode.BadRequest, "Workflow input cannot be empty."); + return await CreateErrorResponseAsync( + req, + context, + HttpStatusCode.BadRequest, + "Workflow input cannot be empty.", + ShouldReturnWorkflowJson(req)); + } + + if (!TryGetWorkflowWaitForResponse(req, out bool waitForResponse, out string? waitForResponseError)) + { + return await CreateErrorResponseAsync( + req, + context, + HttpStatusCode.BadRequest, + waitForResponseError!, + ShouldReturnWorkflowJson(req)); + } + + TimeSpan waitTimeout = default; + if (waitForResponse && !TryGetWaitTimeout(req, out waitTimeout, out string? timeoutError)) + { + return await CreateErrorResponseAsync( + req, + context, + HttpStatusCode.BadRequest, + timeoutError!, + ShouldReturnWorkflowJson(req)); + } + + string? instanceId = req.Query[WorkflowRunIdParameterName]; + if (!TryValidateWorkflowRunId(instanceId, out string? runIdError)) + { + return await CreateErrorResponseAsync( + req, + context, + HttpStatusCode.BadRequest, + runIdError!, + ShouldReturnWorkflowJson(req)); } DurableWorkflowInput orchestrationInput = new() { Input = inputMessage }; // Allow users to provide a custom run ID via query string; otherwise, auto-generate one. - string? instanceId = req.Query["runId"]; StartOrchestrationOptions? options = instanceId is not null ? new StartOrchestrationOptions(instanceId) : null; string resolvedInstanceId = await client.ScheduleNewOrchestrationInstanceAsync(orchestrationFunctionName, orchestrationInput, options); - if (ShouldWaitForResponse(req, defaultValue: false)) + if (waitForResponse) { - return await WaitForWorkflowCompletionAsync(req, client, context, resolvedInstanceId); + return await WaitForWorkflowCompletionAsync( + req, client, context, workflowName, resolvedInstanceId, waitTimeout); } - HttpResponseData response = req.CreateResponse(HttpStatusCode.Accepted); - await response.WriteStringAsync($"Workflow orchestration started for {workflowName}. Orchestration runId: {resolvedInstanceId}"); - return response; + return await CreateWorkflowAcceptedResponseAsync( + req, workflowName, resolvedInstanceId, context.CancellationToken); } /// @@ -298,7 +357,7 @@ public static async Task RunAgentHttpAsync( message = await req.ReadAsStringAsync(); } - // The session ID can come from the query string or the JSON body, under either the canonical + // The session ID can come from the query string or the request body, using either the canonical // "session_id" name or the deprecated "thread_id" alias. Conflicting values are rejected. if (!TryResolveSessionKey( sessionIdFromBody, @@ -329,7 +388,7 @@ public static async Task RunAgentHttpAsync( } // Check if we should wait for response (default is true) - bool waitForResponse = ShouldWaitForResponse(req, defaultValue: true); + bool waitForResponse = ShouldWaitForResponse(req, AgentWaitForResponseParameterName, defaultValue: true); AIAgent agentProxy = client.AsDurableAgentProxy(context, agentName); @@ -462,23 +521,39 @@ await agentProxy.RunAsync( /// /// Waits for a workflow orchestration to complete and returns an appropriate HTTP response. /// - private static async Task WaitForWorkflowCompletionAsync( + internal static async Task WaitForWorkflowCompletionAsync( HttpRequestData req, DurableTaskClient client, FunctionContext context, - string instanceId) + string workflowName, + string instanceId, + TimeSpan timeout) { - bool acceptsJson = AcceptsJson(req); + bool returnJson = ShouldReturnWorkflowJson(req); - OrchestrationMetadata? metadata = await client.WaitForInstanceCompletionAsync( - instanceId, - getInputsAndOutputs: true, - cancellation: context.CancellationToken); + OrchestrationMetadata? metadata; + using (CancellationTokenSource timeoutSource = CancellationTokenSource.CreateLinkedTokenSource(context.CancellationToken)) + { + timeoutSource.CancelAfter(timeout); + + try + { + metadata = await client.WaitForInstanceCompletionAsync( + instanceId, + getInputsAndOutputs: true, + cancellation: timeoutSource.Token); + } + catch (OperationCanceledException) when (!context.CancellationToken.IsCancellationRequested) + { + return await CreateWorkflowAcceptedResponseAsync( + req, workflowName, instanceId, context.CancellationToken); + } + } if (metadata is null) { return await CreateErrorResponseAsync(req, context, HttpStatusCode.NotFound, - $"No workflow orchestration with ID '{instanceId}' was found.", acceptsJson); + $"No workflow orchestration with ID '{instanceId}' was found.", returnJson); } if (metadata.RuntimeStatus is OrchestrationRuntimeStatus.Failed) @@ -486,7 +561,7 @@ private static async Task WaitForWorkflowCompletionAsync( string errorMessage = metadata.FailureDetails?.ErrorMessage ?? "Unknown error"; HttpResponseData failedResponse = req.CreateResponse(HttpStatusCode.OK); - if (acceptsJson) + if (returnJson) { await failedResponse.WriteAsJsonAsync( new WorkflowRunResponse(instanceId, metadata.RuntimeStatus.ToString(), Result: null, Error: errorMessage), @@ -504,14 +579,14 @@ await failedResponse.WriteAsJsonAsync( if (metadata.RuntimeStatus is not OrchestrationRuntimeStatus.Completed) { return await CreateErrorResponseAsync(req, context, HttpStatusCode.InternalServerError, - $"Workflow orchestration '{instanceId}' ended with unexpected status '{metadata.RuntimeStatus}'.", acceptsJson); + $"Workflow orchestration '{instanceId}' ended with unexpected status '{metadata.RuntimeStatus}'.", returnJson); } string? result = metadata.ReadOutputAs()?.Result; HttpResponseData response = req.CreateResponse(HttpStatusCode.OK); - if (acceptsJson) + if (returnJson) { JsonElement? resultElement = null; if (!string.IsNullOrEmpty(result)) @@ -548,6 +623,35 @@ await response.WriteAsJsonAsync( return response; } + /// + /// Creates the response returned when a workflow continues asynchronously. + /// + internal static async Task CreateWorkflowAcceptedResponseAsync( + HttpRequestData req, + string workflowName, + string instanceId, + CancellationToken cancellationToken) + { + HttpResponseData response = req.CreateResponse(HttpStatusCode.Accepted); + if (ShouldReturnWorkflowJson(req)) + { + await response.WriteAsJsonAsync( + new WorkflowAcceptedResponse( + instanceId, + $"Workflow orchestration started for {workflowName}."), + cancellationToken); + } + else + { + response.Headers.Add("Content-Type", "text/plain"); + await response.WriteStringAsync( + $"Workflow orchestration started for {workflowName}. Orchestration runId: {instanceId}", + cancellationToken); + } + + return response; + } + /// /// Creates an error response with the specified status code and error message. /// @@ -644,13 +748,18 @@ private static async Task CreateAcceptedResponseAsync( /// /// Returns when the caller has requested waiting for the workflow/agent to complete, - /// as indicated by the x-ms-wait-for-response header. Falls back to - /// when the header is absent or not a valid boolean. + /// as indicated by the x-ms-wait-for-response header or query parameter. + /// The header takes precedence. Falls back to when neither value is valid. /// - private static bool ShouldWaitForResponse(HttpRequestData req, bool defaultValue) + internal static bool ShouldWaitForResponse(HttpRequestData req, string parameterName, bool defaultValue) { if (req.Headers.TryGetValues(WaitForResponseHeaderName, out IEnumerable? values) && - bool.TryParse(values.FirstOrDefault(), out bool parsed)) + TryParseBoolean(values.FirstOrDefault(), out bool parsed)) + { + return parsed; + } + + if (TryParseBoolean(req.Query[parameterName], out parsed)) { return parsed; } @@ -658,6 +767,114 @@ private static bool ShouldWaitForResponse(HttpRequestData req, bool defaultValue return defaultValue; } + /// + /// Gets the workflow wait preference while rejecting invalid values of the new query parameter. + /// The legacy header retains precedence and its existing invalid-value fallback behavior. + /// + internal static bool TryGetWorkflowWaitForResponse( + HttpRequestData req, + out bool waitForResponse, + out string? error) + { + if (req.Headers.TryGetValues(WaitForResponseHeaderName, out IEnumerable? values) && + TryParseBoolean(values.FirstOrDefault(), out waitForResponse)) + { + error = null; + return true; + } + + string? queryValue = req.Query[WorkflowWaitForResponseParameterName]; + if (queryValue is null) + { + waitForResponse = false; + error = null; + return true; + } + + if (!TryParseBoolean(queryValue, out waitForResponse)) + { + error = $"'{WorkflowWaitForResponseParameterName}' must be a boolean value."; + return false; + } + + error = null; + return true; + } + + private static bool TryParseBoolean(string? value, out bool result) + { + switch (value?.Trim().ToUpperInvariant()) + { + case "TRUE": + case "1": + case "YES": + case "Y": + case "ON": + result = true; + return true; + case "FALSE": + case "0": + case "NO": + case "N": + case "OFF": + result = false; + return true; + default: + result = false; + return false; + } + } + + /// + /// Gets the maximum time to wait for a synchronous workflow response. + /// + internal static bool TryGetWaitTimeout(HttpRequestData req, out TimeSpan timeout, out string? error) + { + string? value = req.Query[WaitTimeoutSecondsParameterName]; + if (value is null) + { + timeout = TimeSpan.FromSeconds(DefaultWaitTimeoutSeconds); + error = null; + return true; + } + + if (!int.TryParse(value, out int seconds) || seconds <= 0 || seconds > MaxWaitTimeoutSeconds) + { + timeout = default; + error = $"'{WaitTimeoutSecondsParameterName}' must be an integer between 1 and {MaxWaitTimeoutSeconds}."; + return false; + } + + timeout = TimeSpan.FromSeconds(seconds); + error = null; + return true; + } + + /// + /// Validates a caller-provided workflow run ID against the Durable Task instance ID contract. + /// + internal static bool TryValidateWorkflowRunId(string? runId, out string? error) + { + if (runId is null) + { + error = null; + return true; + } + + if (runId.Length is < 1 or > MaxWorkflowRunIdLength || + runId[0] == '@' || + runId.IndexOfAny(['/', '\\', '#', '?']) >= 0 || + runId.Any(char.IsControl)) + { + error = $"'{WorkflowRunIdParameterName}' must be between 1 and {MaxWorkflowRunIdLength} characters, " + + "must not start with '@', and must not contain '/', '\\', '#', '?', or control characters."; + return false; + } + + error = null; + return true; + } + /// /// Returns when the request accepts the application/json media type. /// @@ -670,6 +887,36 @@ private static bool AcceptsJson(HttpRequestData req) .Contains("application/json", StringComparer.OrdinalIgnoreCase); } + /// + /// Returns only when the caller explicitly requests a text response + /// without also accepting JSON. Workflow endpoints otherwise default to JSON. + /// + internal static bool ShouldReturnWorkflowJson(HttpRequestData req) + { + if (!req.Headers.TryGetValues("Accept", out IEnumerable? acceptValues)) + { + return true; + } + + string[] mediaTypes = acceptValues + .SelectMany(v => v.Split(',', StringSplitOptions.RemoveEmptyEntries | StringSplitOptions.TrimEntries)) + .Select(v => MediaTypeWithQualityHeaderValue.TryParse(v, out MediaTypeWithQualityHeaderValue? mediaType) + ? mediaType + : null) + .OfType() + .Where(v => (v.Quality ?? 1) > 0) + .Select(v => v.MediaType!) + .ToArray(); + + bool acceptsJson = mediaTypes.Contains("application/json", StringComparer.OrdinalIgnoreCase) || + mediaTypes.Contains("application/*", StringComparer.OrdinalIgnoreCase) || + mediaTypes.Contains("*/*", StringComparer.OrdinalIgnoreCase); + bool acceptsText = mediaTypes.Contains("text/plain", StringComparer.OrdinalIgnoreCase) || + mediaTypes.Contains("text/*", StringComparer.OrdinalIgnoreCase); + + return acceptsJson || !acceptsText; + } + /// /// Resolves the session key for an agent run request from the four places a caller may supply it: /// the canonical session_id and its deprecated thread_id alias, in both the request body @@ -851,6 +1098,15 @@ private sealed record WorkflowRunResponse( [property: JsonPropertyName("result")] JsonElement? Result, [property: JsonPropertyName("error"), JsonIgnore(Condition = JsonIgnoreCondition.WhenWritingNull)] string? Error = null); + /// + /// Represents a workflow run that continues asynchronously. + /// + /// The orchestration run ID. + /// A human-readable description of the accepted workflow run. + private sealed record WorkflowAcceptedResponse( + [property: JsonPropertyName("runId")] string RunId, + [property: JsonPropertyName("message")] string Message); + /// /// A service provider that combines the original service provider with an additional DurableTaskClient instance. /// diff --git a/dotnet/src/Microsoft.Agents.AI.Hosting.AzureFunctions/CHANGELOG.md b/dotnet/src/Microsoft.Agents.AI.Hosting.AzureFunctions/CHANGELOG.md index 95e7ce0..7f27f09 100644 --- a/dotnet/src/Microsoft.Agents.AI.Hosting.AzureFunctions/CHANGELOG.md +++ b/dotnet/src/Microsoft.Agents.AI.Hosting.AzureFunctions/CHANGELOG.md @@ -2,6 +2,7 @@ ## [Unreleased] +- [BREAKING] Support bounded synchronous workflow HTTP invocation through query parameters and default workflow run responses to JSON, with `Accept: text/plain` available for the legacy text format and the same negotiated asynchronous response returned on timeout ([#52](https://github.com/microsoft/agent-framework-durable-extension/pull/52)) - Added `DurableTaskClient.AsWorkflowClient` so functions can invoke durable workflows without constructing an `HttpClient` ([#48](https://github.com/microsoft/agent-framework-durable-extension/pull/48)) - [BREAKING] Consolidated the `AddWorkflow` extension overloads into a single method with optional `enableStatusEndpoint` and `enableMcpToolTrigger` parameters, and changed it to return `DurableWorkflowOptions` instead of `void` so multiple workflows can be registered fluently ([#39](https://github.com/microsoft/agent-framework-durable-extension/pull/39)) - [BREAKING] Replace "thread" with "session" in HTTP and MCP APIs ([#47](https://github.com/microsoft/agent-framework-durable-extension/pull/47)) diff --git a/dotnet/tests/Microsoft.Agents.AI.Hosting.AzureFunctions.UnitTests/BuiltInFunctionsWorkflowRoutingTests.cs b/dotnet/tests/Microsoft.Agents.AI.Hosting.AzureFunctions.UnitTests/BuiltInFunctionsWorkflowRoutingTests.cs index 5c671d3..8f28849 100644 --- a/dotnet/tests/Microsoft.Agents.AI.Hosting.AzureFunctions.UnitTests/BuiltInFunctionsWorkflowRoutingTests.cs +++ b/dotnet/tests/Microsoft.Agents.AI.Hosting.AzureFunctions.UnitTests/BuiltInFunctionsWorkflowRoutingTests.cs @@ -1,5 +1,15 @@ // Copyright (c) Microsoft. All rights reserved. +using System.Collections.Specialized; +using System.Net; +using System.Text.Json; +using Azure.Core.Serialization; +using Microsoft.Azure.Functions.Worker; +using Microsoft.Azure.Functions.Worker.Http; +using Microsoft.DurableTask.Client; +using Microsoft.Extensions.Options; +using Moq; + namespace Microsoft.Agents.AI.Hosting.AzureFunctions.UnitTests; public sealed class BuiltInFunctionsWorkflowRoutingTests @@ -49,4 +59,429 @@ public void IsOrchestrationOwnedByWorkflow_ValidatesCorrectly( // Assert Assert.Equal(expectedResult, result); } + + [Theory] + [InlineData(null, null, false, false)] + [InlineData(null, "true", false, true)] + [InlineData("false", "true", true, false)] + [InlineData("invalid", "true", false, true)] + [InlineData("1", "false", false, true)] + [InlineData("0", "true", true, false)] + [InlineData(null, "yes", false, true)] + [InlineData(null, "off", true, false)] + public void ShouldWaitForResponse_UsesHeaderThenQueryThenDefault( + string? headerValue, + string? queryValue, + bool defaultValue, + bool expected) + { + HttpRequestData request = CreateRequest(headerValue, queryValue); + + bool result = BuiltInFunctions.ShouldWaitForResponse(request, "waitForResponse", defaultValue); + + Assert.Equal(expected, result); + } + + [Theory] + [InlineData("waitForResponse", "wait_for_response")] // workflow endpoints use camelCase + [InlineData("wait_for_response", "waitForResponse")] // agent endpoints use snake_case + public void ShouldWaitForResponse_OnlyHonorsTheParameterForItsSurface(string parameterName, string otherName) + { + HttpRequestData matching = CreateRequest(waitForResponse: "true", waitForResponseParameterName: parameterName); + HttpRequestData mismatched = CreateRequest(waitForResponse: "true", waitForResponseParameterName: otherName); + + Assert.True(BuiltInFunctions.ShouldWaitForResponse(matching, parameterName, defaultValue: false)); + Assert.False(BuiltInFunctions.ShouldWaitForResponse(mismatched, parameterName, defaultValue: false)); + } + + [Theory] + [InlineData(null, null, true, false)] + [InlineData("invalid", null, true, false)] + [InlineData("invalid", "true", true, true)] + [InlineData("false", "invalid", true, false)] + [InlineData(null, "invalid", false, false)] + public void TryGetWorkflowWaitForResponse_ValidatesQueryAfterHeader( + string? headerValue, + string? queryValue, + bool expectedSuccess, + bool expectedWait) + { + HttpRequestData request = CreateRequest(headerValue, queryValue); + + bool success = BuiltInFunctions.TryGetWorkflowWaitForResponse( + request, + out bool waitForResponse, + out string? error); + + Assert.Equal(expectedSuccess, success); + Assert.Equal(expectedWait, waitForResponse); + Assert.Equal(expectedSuccess, error is null); + } + + [Theory] + [InlineData(null, true, 10)] + [InlineData("1", true, 1)] + [InlineData("200", true, 200)] + [InlineData("0", false, 0)] + [InlineData("201", false, 0)] + [InlineData("invalid", false, 0)] + [InlineData("", false, 0)] + [InlineData(" ", false, 0)] + public void TryGetWaitTimeout_ValidatesSeconds(string? value, bool expectedSuccess, int expectedSeconds) + { + HttpRequestData request = CreateRequest(timeoutSeconds: value); + + bool success = BuiltInFunctions.TryGetWaitTimeout(request, out TimeSpan timeout, out string? error); + + Assert.Equal(expectedSuccess, success); + Assert.Equal(expectedSeconds, timeout.TotalSeconds); + Assert.Equal(expectedSuccess, error is null); + } + + [Theory] + [InlineData(null, true)] + [InlineData("workflow-123", true)] + [InlineData("xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx", true)] + [InlineData("", false)] + [InlineData("@workflow", false)] + [InlineData("workflow/123", false)] + [InlineData("workflow\\123", false)] + [InlineData("workflow#123", false)] + [InlineData("workflow?123", false)] + [InlineData("workflow\n123", false)] + [InlineData("xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx", false)] + public void TryValidateWorkflowRunId_EnforcesDurableTaskContract(string? runId, bool expected) + { + bool success = BuiltInFunctions.TryValidateWorkflowRunId(runId, out string? error); + + Assert.Equal(expected, success); + Assert.Equal(expected, error is null); + } + + [Theory] + [InlineData(null, true)] + [InlineData("application/json", true)] + [InlineData("application/*", true)] + [InlineData("*/*", true)] + [InlineData("text/plain", false)] + [InlineData("text/*", false)] + [InlineData("text/plain, application/json", true)] + [InlineData("text/plain, application/json;q=0", false)] + [InlineData("text/plain;q=0", true)] + [InlineData("image/png", true)] + [InlineData("not-a-media-type", true)] + [InlineData("not-a-media-type, text/plain", false)] + public void ShouldReturnWorkflowJson_DefaultsToJsonUnlessOnlyTextIsAccepted( + string? accept, + bool expected) + { + (HttpRequestData request, _, _) = CreateCompletionRequest(CancellationToken.None, accept); + + Assert.Equal(expected, BuiltInFunctions.ShouldReturnWorkflowJson(request)); + } + + [Theory] + [InlineData(null)] + [InlineData("text/plain")] + public async Task WaitForWorkflowCompletionAsync_Timeout_ReturnsAsyncWorkflowResponseAsync(string? accept) + { + const string WorkflowName = "TestWorkflow"; + const string InstanceId = "workflow-123"; + (HttpRequestData asyncRequest, _, FunctionContext asyncContext) = + CreateCompletionRequest(CancellationToken.None, accept); + HttpResponseData expectedResponse = await BuiltInFunctions.CreateWorkflowAcceptedResponseAsync( + asyncRequest, WorkflowName, InstanceId, asyncContext.CancellationToken); + (HttpRequestData request, _, FunctionContext context) = + CreateCompletionRequest(CancellationToken.None, accept); + Mock client = CreateWaitingClient(InstanceId); + + HttpResponseData response = await BuiltInFunctions.WaitForWorkflowCompletionAsync( + request, + client.Object, + context, + WorkflowName, + InstanceId, + TimeSpan.Zero); + + Assert.Equal(expectedResponse.StatusCode, response.StatusCode); + Assert.Equal(GetResponseBody(expectedResponse), GetResponseBody(response)); + Assert.Equal( + expectedResponse.Headers.Select(header => (header.Key, string.Join(",", header.Value))), + response.Headers.Select(header => (header.Key, string.Join(",", header.Value)))); + } + + [Theory] + [InlineData(null)] + [InlineData("text/plain")] + public async Task CreateWorkflowAcceptedResponseAsync_HonorsContentNegotiationAsync(string? accept) + { + const string WorkflowName = "TestWorkflow"; + const string InstanceId = "workflow-123"; + (HttpRequestData request, _, FunctionContext context) = + CreateCompletionRequest(CancellationToken.None, accept); + + HttpResponseData response = await BuiltInFunctions.CreateWorkflowAcceptedResponseAsync( + request, WorkflowName, InstanceId, context.CancellationToken); + + Assert.Equal(HttpStatusCode.Accepted, response.StatusCode); + if (accept is null) + { + AssertJsonContentType(response); + using JsonDocument body = JsonDocument.Parse(GetResponseBody(response)); + Assert.Equal(InstanceId, body.RootElement.GetProperty("runId").GetString()); + Assert.Equal( + $"Workflow orchestration started for {WorkflowName}.", + body.RootElement.GetProperty("message").GetString()); + } + else + { + Assert.Equal( + $"Workflow orchestration started for {WorkflowName}. Orchestration runId: {InstanceId}", + GetResponseBody(response)); + AssertTextContentType(response); + } + } + + [Theory] + [InlineData(null)] + [InlineData("text/plain")] + public async Task WaitForWorkflowCompletionAsync_Completed_HonorsContentNegotiationAsync(string? accept) + { + const string InstanceId = "workflow-123"; + const string Result = "Workflow completed."; + (HttpRequestData request, _, FunctionContext context) = + CreateCompletionRequest(CancellationToken.None, accept); + OrchestrationMetadata metadata = new("dafx-TestWorkflow", InstanceId) + { + RuntimeStatus = OrchestrationRuntimeStatus.Completed, + DataConverter = Microsoft.DurableTask.Converters.JsonDataConverter.Default, + SerializedOutput = JsonSerializer.Serialize(new { Result }), + }; + Mock client = CreateCompletedClient(InstanceId, metadata); + + HttpResponseData response = await BuiltInFunctions.WaitForWorkflowCompletionAsync( + request, + client.Object, + context, + "TestWorkflow", + InstanceId, + TimeSpan.FromSeconds(1)); + + Assert.Equal(HttpStatusCode.OK, response.StatusCode); + if (accept is null) + { + AssertJsonContentType(response); + using JsonDocument body = JsonDocument.Parse(GetResponseBody(response)); + Assert.Equal(InstanceId, body.RootElement.GetProperty("runId").GetString()); + Assert.Equal("Completed", body.RootElement.GetProperty("workflowStatus").GetString()); + Assert.Equal(Result, body.RootElement.GetProperty("result").GetString()); + } + else + { + Assert.Equal(Result, GetResponseBody(response)); + AssertTextContentType(response); + } + } + + [Theory] + [InlineData(null)] + [InlineData("text/plain")] + public async Task WaitForWorkflowCompletionAsync_Failed_HonorsContentNegotiationAsync(string? accept) + { + const string InstanceId = "workflow-123"; + (HttpRequestData request, _, FunctionContext context) = + CreateCompletionRequest(CancellationToken.None, accept); + OrchestrationMetadata metadata = new("dafx-TestWorkflow", InstanceId) + { + RuntimeStatus = OrchestrationRuntimeStatus.Failed, + }; + Mock client = CreateCompletedClient(InstanceId, metadata); + + HttpResponseData response = await BuiltInFunctions.WaitForWorkflowCompletionAsync( + request, + client.Object, + context, + "TestWorkflow", + InstanceId, + TimeSpan.FromSeconds(1)); + + Assert.Equal(HttpStatusCode.OK, response.StatusCode); + if (accept is null) + { + AssertJsonContentType(response); + using JsonDocument body = JsonDocument.Parse(GetResponseBody(response)); + Assert.Equal(InstanceId, body.RootElement.GetProperty("runId").GetString()); + Assert.Equal("Failed", body.RootElement.GetProperty("workflowStatus").GetString()); + Assert.Equal("Unknown error", body.RootElement.GetProperty("error").GetString()); + } + else + { + Assert.Equal("Unknown error", GetResponseBody(response)); + AssertTextContentType(response); + } + } + + [Theory] + [InlineData(null)] + [InlineData("text/plain")] + public async Task RunWorkflowOrchestrationHttpTriggerAsync_ValidationError_HonorsContentNegotiationAsync( + string? accept) + { + (HttpRequestData request, _, FunctionContext context) = + CreateCompletionRequest(CancellationToken.None, accept); + Mock.Get(request).SetupGet(r => r.Body).Returns(new MemoryStream()); + Mock functionDefinition = new(); + functionDefinition.SetupGet(d => d.Name).Returns("http-TestWorkflow"); + Mock.Get(context).SetupGet(c => c.FunctionDefinition).Returns(functionDefinition.Object); + Mock client = new("test"); + + HttpResponseData response = await BuiltInFunctions.RunWorkflowOrchestrationHttpTriggerAsync( + request, client.Object, context); + + Assert.Equal(HttpStatusCode.BadRequest, response.StatusCode); + if (accept is null) + { + AssertJsonContentType(response); + using JsonDocument body = JsonDocument.Parse(GetResponseBody(response)); + Assert.Equal(400, body.RootElement.GetProperty("status").GetInt32()); + Assert.Equal("Workflow input cannot be empty.", body.RootElement.GetProperty("error").GetString()); + } + else + { + Assert.Equal("Workflow input cannot be empty.", GetResponseBody(response)); + AssertTextContentType(response); + } + } + + [Fact] + public async Task WaitForWorkflowCompletionAsync_CallerCancellation_PropagatesAsync() + { + const string WorkflowName = "TestWorkflow"; + const string InstanceId = "workflow-123"; + using CancellationTokenSource callerCancellation = new(); + callerCancellation.Cancel(); + HttpRequestData request = CreateRequest(); + Mock context = new(); + context.SetupGet(c => c.CancellationToken).Returns(callerCancellation.Token); + Mock client = CreateWaitingClient(InstanceId); + + await Assert.ThrowsAnyAsync(() => + BuiltInFunctions.WaitForWorkflowCompletionAsync( + request, + client.Object, + context.Object, + WorkflowName, + InstanceId, + TimeSpan.FromSeconds(30))); + } + + private static HttpRequestData CreateRequest( + string? headerValue = null, + string? waitForResponse = null, + string? timeoutSeconds = null, + string waitForResponseParameterName = "waitForResponse") + { + HttpHeadersCollection headers = new(); + if (headerValue is not null) + { + headers.Add("x-ms-wait-for-response", headerValue); + } + + NameValueCollection query = new(); + if (waitForResponse is not null) + { + query.Add(waitForResponseParameterName, waitForResponse); + } + + if (timeoutSeconds is not null) + { + query.Add("timeoutSeconds", timeoutSeconds); + } + + Mock request = new(MockBehavior.Strict, Mock.Of()); + request.SetupGet(r => r.Headers).Returns(headers); + request.SetupGet(r => r.Query).Returns(query); + return request.Object; + } + + private static Mock CreateWaitingClient(string instanceId) + { + Mock client = new("test"); + client.Setup(c => c.WaitForInstanceCompletionAsync(instanceId, true, It.IsAny())) + .Returns(async (string _, bool _, CancellationToken cancellation) => + { + await Task.Delay(Timeout.InfiniteTimeSpan, cancellation); + throw new InvalidOperationException("The infinite wait completed without cancellation."); + }); + return client; + } + + private static Mock CreateCompletedClient( + string instanceId, + OrchestrationMetadata metadata) + { + Mock client = new("test"); + client.Setup(c => c.WaitForInstanceCompletionAsync(instanceId, true, It.IsAny())) + .ReturnsAsync(metadata); + return client; + } + + private static (HttpRequestData Request, HttpResponseData Response, FunctionContext Context) CreateCompletionRequest( + CancellationToken cancellationToken, + string? accept = null) + { + Mock serializer = new(); + serializer + .Setup(s => s.SerializeAsync( + It.IsAny(), + It.IsAny(), + It.IsAny(), + It.IsAny())) + .Returns(async (stream, value, type, token) => + await JsonSerializer.SerializeAsync(stream, value, type, cancellationToken: token)); + + WorkerOptions workerOptions = new() { Serializer = serializer.Object }; + Mock> options = new(); + options.SetupGet(o => o.Value).Returns(workerOptions); + Mock services = new(); + services.Setup(s => s.GetService(typeof(IOptions))).Returns(options.Object); + + Mock context = new(); + context.SetupGet(c => c.CancellationToken).Returns(cancellationToken); + context.SetupGet(c => c.InstanceServices).Returns(services.Object); + + Mock response = new(context.Object); + response.SetupProperty(r => r.StatusCode, HttpStatusCode.OK); + response.SetupProperty(r => r.Body, new MemoryStream()); + response.SetupGet(r => r.Headers).Returns(new HttpHeadersCollection()); + + HttpHeadersCollection requestHeaders = new(); + if (accept is not null) + { + Assert.True(requestHeaders.TryAddWithoutValidation("Accept", accept)); + } + + Mock request = new(context.Object); + request.SetupGet(r => r.Headers).Returns(requestHeaders); + request.Setup(r => r.CreateResponse()).Returns(response.Object); + + return (request.Object, response.Object, context.Object); + } + + private static string GetResponseBody(HttpResponseData response) + { + return System.Text.Encoding.UTF8.GetString(((MemoryStream)response.Body).ToArray()); + } + + private static void AssertJsonContentType(HttpResponseData response) + { + Assert.True(response.Headers.TryGetValues("Content-Type", out IEnumerable? contentTypes)); + Assert.Contains("application/json", Assert.Single(contentTypes), StringComparison.OrdinalIgnoreCase); + } + + private static void AssertTextContentType(HttpResponseData response) + { + Assert.True(response.Headers.TryGetValues("Content-Type", out IEnumerable? contentTypes)); + Assert.Contains("text/plain", Assert.Single(contentTypes), StringComparison.OrdinalIgnoreCase); + } } diff --git a/python/packages/azurefunctions/agent_framework_azurefunctions/_app.py b/python/packages/azurefunctions/agent_framework_azurefunctions/_app.py index 3ea9137..dc90d13 100644 --- a/python/packages/azurefunctions/agent_framework_azurefunctions/_app.py +++ b/python/packages/azurefunctions/agent_framework_azurefunctions/_app.py @@ -12,11 +12,12 @@ import json import logging import re +import unicodedata import uuid from collections.abc import Callable, Mapping from dataclasses import asdict, dataclass, is_dataclass from datetime import datetime, timezone -from typing import TYPE_CHECKING, Any, TypeVar, cast +from typing import TYPE_CHECKING, Any, Protocol, TypeVar, cast import azure.durable_functions as df import azure.functions as func @@ -61,12 +62,33 @@ from ._routes import build_workflow_respond_url, build_workflow_status_url, split_request_url from ._workflow import run_workflow_orchestrator +_DEFAULT_WORKFLOW_WAIT_TIMEOUT_SECONDS = 10 +_MAX_WORKFLOW_WAIT_TIMEOUT_SECONDS = 200 + +# The workflow endpoints use camelCase names throughout their query string, request body, and +# response body. The agent endpoints use snake_case throughout, so they reuse the shared +# ``SESSION_ID_FIELD`` / ``WAIT_FOR_RESPONSE_FIELD`` constants instead of the names below. +_RUN_ID_QUERY_PARAMETER = "runId" +_MAX_WORKFLOW_RUN_ID_LENGTH = 100 +_WORKFLOW_WAIT_FOR_RESPONSE_QUERY_PARAMETER = "waitForResponse" +_WORKFLOW_WAIT_TIMEOUT_SECONDS_QUERY_PARAMETER = "timeoutSeconds" + logger = logging.getLogger("agent_framework.azurefunctions") EntityHandler = Callable[[df.DurableEntityContext], None] HandlerT = TypeVar("HandlerT", bound=Callable[..., Any]) +class _WorkflowCompletionClient(Protocol): + async def wait_for_completion_or_create_check_status_response( + self, + request: func.HttpRequest, + instance_id: str, + timeout_in_milliseconds: int = 10_000, + retry_interval_in_milliseconds: int = 1_000, + ) -> func.HttpResponse: ... + + def _json_default(obj: Any) -> Any: """JSON fallback encoder for reconstructed workflow outputs. @@ -490,6 +512,19 @@ async def start_workflow_orchestration( req: func.HttpRequest, client: df.DurableOrchestrationClient ) -> func.HttpResponse: """HTTP endpoint to start the workflow.""" + try: + wait_for_response = self._get_workflow_wait_for_response(req) + wait_timeout_seconds = _DEFAULT_WORKFLOW_WAIT_TIMEOUT_SECONDS + if wait_for_response: + wait_timeout_seconds = self._get_workflow_wait_timeout_seconds(req) + except ValueError as exc: + return self._build_error_response(str(exc), status_code=400) + + try: + requested_instance_id = self._validate_workflow_run_id((req.params or {}).get(_RUN_ID_QUERY_PARAMETER)) + except ValueError as exc: + return self._build_error_response(str(exc), status_code=400) + try: client_input: Any = req.get_json() except ValueError: @@ -507,23 +542,28 @@ async def start_workflow_orchestration( client_input = strip_subworkflow_markers(client_input) client_input = strip_pickle_markers(client_input) - instance_id = await client.start_new(orchestrator_name, client_input=client_input) + instance_id = await client.start_new( + orchestrator_name, + instance_id=requested_instance_id, + client_input=client_input, + ) - base_url, route_prefix = split_request_url(req.url) - status_url = build_workflow_status_url(base_url, workflow_name, instance_id, prefix=route_prefix) + if wait_for_response: + timeout_in_milliseconds = wait_timeout_seconds * 1000 + # The SDK leaves the request parameter untyped, so use the local protocol + # to give pyright a complete signature for this runtime method. + completion_client = cast(_WorkflowCompletionClient, client) + completion_response = await completion_client.wait_for_completion_or_create_check_status_response( + req, + instance_id, + timeout_in_milliseconds=timeout_in_milliseconds, + retry_interval_in_milliseconds=1000, + ) + if completion_response.status_code != 202: + status = await client.get_status(instance_id) + return self._build_workflow_terminal_response(status, instance_id) - return func.HttpResponse( - json.dumps({ - "instanceId": instance_id, - "statusQueryGetUri": status_url, - "respondUri": build_workflow_respond_url( - base_url, workflow_name, instance_id, "{requestId}", prefix=route_prefix - ), - "message": "Workflow started", - }), - status_code=202, - mimetype="application/json", - ) + return self._build_workflow_accepted_response(req, workflow_name, instance_id) @self.function_name(f"{orchestrator_name}-status") @self.route(route=f"workflow/{workflow_name}/status/{{instanceId}}", methods=["GET"]) @@ -1455,6 +1495,60 @@ def _build_accepted_response(self, message: str, session_id: str, correlation_id correlation_id=correlation_id, ) + @staticmethod + def _build_workflow_accepted_response( + req: func.HttpRequest, workflow_name: str, instance_id: str + ) -> func.HttpResponse: + """Build the response returned when a workflow continues asynchronously.""" + base_url, route_prefix = split_request_url(req.url) + return func.HttpResponse( + json.dumps({ + "instanceId": instance_id, + "statusQueryGetUri": build_workflow_status_url( + base_url, workflow_name, instance_id, prefix=route_prefix + ), + "respondUri": build_workflow_respond_url( + base_url, workflow_name, instance_id, "{requestId}", prefix=route_prefix + ), + "message": "Workflow started", + }), + status_code=202, + mimetype=MIMETYPE_APPLICATION_JSON, + ) + + def _build_workflow_terminal_response(self, status: Any, instance_id: str) -> func.HttpResponse: + """Build a compact response for a completed or failed workflow.""" + if status is None or status.runtime_status is None: + return self._build_error_response( + f"Workflow orchestration '{instance_id}' returned no status.", + status_code=500, + ) + + runtime_status = status.runtime_status + if runtime_status not in { + df.OrchestrationRuntimeStatus.Completed, + df.OrchestrationRuntimeStatus.Failed, + }: + return self._build_error_response( + f"Workflow orchestration '{instance_id}' ended with unexpected status '{runtime_status.name}'.", + status_code=500, + ) + + decoded_output = deserialize_workflow_output(status.output) if status.output is not None else None + response: dict[str, Any] = { + "instanceId": status.instance_id or instance_id, + "runtimeStatus": runtime_status.name, + "output": decoded_output if runtime_status == df.OrchestrationRuntimeStatus.Completed else None, + } + if runtime_status == df.OrchestrationRuntimeStatus.Failed: + response["error"] = decoded_output + + return func.HttpResponse( + json.dumps(response, default=_json_default), + status_code=200, + mimetype=MIMETYPE_APPLICATION_JSON, + ) + def _create_http_response( self, payload: dict[str, Any] | str, @@ -1621,31 +1715,110 @@ def _parse_text_body(req: func.HttpRequest) -> tuple[dict[str, Any], str]: return {}, message - def _should_wait_for_response(self, req: func.HttpRequest, req_body: dict[str, Any]) -> bool: - """Determine whether the caller requested to wait for the response.""" + def _should_wait_for_response( + self, + req: func.HttpRequest, + req_body: dict[str, Any], + *, + query_parameter: str = WAIT_FOR_RESPONSE_FIELD, + default_value: bool = True, + ) -> bool: + """Determine whether the caller requested to wait for the response. + + The ``x-ms-wait-for-response`` header takes precedence, followed by ``query_parameter`` + (``wait_for_response`` for the snake_case agent endpoints, ``waitForResponse`` for the + camelCase workflow endpoints), and finally the ``wait_for_response`` request body field. + Values that cannot be parsed as a boolean are ignored so that the next source is consulted. + """ headers: dict[str, str] = self._extract_normalized_headers(req) header_value: str | None = headers.get(WAIT_FOR_RESPONSE_HEADER) - if header_value is not None: - return self._coerce_to_bool(header_value) + parsed_header = self._try_coerce_to_bool(header_value) + if parsed_header is not None: + return parsed_header params = req.params or {} - if WAIT_FOR_RESPONSE_FIELD in params: - return self._coerce_to_bool(params.get(WAIT_FOR_RESPONSE_FIELD)) + parsed_query = self._try_coerce_to_bool(params.get(query_parameter)) + if parsed_query is not None: + return parsed_query - if WAIT_FOR_RESPONSE_FIELD in req_body: - return self._coerce_to_bool(req_body.get(WAIT_FOR_RESPONSE_FIELD)) + parsed_body = self._try_coerce_to_bool(req_body.get(WAIT_FOR_RESPONSE_FIELD)) + if parsed_body is not None: + return parsed_body + return default_value - return True + @staticmethod + def _get_workflow_wait_timeout_seconds(req: func.HttpRequest) -> int: + """Get the positive workflow wait timeout from the query string.""" + value = (req.params or {}).get(_WORKFLOW_WAIT_TIMEOUT_SECONDS_QUERY_PARAMETER) + if value is None: + return _DEFAULT_WORKFLOW_WAIT_TIMEOUT_SECONDS - def _coerce_to_bool(self, value: Any) -> bool: - """Convert various representations into a boolean flag.""" + error_message = ( + f"'{_WORKFLOW_WAIT_TIMEOUT_SECONDS_QUERY_PARAMETER}' must be an integer between " + f"1 and {_MAX_WORKFLOW_WAIT_TIMEOUT_SECONDS}." + ) + try: + seconds = int(value) + except (TypeError, ValueError) as exc: + raise ValueError(error_message) from exc + + if seconds <= 0 or seconds > _MAX_WORKFLOW_WAIT_TIMEOUT_SECONDS: + raise ValueError(error_message) + + return seconds + + def _get_workflow_wait_for_response(self, req: func.HttpRequest) -> bool: + """Get the workflow wait preference while rejecting an invalid query value.""" + headers = self._extract_normalized_headers(req) + parsed_header = self._try_coerce_to_bool(headers.get(WAIT_FOR_RESPONSE_HEADER)) + if parsed_header is not None: + return parsed_header + + query_value = (req.params or {}).get(_WORKFLOW_WAIT_FOR_RESPONSE_QUERY_PARAMETER) + if query_value is None: + return False + + parsed_query = self._try_coerce_to_bool(query_value) + if parsed_query is None: + raise ValueError(f"'{_WORKFLOW_WAIT_FOR_RESPONSE_QUERY_PARAMETER}' must be a boolean value.") + + return parsed_query + + @staticmethod + def _validate_workflow_run_id(run_id: str | None) -> str | None: + """Validate a custom workflow run ID against the Durable Task instance ID contract.""" + if run_id is None: + return None + + if ( + not 1 <= len(run_id) <= _MAX_WORKFLOW_RUN_ID_LENGTH + or run_id.startswith("@") + or any(character in run_id for character in "/\\#?") + or any(unicodedata.category(character) == "Cc" for character in run_id) + ): + raise ValueError( + f"'{_RUN_ID_QUERY_PARAMETER}' must be between 1 and {_MAX_WORKFLOW_RUN_ID_LENGTH} characters, " + "must not start with '@', and must not contain '/', '\\', '#', '?', or control characters." + ) + + return run_id + + @staticmethod + def _try_coerce_to_bool(value: Any) -> bool | None: + """Convert recognized boolean representations, or return None.""" if isinstance(value, bool): return value - if value is None: - return False if isinstance(value, (int, float)): return bool(value) if isinstance(value, str): - return value.strip().lower() in {"true", "1", "yes", "y", "on"} - return False + normalized = value.strip().lower() + if normalized in {"true", "1", "yes", "y", "on"}: + return True + if normalized in {"false", "0", "no", "n", "off"}: + return False + return None + + def _coerce_to_bool(self, value: Any) -> bool: + """Convert various representations into a boolean flag.""" + return bool(self._try_coerce_to_bool(value)) diff --git a/python/packages/azurefunctions/tests/test_app.py b/python/packages/azurefunctions/tests/test_app.py index 71497ac..15fff90 100644 --- a/python/packages/azurefunctions/tests/test_app.py +++ b/python/packages/azurefunctions/tests/test_app.py @@ -26,6 +26,10 @@ ) from agent_framework_azurefunctions import AgentFunctionApp +from agent_framework_azurefunctions._app import ( + _DEFAULT_WORKFLOW_WAIT_TIMEOUT_SECONDS, + _MAX_WORKFLOW_WAIT_TIMEOUT_SECONDS, +) from agent_framework_azurefunctions._entities import create_agent_entity from agent_framework_azurefunctions._errors import IncomingRequestError @@ -337,7 +341,7 @@ def test_wait_for_response_body_snake_case(self) -> None: assert app._should_wait_for_response(request, {WAIT_FOR_RESPONSE_FIELD: "0"}) is False def test_wait_for_response_query_parameter(self) -> None: - """Test that query parameter controls wait_for_response.""" + """Test that the agent snake_case query parameter controls wait_for_response.""" app = self._create_app() request = self._make_request(params={WAIT_FOR_RESPONSE_FIELD: "true"}) @@ -350,6 +354,77 @@ def test_wait_for_response_query_precedence(self) -> None: assert app._should_wait_for_response(request, {WAIT_FOR_RESPONSE_FIELD: "true"}) is False + def test_wait_for_response_query_parameter_is_surface_specific(self) -> None: + """Test that each surface only honors the query parameter matching its casing convention.""" + app = self._create_app() + snake_case = self._make_request(params={WAIT_FOR_RESPONSE_FIELD: "true"}) + camel_case = self._make_request(params={"waitForResponse": "true"}) + + # Agent endpoints default to the snake_case name. + assert app._should_wait_for_response(snake_case, {}, default_value=False) is True + assert app._should_wait_for_response(camel_case, {}, default_value=False) is False + + # Workflow endpoints opt in to the camelCase name. + assert ( + app._should_wait_for_response(camel_case, {}, query_parameter="waitForResponse", default_value=False) + is True + ) + assert ( + app._should_wait_for_response(snake_case, {}, query_parameter="waitForResponse", default_value=False) + is False + ) + + def test_invalid_wait_for_response_header_falls_back_to_query(self) -> None: + """Test that an invalid header does not suppress a valid query option.""" + app = self._create_app() + request = self._make_request( + headers={WAIT_FOR_RESPONSE_HEADER: "invalid"}, + params={WAIT_FOR_RESPONSE_FIELD: "true"}, + ) + + assert app._should_wait_for_response(request, {}) is True + + def test_wait_for_response_uses_configured_default(self) -> None: + """Test that callers can select the fallback behavior.""" + app = self._create_app() + request = self._make_request() + + assert app._should_wait_for_response(request, {}, default_value=False) is False + assert app._should_wait_for_response(request, {WAIT_FOR_RESPONSE_FIELD: "invalid"}) is True + assert ( + app._should_wait_for_response( + request, + {WAIT_FOR_RESPONSE_FIELD: "invalid"}, + default_value=False, + ) + is False + ) + + @pytest.mark.parametrize( + ("value", "expected"), + [ + (None, _DEFAULT_WORKFLOW_WAIT_TIMEOUT_SECONDS), + ("1", 1), + ("30", 30), + (str(_MAX_WORKFLOW_WAIT_TIMEOUT_SECONDS), _MAX_WORKFLOW_WAIT_TIMEOUT_SECONDS), + ], + ) + def test_workflow_wait_timeout_seconds(self, value: str | None, expected: int) -> None: + """Test workflow wait timeout parsing.""" + app = self._create_app() + request = self._make_request(params={"timeoutSeconds": value} if value is not None else None) + + assert app._get_workflow_wait_timeout_seconds(request) == expected + + @pytest.mark.parametrize("value", ["0", "-1", str(_MAX_WORKFLOW_WAIT_TIMEOUT_SECONDS + 1), "invalid"]) + def test_workflow_wait_timeout_seconds_rejects_invalid_values(self, value: str) -> None: + """Test workflow wait timeout validation.""" + app = self._create_app() + request = self._make_request(params={"timeoutSeconds": value}) + + with pytest.raises(ValueError, match=rf"between 1 and {_MAX_WORKFLOW_WAIT_TIMEOUT_SECONDS}"): + app._get_workflow_wait_timeout_seconds(request) + class TestAgentEntityOperations: """Test suite for entity operations.""" @@ -941,6 +1016,318 @@ async def test_http_run_rejects_empty_message(self) -> None: client.signal_entity.assert_not_called() +class TestWorkflowRunRoute: + """Tests for the workflow HTTP run route behavior.""" + + @staticmethod + def _get_run_handler(workflow_name: str) -> Callable[[func.HttpRequest, Any], Awaitable[func.HttpResponse]]: + captured_handlers: dict[str | None, Callable[..., Awaitable[func.HttpResponse]]] = {} + + def capture_decorator(*args: Any, **kwargs: Any) -> Callable[[FuncT], FuncT]: + def decorator(func: FuncT) -> FuncT: + return func + + return decorator + + def capture_route(*args: Any, **kwargs: Any) -> Callable[[FuncT], FuncT]: + def decorator(func: FuncT) -> FuncT: + captured_handlers[kwargs.get("route")] = func + return func + + return decorator + + workflow = Mock() + workflow.name = workflow_name + app = AgentFunctionApp(enable_health_check=False) + + with ( + patch.object(AgentFunctionApp, "function_name", new=capture_decorator), + patch.object(AgentFunctionApp, "route", new=capture_route), + patch.object(AgentFunctionApp, "durable_client_input", new=capture_decorator), + ): + app._register_workflow_routes(workflow) + + return captured_handlers[f"workflow/{workflow_name}/run"] + + async def test_wait_for_response_query_waits_with_timeout(self) -> None: + """Test synchronous workflow invocation through query options.""" + handler = self._get_run_handler("test_workflow") + request = Mock() + request.headers = {} + request.params = {"waitForResponse": "true", "timeoutSeconds": "30", "runId": "custom-run"} + request.get_json.return_value = {"message": "hello"} + + client = AsyncMock() + client.start_new.return_value = "custom-run" + client.wait_for_completion_or_create_check_status_response.return_value = func.HttpResponse( + "completed", status_code=200 + ) + client.get_status.return_value = Mock( + instance_id="custom-run", + runtime_status=df.OrchestrationRuntimeStatus.Completed, + output="completed", + ) + + response = await handler(request, client) + + assert response.status_code == 200 + assert json.loads(response.get_body()) == { + "instanceId": "custom-run", + "runtimeStatus": "Completed", + "output": "completed", + } + client.wait_for_completion_or_create_check_status_response.assert_awaited_once_with( + request, + "custom-run", + timeout_in_milliseconds=30_000, + retry_interval_in_milliseconds=1000, + ) + client.start_new.assert_awaited_once_with( + "dafx-test_workflow", + instance_id="custom-run", + client_input={"message": "hello"}, + ) + + async def test_wait_for_response_header_waits_with_default_timeout(self) -> None: + """Test that the legacy wait header remains supported.""" + handler = self._get_run_handler("test_workflow") + request = Mock() + request.headers = {WAIT_FOR_RESPONSE_HEADER: "true"} + request.params = {} + request.get_json.return_value = {"message": "hello"} + + client = AsyncMock() + client.start_new.return_value = "instance-1" + client.wait_for_completion_or_create_check_status_response.return_value = func.HttpResponse( + "completed", status_code=200 + ) + client.get_status.return_value = Mock( + instance_id="instance-1", + runtime_status=df.OrchestrationRuntimeStatus.Completed, + output="completed", + ) + + response = await handler(request, client) + + assert response.status_code == 200 + assert json.loads(response.get_body()) == { + "instanceId": "instance-1", + "runtimeStatus": "Completed", + "output": "completed", + } + client.wait_for_completion_or_create_check_status_response.assert_awaited_once_with( + request, + "instance-1", + timeout_in_milliseconds=_DEFAULT_WORKFLOW_WAIT_TIMEOUT_SECONDS * 1000, + retry_interval_in_milliseconds=1000, + ) + + async def test_default_returns_async_workflow_handle(self) -> None: + """Test that asynchronous workflow invocation remains the default.""" + handler = self._get_run_handler("test_workflow") + request = Mock() + request.url = "http://localhost:7071/api/workflow/test_workflow/run" + request.headers = {} + request.params = {} + request.get_json.return_value = {"message": "hello"} + + client = AsyncMock() + client.start_new.return_value = "instance-1" + + response = await handler(request, client) + + assert response.status_code == 202 + assert json.loads(response.get_body())["instanceId"] == "instance-1" + client.wait_for_completion_or_create_check_status_response.assert_not_awaited() + + async def test_wait_timeout_returns_same_async_workflow_handle(self) -> None: + """Test that a wait timeout preserves the default asynchronous response contract.""" + handler = self._get_run_handler("test_workflow") + async_request = Mock() + async_request.url = "http://localhost:7071/api/workflow/test_workflow/run" + async_request.headers = {} + async_request.params = {} + async_request.get_json.return_value = {"message": "hello"} + async_client = AsyncMock() + async_client.start_new.return_value = "instance-1" + + timeout_request = Mock() + timeout_request.url = async_request.url + timeout_request.headers = {} + timeout_request.params = {"waitForResponse": "true"} + timeout_request.get_json.return_value = {"message": "hello"} + timeout_client = AsyncMock() + timeout_client.start_new.return_value = "instance-1" + timeout_client.wait_for_completion_or_create_check_status_response.return_value = func.HttpResponse( + json.dumps({"id": "instance-1", "statusQueryGetUri": "https://durable-webhook.example/status"}), + status_code=202, + mimetype=MIMETYPE_APPLICATION_JSON, + ) + + async_response = await handler(async_request, async_client) + timeout_response = await handler(timeout_request, timeout_client) + + assert timeout_response.status_code == async_response.status_code + assert timeout_response.mimetype == async_response.mimetype + assert timeout_response.headers == async_response.headers + assert timeout_response.get_body() == async_response.get_body() + + async def test_wait_failure_returns_domain_result(self) -> None: + """Test that workflow failure is returned as a successful HTTP domain result.""" + handler = self._get_run_handler("test_workflow") + request = Mock() + request.url = "http://localhost:7071/api/workflow/test_workflow/run" + request.headers = {} + request.params = {"waitForResponse": "true"} + request.get_json.return_value = {"message": "hello"} + client = AsyncMock() + client.start_new.return_value = "instance-1" + client.wait_for_completion_or_create_check_status_response.return_value = func.HttpResponse( + json.dumps({"runtimeStatus": "Failed", "output": "Something went wrong"}), + status_code=500, + mimetype=MIMETYPE_APPLICATION_JSON, + ) + client.get_status.return_value = Mock( + instance_id="instance-1", + runtime_status=df.OrchestrationRuntimeStatus.Failed, + output="Something went wrong", + ) + + response = await handler(request, client) + + assert response.status_code == 200 + assert response.mimetype == MIMETYPE_APPLICATION_JSON + assert json.loads(response.get_body()) == { + "instanceId": "instance-1", + "runtimeStatus": "Failed", + "output": None, + "error": "Something went wrong", + } + + async def test_wait_completion_decodes_typed_output(self) -> None: + """Test that synchronous completion uses the shared workflow output decoder.""" + handler = self._get_run_handler("test_workflow") + request = Mock() + request.headers = {} + request.params = {"waitForResponse": "true"} + request.get_json.return_value = {"message": "hello"} + encoded_output = {"__pickled__": "checkpoint-data"} + client = AsyncMock() + client.start_new.return_value = "instance-1" + client.wait_for_completion_or_create_check_status_response.return_value = func.HttpResponse( + json.dumps(encoded_output), status_code=200 + ) + client.get_status.return_value = Mock( + instance_id="instance-1", + runtime_status=df.OrchestrationRuntimeStatus.Completed, + output=encoded_output, + ) + + with patch( + "agent_framework_azurefunctions._app.deserialize_workflow_output", + return_value={"approved": True}, + ) as deserialize: + response = await handler(request, client) + + deserialize.assert_called_once_with(encoded_output) + assert json.loads(response.get_body())["output"] == {"approved": True} + + async def test_wait_terminated_returns_unexpected_status_error(self) -> None: + """Test that an unexpected terminal status matches the .NET endpoint semantics.""" + handler = self._get_run_handler("test_workflow") + request = Mock() + request.headers = {} + request.params = {"waitForResponse": "true"} + request.get_json.return_value = {"message": "hello"} + client = AsyncMock() + client.start_new.return_value = "instance-1" + client.wait_for_completion_or_create_check_status_response.return_value = func.HttpResponse( + json.dumps({"runtimeStatus": "Terminated"}), + status_code=200, + mimetype=MIMETYPE_APPLICATION_JSON, + ) + client.get_status.return_value = Mock( + instance_id="instance-1", + runtime_status=df.OrchestrationRuntimeStatus.Terminated, + output=None, + ) + + response = await handler(request, client) + + assert response.status_code == 500 + assert "unexpected status 'Terminated'" in response.get_body().decode("utf-8") + + async def test_invalid_wait_timeout_does_not_start_workflow(self) -> None: + """Test invalid synchronous timeout rejection before scheduling.""" + handler = self._get_run_handler("test_workflow") + request = Mock() + request.headers = {} + request.params = {"waitForResponse": "true", "timeoutSeconds": "0"} + client = AsyncMock() + + response = await handler(request, client) + + assert response.status_code == 400 + client.start_new.assert_not_awaited() + + async def test_invalid_wait_for_response_does_not_start_workflow(self) -> None: + """Test invalid synchronous wait rejection before scheduling.""" + handler = self._get_run_handler("test_workflow") + request = Mock() + request.headers = {} + request.params = {"waitForResponse": "invalid"} + client = AsyncMock() + + response = await handler(request, client) + + assert response.status_code == 400 + assert "waitForResponse" in response.get_body().decode("utf-8") + client.start_new.assert_not_awaited() + + async def test_wait_header_takes_precedence_over_invalid_query(self) -> None: + """Test that the legacy wait header retains precedence over the query parameter.""" + handler = self._get_run_handler("test_workflow") + request = Mock() + request.headers = {WAIT_FOR_RESPONSE_HEADER: "false"} + request.params = {"waitForResponse": "invalid"} + request.url = "http://localhost:7071/api/workflow/test_workflow/run" + request.get_json.return_value = {"message": "hello"} + client = AsyncMock() + client.start_new.return_value = "instance-1" + + response = await handler(request, client) + + assert response.status_code == 202 + client.wait_for_completion_or_create_check_status_response.assert_not_awaited() + + @pytest.mark.parametrize( + "run_id", + [ + "", + "@workflow", + "workflow/123", + "workflow\\123", + "workflow#123", + "workflow?123", + "workflow\n123", + "x" * 101, + ], + ) + async def test_invalid_run_id_does_not_start_workflow(self, run_id: str) -> None: + """Test that invalid custom run IDs are rejected before scheduling.""" + handler = self._get_run_handler("test_workflow") + request = Mock() + request.headers = {} + request.params = {"runId": run_id} + client = AsyncMock() + + response = await handler(request, client) + + assert response.status_code == 400 + assert "runId" in response.get_body().decode("utf-8") + client.start_new.assert_not_awaited() + + class TestMCPToolEndpoint: """Test suite for MCP tool endpoint functionality.""" diff --git a/python/packages/azurefunctions/tests/test_azurefunctions_workflow_initial_input.py b/python/packages/azurefunctions/tests/test_azurefunctions_workflow_initial_input.py index 7ec3a9f..b1de0bc 100644 --- a/python/packages/azurefunctions/tests/test_azurefunctions_workflow_initial_input.py +++ b/python/packages/azurefunctions/tests/test_azurefunctions_workflow_initial_input.py @@ -63,6 +63,8 @@ async def test_workflow_run_route_neutralizes_reserved_marker_shaped_input() -> "__pickled__": "not-checkpoint-data", "__type__": "builtins:int", } + request.headers = {} + request.params = {} request.url = "https://example.test/api/workflow/input_boundary/run" client = AsyncMock() client.start_new.return_value = "instance-1" diff --git a/python/samples/azure_functions/09_workflow_shared_state/README.md b/python/samples/azure_functions/09_workflow_shared_state/README.md index 6fd2a6b..0c8ec16 100644 --- a/python/samples/azure_functions/09_workflow_shared_state/README.md +++ b/python/samples/azure_functions/09_workflow_shared_state/README.md @@ -81,6 +81,18 @@ curl -X POST http://localhost:7071/api/workflow/email_triage_shared_state/run \ -d '"Hi team, reminder about our meeting tomorrow at 10 AM."' ``` +### Wait for a Result + +Add `waitForResponse=true` to wait up to 30 seconds for the workflow result: + +```bash +curl -X POST "http://localhost:7071/api/workflow/email_triage_shared_state/run?waitForResponse=true&timeoutSeconds=30" \ + -H "Content-Type: application/json" \ + -d '"Hi team, reminder about our meeting tomorrow at 10 AM."' +``` + +`timeoutSeconds` must be an integer from 1 to 200. If the workflow does not finish before the timeout, the endpoint returns the same `202 Accepted` workflow handle as the default asynchronous invocation, including `statusQueryGetUri` and `respondUri`. Completed and failed workflows return `200 OK`; inspect the response payload to distinguish the workflow outcome. Use `runId` to supply a custom workflow run identifier. + ## Expected Output **Spam email:** diff --git a/python/samples/azure_functions/09_workflow_shared_state/demo.http b/python/samples/azure_functions/09_workflow_shared_state/demo.http index 50f0ef0..40aa678 100644 --- a/python/samples/azure_functions/09_workflow_shared_state/demo.http +++ b/python/samples/azure_functions/09_workflow_shared_state/demo.http @@ -12,6 +12,12 @@ Content-Type: application/json "Hi team, just a reminder about the sprint planning meeting tomorrow at 10 AM. Please review the agenda items in Jira before the call." +### Start the workflow and wait up to 30 seconds for its result +POST {{endpoint}}/api/workflow/email_triage_shared_state/run?waitForResponse=true&timeoutSeconds=30 +Content-Type: application/json + +"Hi team, reminder about our meeting tomorrow at 10 AM." + ### Start the workflow with another legitimate email POST {{endpoint}}/api/workflow/email_triage_shared_state/run Content-Type: application/json