mirror of
https://github.com/microsoft/agent-framework.git
synced 2026-06-16 21:04:09 +08:00
.NET: [Feature Branch] Add Human In the Loop support for durable workflows (#4358)
* Add Azure Functions HITL workflow sample Add 06_WorkflowHITL Azure Functions sample demonstrating Human-in-the-Loop workflow support with HTTP endpoints for status checking and approval responses. The sample includes: - ExpenseReimbursement workflow with RequestPort for manager approval - Custom HTTP endpoint to check workflow status and pending approvals - Custom HTTP endpoint to send approval responses via RaiseEventAsync - demo.http file with step-by-step interaction examples * PR feedback fixes * Minor comment cleanup * Minor comment clReverted the `!context.IsReplaying` guards on `PendingEvents.Add`/`RemoveAll` and `SetCustomStatus` in `ExecuteRequestPortAsync`. The guards broke fan-out scenarios where parallel RequestPorts need to be discoverable after replay. `SetCustomStatus` is idempotent metadata that doesn't affect replay determinism.eanup * fix for PR feedback * PR feedback updates * Improvements to samples * Improvements to README * Update samples to use parallel request ports. * Unit tests * Introduce local variables to improve readability of Workflows.Workflows access patter * Use GitHub-style callouts and add PowerShell command variants in HITL sample README
This commit is contained in:
Shyju Krishnankutty
committed by
GitHub
Unverified
GitHub
parent
2988568cab
commit
ad51aee47b
@@ -55,10 +55,12 @@
|
||||
<Project Path="samples/Durable/Workflow/ConsoleApps/05_WorkflowEvents/05_WorkflowEvents.csproj" />
|
||||
<Project Path="samples/Durable/Workflow/ConsoleApps/06_WorkflowSharedState/06_WorkflowSharedState.csproj" />
|
||||
<Project Path="samples/Durable/Workflow/ConsoleApps/07_SubWorkflows/07_SubWorkflows.csproj" />
|
||||
<Project Path="samples/Durable/Workflow/ConsoleApps/08_WorkflowHITL/08_WorkflowHITL.csproj" />
|
||||
</Folder>
|
||||
<Folder Name="/Samples/Durable/Workflows/AzureFunctions/">
|
||||
<Project Path="samples/Durable/Workflow/AzureFunctions/01_SequentialWorkflow/01_SequentialWorkflow.csproj" />
|
||||
<Project Path="samples/Durable/Workflow/AzureFunctions/02_ConcurrentWorkflow/02_ConcurrentWorkflow.csproj" />
|
||||
<Project Path="samples/Durable/Workflow/AzureFunctions/03_WorkflowHITL/03_WorkflowHITL.csproj" />
|
||||
</Folder>
|
||||
<Folder Name="/Samples/GettingStarted/">
|
||||
<File Path="samples/GettingStarted/README.md" />
|
||||
@@ -475,4 +477,4 @@
|
||||
<Project Path="tests/Microsoft.Agents.AI.Workflows.Generators.UnitTests/Microsoft.Agents.AI.Workflows.Generators.UnitTests.csproj" />
|
||||
<Project Path="tests/Microsoft.Agents.AI.Workflows.UnitTests/Microsoft.Agents.AI.Workflows.UnitTests.csproj" />
|
||||
</Folder>
|
||||
</Solution>
|
||||
</Solution>
|
||||
|
||||
@@ -0,0 +1,43 @@
|
||||
<Project Sdk="Microsoft.NET.Sdk">
|
||||
<PropertyGroup>
|
||||
<TargetFrameworks>net10.0</TargetFrameworks>
|
||||
<AzureFunctionsVersion>v4</AzureFunctionsVersion>
|
||||
<OutputType>Exe</OutputType>
|
||||
<ImplicitUsings>enable</ImplicitUsings>
|
||||
<Nullable>enable</Nullable>
|
||||
<!-- The Functions build tools don't like namespaces that start with a number -->
|
||||
<AssemblyName>WorkflowHITLFunctions</AssemblyName>
|
||||
<RootNamespace>WorkflowHITLFunctions</RootNamespace>
|
||||
</PropertyGroup>
|
||||
|
||||
<ItemGroup>
|
||||
<FrameworkReference Include="Microsoft.AspNetCore.App" />
|
||||
</ItemGroup>
|
||||
|
||||
<ItemGroup>
|
||||
<None Include="local.settings.json" />
|
||||
</ItemGroup>
|
||||
|
||||
<!-- Azure Functions packages -->
|
||||
<ItemGroup>
|
||||
<PackageReference Include="Microsoft.Azure.Functions.Worker" />
|
||||
<PackageReference Include="Microsoft.Azure.Functions.Worker.Extensions.DurableTask" />
|
||||
<PackageReference Include="Microsoft.Azure.Functions.Worker.Extensions.DurableTask.AzureManaged" />
|
||||
<PackageReference Include="Microsoft.Azure.Functions.Worker.Extensions.Http.AspNetCore" />
|
||||
<PackageReference Include="Microsoft.Azure.Functions.Worker.Sdk" />
|
||||
</ItemGroup>
|
||||
|
||||
<ItemGroup>
|
||||
<PackageReference Include="Azure.Identity" />
|
||||
</ItemGroup>
|
||||
|
||||
<!-- Local projects that should be switched to package references when using the sample outside of this MAF repo -->
|
||||
<!--
|
||||
<ItemGroup>
|
||||
<PackageReference Include="Microsoft.Agents.AI.Hosting.AzureFunctions" />
|
||||
</ItemGroup>
|
||||
-->
|
||||
<ItemGroup>
|
||||
<ProjectReference Include="..\..\..\..\..\src\Microsoft.Agents.AI.Hosting.AzureFunctions\Microsoft.Agents.AI.Hosting.AzureFunctions.csproj" />
|
||||
</ItemGroup>
|
||||
</Project>
|
||||
@@ -0,0 +1,63 @@
|
||||
// Copyright (c) Microsoft. All rights reserved.
|
||||
|
||||
using Microsoft.Agents.AI.Workflows;
|
||||
|
||||
namespace WorkflowHITLFunctions;
|
||||
|
||||
/// <summary>Expense approval request passed to the RequestPort.</summary>
|
||||
public record ApprovalRequest(string ExpenseId, decimal Amount, string EmployeeName);
|
||||
|
||||
/// <summary>Approval response received from the RequestPort.</summary>
|
||||
public record ApprovalResponse(bool Approved, string? Comments);
|
||||
|
||||
/// <summary>Looks up expense details and creates an approval request.</summary>
|
||||
internal sealed class CreateApprovalRequest() : Executor<string, ApprovalRequest>("RetrieveRequest")
|
||||
{
|
||||
public override ValueTask<ApprovalRequest> HandleAsync(
|
||||
string message,
|
||||
IWorkflowContext context,
|
||||
CancellationToken cancellationToken = default)
|
||||
{
|
||||
// In a real scenario, this would look up expense details from a database
|
||||
return new ValueTask<ApprovalRequest>(new ApprovalRequest(message, 1500.00m, "Jerry"));
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>Prepares the approval request for finance review after manager approval.</summary>
|
||||
internal sealed class PrepareFinanceReview() : Executor<ApprovalResponse, ApprovalRequest>("PrepareFinanceReview")
|
||||
{
|
||||
public override ValueTask<ApprovalRequest> HandleAsync(
|
||||
ApprovalResponse message,
|
||||
IWorkflowContext context,
|
||||
CancellationToken cancellationToken = default)
|
||||
{
|
||||
if (!message.Approved)
|
||||
{
|
||||
throw new InvalidOperationException("Cannot proceed to finance review — manager denied the expense.");
|
||||
}
|
||||
|
||||
// In a real scenario, this would retrieve the original expense details
|
||||
return new ValueTask<ApprovalRequest>(new ApprovalRequest("EXP-2025-001", 1500.00m, "Jerry"));
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>Processes the expense reimbursement based on the parallel approval responses.</summary>
|
||||
internal sealed class ExpenseReimburse() : Executor<ApprovalResponse[], string>("Reimburse")
|
||||
{
|
||||
public override async ValueTask<string> HandleAsync(
|
||||
ApprovalResponse[] message,
|
||||
IWorkflowContext context,
|
||||
CancellationToken cancellationToken = default)
|
||||
{
|
||||
// Check that all parallel approvals passed
|
||||
ApprovalResponse? denied = Array.Find(message, r => !r.Approved);
|
||||
if (denied is not null)
|
||||
{
|
||||
return $"Expense reimbursement denied. Comments: {denied.Comments}";
|
||||
}
|
||||
|
||||
// Simulate payment processing
|
||||
await Task.Delay(1000, cancellationToken);
|
||||
return $"Expense reimbursed at {DateTime.UtcNow:O}";
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,51 @@
|
||||
// Copyright (c) Microsoft. All rights reserved.
|
||||
|
||||
// This sample demonstrates a Human-in-the-Loop (HITL) workflow hosted in Azure Functions.
|
||||
//
|
||||
// ┌──────────────────────┐ ┌────────────────┐ ┌─────────────────────┐ ┌────────────────────┐
|
||||
// │ CreateApprovalRequest│──►│ManagerApproval │──►│PrepareFinanceReview │──┬►│ BudgetApproval │──┐
|
||||
// └──────────────────────┘ │ (RequestPort) │ └─────────────────────┘ │ │ (RequestPort) │ │
|
||||
// └────────────────┘ │ └────────────────────┘ │ ┌─────────────────┐
|
||||
// │ ├─►│ExpenseReimburse │
|
||||
// │ ┌────────────────────┐ │ └─────────────────┘
|
||||
// └►│ComplianceApproval │──┘
|
||||
// │ (RequestPort) │
|
||||
// └────────────────────┘
|
||||
//
|
||||
// The workflow pauses at three RequestPorts — one for the manager, then two in parallel for finance.
|
||||
// After manager approval, BudgetApproval and ComplianceApproval run concurrently via fan-out/fan-in.
|
||||
// The framework auto-generates three HTTP endpoints for each workflow:
|
||||
// POST /api/workflows/{name}/run - Start the workflow
|
||||
// GET /api/workflows/{name}/status/{id} - Check status and pending approvals
|
||||
// POST /api/workflows/{name}/respond/{id} - Send approval response to resume
|
||||
|
||||
using Microsoft.Agents.AI.Hosting.AzureFunctions;
|
||||
using Microsoft.Agents.AI.Workflows;
|
||||
using Microsoft.Azure.Functions.Worker.Builder;
|
||||
using Microsoft.Extensions.Hosting;
|
||||
using WorkflowHITLFunctions;
|
||||
|
||||
// Define executors and RequestPorts for the three HITL pause points
|
||||
CreateApprovalRequest createRequest = new();
|
||||
RequestPort<ApprovalRequest, ApprovalResponse> managerApproval = RequestPort.Create<ApprovalRequest, ApprovalResponse>("ManagerApproval");
|
||||
PrepareFinanceReview prepareFinanceReview = new();
|
||||
RequestPort<ApprovalRequest, ApprovalResponse> budgetApproval = RequestPort.Create<ApprovalRequest, ApprovalResponse>("BudgetApproval");
|
||||
RequestPort<ApprovalRequest, ApprovalResponse> complianceApproval = RequestPort.Create<ApprovalRequest, ApprovalResponse>("ComplianceApproval");
|
||||
ExpenseReimburse reimburse = new();
|
||||
|
||||
// Build the workflow: CreateApprovalRequest -> ManagerApproval -> PrepareFinanceReview -> [BudgetApproval AND ComplianceApproval] -> ExpenseReimburse
|
||||
Workflow expenseApproval = new WorkflowBuilder(createRequest)
|
||||
.WithName("ExpenseReimbursement")
|
||||
.WithDescription("Expense reimbursement with manager and parallel finance approvals")
|
||||
.AddEdge(createRequest, managerApproval)
|
||||
.AddEdge(managerApproval, prepareFinanceReview)
|
||||
.AddFanOutEdge(prepareFinanceReview, [budgetApproval, complianceApproval])
|
||||
.AddFanInEdge([budgetApproval, complianceApproval], reimburse)
|
||||
.Build();
|
||||
|
||||
using IHost app = FunctionsApplication
|
||||
.CreateBuilder(args)
|
||||
.ConfigureFunctionsWebApplication()
|
||||
.ConfigureDurableWorkflows(workflows => workflows.AddWorkflow(expenseApproval, exposeStatusEndpoint: true))
|
||||
.Build();
|
||||
app.Run();
|
||||
@@ -0,0 +1,266 @@
|
||||
# Human-in-the-Loop (HITL) Workflow — Azure Functions
|
||||
|
||||
This sample demonstrates a durable workflow with Human-in-the-Loop support hosted in Azure Functions. The workflow pauses at three `RequestPort` nodes — one sequential manager approval, then two parallel finance approvals (budget and compliance) via fan-out/fan-in. Approval responses are sent via HTTP endpoints.
|
||||
|
||||
## Key Concepts Demonstrated
|
||||
|
||||
- Using multiple `RequestPort` nodes for sequential and parallel human-in-the-loop interactions in a durable workflow
|
||||
- Fan-out/fan-in pattern for parallel approval steps
|
||||
- Auto-generated HTTP endpoints for running workflows, checking status, and sending HITL responses
|
||||
- Pausing orchestrations via `WaitForExternalEvent` and resuming via `RaiseEventAsync`
|
||||
- Viewing inputs the workflow is waiting for via the status endpoint
|
||||
|
||||
## Workflow
|
||||
|
||||
This sample implements the following workflow:
|
||||
|
||||
```
|
||||
┌──────────────────────┐ ┌────────────────┐ ┌─────────────────────┐ ┌────────────────────┐
|
||||
│ CreateApprovalRequest│──►│ManagerApproval │──►│PrepareFinanceReview │──┬►│ BudgetApproval │──┐
|
||||
└──────────────────────┘ │ (RequestPort) │ └─────────────────────┘ │ │ (RequestPort) │ │
|
||||
└────────────────┘ │ └────────────────────┘ │ ┌─────────────────┐
|
||||
│ ├─►│ExpenseReimburse │
|
||||
│ ┌────────────────────┐ │ └─────────────────┘
|
||||
└►│ComplianceApproval │──┘
|
||||
│ (RequestPort) │
|
||||
└────────────────────┘
|
||||
```
|
||||
|
||||
## HTTP Endpoints
|
||||
|
||||
The framework auto-generates these endpoints for workflows with `RequestPort` nodes:
|
||||
|
||||
| Method | Endpoint | Description |
|
||||
|--------|----------|-------------|
|
||||
| POST | `/api/workflows/ExpenseReimbursement/run` | Start the workflow |
|
||||
| GET | `/api/workflows/ExpenseReimbursement/status/{runId}` | Check status and inputs the workflow is waiting for |
|
||||
| POST | `/api/workflows/ExpenseReimbursement/respond/{runId}` | Send approval response to resume |
|
||||
|
||||
## Environment Setup
|
||||
|
||||
See the [README.md](../../README.md) file in the parent directory for information on how to configure the environment, including how to install and run the Durable Task Scheduler.
|
||||
|
||||
## Running the Sample
|
||||
|
||||
With the environment setup and function app running, you can test the sample by sending HTTP requests to the workflow endpoints.
|
||||
|
||||
You can use the `demo.http` file to trigger the workflow, or a command line tool like `curl` as shown below:
|
||||
|
||||
### Step 1: Start the Workflow
|
||||
|
||||
Bash (Linux/macOS/WSL):
|
||||
|
||||
```bash
|
||||
curl -X POST http://localhost:7071/api/workflows/ExpenseReimbursement/run \
|
||||
-H "Content-Type: text/plain" -d "EXP-2025-001"
|
||||
```
|
||||
|
||||
PowerShell:
|
||||
|
||||
```powershell
|
||||
Invoke-RestMethod -Method Post `
|
||||
-Uri http://localhost:7071/api/workflows/ExpenseReimbursement/run `
|
||||
-ContentType text/plain `
|
||||
-Body "EXP-2025-001"
|
||||
```
|
||||
|
||||
The response will confirm the workflow orchestration has started:
|
||||
|
||||
```text
|
||||
Workflow orchestration started for ExpenseReimbursement. Orchestration runId: abc123def456
|
||||
```
|
||||
|
||||
> [!TIP]
|
||||
> You can provide a custom run ID by appending a `runId` query parameter:
|
||||
>
|
||||
> Bash (Linux/macOS/WSL):
|
||||
>
|
||||
> ```bash
|
||||
> curl -X POST "http://localhost:7071/api/workflows/ExpenseReimbursement/run?runId=expense-001" \
|
||||
> -H "Content-Type: text/plain" -d "EXP-2025-001"
|
||||
> ```
|
||||
>
|
||||
> PowerShell:
|
||||
>
|
||||
> ```powershell
|
||||
> Invoke-RestMethod -Method Post `
|
||||
> -Uri "http://localhost:7071/api/workflows/ExpenseReimbursement/run?runId=expense-001" `
|
||||
> -ContentType text/plain `
|
||||
> -Body "EXP-2025-001"
|
||||
> ```
|
||||
>
|
||||
> If not provided, a unique run ID is auto-generated.
|
||||
|
||||
### Step 2: Check Workflow Status
|
||||
|
||||
The workflow pauses at the `ManagerApproval` RequestPort. Query the status endpoint to see what input it is waiting for:
|
||||
|
||||
Bash (Linux/macOS/WSL):
|
||||
|
||||
```bash
|
||||
curl http://localhost:7071/api/workflows/ExpenseReimbursement/status/{runId}
|
||||
```
|
||||
|
||||
PowerShell:
|
||||
|
||||
```powershell
|
||||
Invoke-RestMethod -Uri http://localhost:7071/api/workflows/ExpenseReimbursement/status/{runId}
|
||||
```
|
||||
|
||||
```json
|
||||
{
|
||||
"runId": "{runId}",
|
||||
"status": "Running",
|
||||
"waitingForInput": [
|
||||
{ "eventName": "ManagerApproval", "input": { "ExpenseId": "EXP-2025-001", "Amount": 1500.00, "EmployeeName": "Jerry" } }
|
||||
]
|
||||
}
|
||||
```
|
||||
|
||||
> [!TIP]
|
||||
> You can also verify this in the DTS dashboard at `http://localhost:8082`. Find the orchestration by its `runId` and you will see it is in a "Running" state, paused at a `WaitForExternalEvent` call for the `ManagerApproval` event.
|
||||
|
||||
### Step 3: Send Manager Approval Response
|
||||
|
||||
Bash (Linux/macOS/WSL):
|
||||
|
||||
```bash
|
||||
curl -X POST http://localhost:7071/api/workflows/ExpenseReimbursement/respond/{runId} \
|
||||
-H "Content-Type: application/json" \
|
||||
-d '{"eventName": "ManagerApproval", "response": {"Approved": true, "Comments": "Approved by manager."}}'
|
||||
```
|
||||
|
||||
PowerShell:
|
||||
|
||||
```powershell
|
||||
Invoke-RestMethod -Method Post `
|
||||
-Uri http://localhost:7071/api/workflows/ExpenseReimbursement/respond/{runId} `
|
||||
-ContentType application/json `
|
||||
-Body '{"eventName": "ManagerApproval", "response": {"Approved": true, "Comments": "Approved by manager."}}'
|
||||
```
|
||||
|
||||
```json
|
||||
{
|
||||
"message": "Response sent to workflow.",
|
||||
"runId": "{runId}",
|
||||
"eventName": "ManagerApproval",
|
||||
"validated": true
|
||||
}
|
||||
```
|
||||
|
||||
### Step 4: Check Workflow Status Again
|
||||
|
||||
The workflow now pauses at both the `BudgetApproval` and `ComplianceApproval` RequestPorts in parallel:
|
||||
|
||||
Bash (Linux/macOS/WSL):
|
||||
|
||||
```bash
|
||||
curl http://localhost:7071/api/workflows/ExpenseReimbursement/status/{runId}
|
||||
```
|
||||
|
||||
PowerShell:
|
||||
|
||||
```powershell
|
||||
Invoke-RestMethod -Uri http://localhost:7071/api/workflows/ExpenseReimbursement/status/{runId}
|
||||
```
|
||||
|
||||
```json
|
||||
{
|
||||
"runId": "{runId}",
|
||||
"status": "Running",
|
||||
"waitingForInput": [
|
||||
{ "eventName": "BudgetApproval", "input": { "ExpenseId": "EXP-2025-001", "Amount": 1500.00, "EmployeeName": "Jerry" } },
|
||||
{ "eventName": "ComplianceApproval", "input": { "ExpenseId": "EXP-2025-001", "Amount": 1500.00, "EmployeeName": "Jerry" } }
|
||||
]
|
||||
}
|
||||
```
|
||||
|
||||
### Step 5a: Send Budget Approval Response
|
||||
|
||||
Bash (Linux/macOS/WSL):
|
||||
|
||||
```bash
|
||||
curl -X POST http://localhost:7071/api/workflows/ExpenseReimbursement/respond/{runId} \
|
||||
-H "Content-Type: application/json" \
|
||||
-d '{"eventName": "BudgetApproval", "response": {"Approved": true, "Comments": "Budget approved."}}'
|
||||
```
|
||||
|
||||
PowerShell:
|
||||
|
||||
```powershell
|
||||
Invoke-RestMethod -Method Post `
|
||||
-Uri http://localhost:7071/api/workflows/ExpenseReimbursement/respond/{runId} `
|
||||
-ContentType application/json `
|
||||
-Body '{"eventName": "BudgetApproval", "response": {"Approved": true, "Comments": "Budget approved."}}'
|
||||
```
|
||||
|
||||
```json
|
||||
{
|
||||
"message": "Response sent to workflow.",
|
||||
"runId": "{runId}",
|
||||
"eventName": "BudgetApproval",
|
||||
"validated": true
|
||||
}
|
||||
```
|
||||
|
||||
### Step 5b: Send Compliance Approval Response
|
||||
|
||||
Bash (Linux/macOS/WSL):
|
||||
|
||||
```bash
|
||||
curl -X POST http://localhost:7071/api/workflows/ExpenseReimbursement/respond/{runId} \
|
||||
-H "Content-Type: application/json" \
|
||||
-d '{"eventName": "ComplianceApproval", "response": {"Approved": true, "Comments": "Compliance approved."}}'
|
||||
```
|
||||
|
||||
PowerShell:
|
||||
|
||||
```powershell
|
||||
Invoke-RestMethod -Method Post `
|
||||
-Uri http://localhost:7071/api/workflows/ExpenseReimbursement/respond/{runId} `
|
||||
-ContentType application/json `
|
||||
-Body '{"eventName": "ComplianceApproval", "response": {"Approved": true, "Comments": "Compliance approved."}}'
|
||||
```
|
||||
|
||||
```json
|
||||
{
|
||||
"message": "Response sent to workflow.",
|
||||
"runId": "{runId}",
|
||||
"eventName": "ComplianceApproval",
|
||||
"validated": true
|
||||
}
|
||||
```
|
||||
|
||||
### Step 6: Check Final Status
|
||||
|
||||
After all approvals, the workflow completes and the expense is reimbursed:
|
||||
|
||||
Bash (Linux/macOS/WSL):
|
||||
|
||||
```bash
|
||||
curl http://localhost:7071/api/workflows/ExpenseReimbursement/status/{runId}
|
||||
```
|
||||
|
||||
PowerShell:
|
||||
|
||||
```powershell
|
||||
Invoke-RestMethod -Uri http://localhost:7071/api/workflows/ExpenseReimbursement/status/{runId}
|
||||
```
|
||||
|
||||
```json
|
||||
{
|
||||
"runId": "{runId}",
|
||||
"status": "Completed",
|
||||
"waitingForInput": null
|
||||
}
|
||||
```
|
||||
|
||||
### Viewing Workflows in the DTS Dashboard
|
||||
|
||||
After running a workflow, you can navigate to the Durable Task Scheduler (DTS) dashboard to visualize the orchestration and inspect its execution history.
|
||||
|
||||
If you are using the DTS emulator, the dashboard is available at `http://localhost:8082`.
|
||||
|
||||
1. Open the dashboard and look for the orchestration instance matching the `runId` returned in Step 1 (e.g., `abc123def456` or your custom ID like `expense-001`).
|
||||
2. Click into the instance to see the execution timeline, which shows each executor activity and the `WaitForExternalEvent` pauses where the workflow waited for human input — including the two parallel finance approvals.
|
||||
3. Expand individual activity steps to inspect inputs and outputs — for example, the `ManagerApproval`, `BudgetApproval`, and `ComplianceApproval` external events will show the approval request sent and the response received.
|
||||
@@ -0,0 +1,53 @@
|
||||
# Default endpoint address for local testing
|
||||
@authority=http://localhost:7071
|
||||
|
||||
### Step 1: Start the expense reimbursement workflow
|
||||
POST {{authority}}/api/workflows/ExpenseReimbursement/run
|
||||
Content-Type: text/plain
|
||||
|
||||
EXP-2025-001
|
||||
|
||||
### Step 1 (alternative): Start the workflow with a custom run ID
|
||||
POST {{authority}}/api/workflows/ExpenseReimbursement/run?runId=expense-001
|
||||
Content-Type: text/plain
|
||||
|
||||
EXP-2025-001
|
||||
|
||||
### Step 2: Check workflow status (replace {runId} with actual run ID from Step 1)
|
||||
GET {{authority}}/api/workflows/ExpenseReimbursement/status/{runId}
|
||||
|
||||
### Step 3: Send manager approval (replace {runId} with actual run ID from Step 1)
|
||||
POST {{authority}}/api/workflows/ExpenseReimbursement/respond/{runId}
|
||||
Content-Type: application/json
|
||||
|
||||
{"eventName": "ManagerApproval", "response": {"Approved": true, "Comments": "Approved by manager."}}
|
||||
|
||||
### Step 3 (alternative): Deny the expense at manager level
|
||||
POST {{authority}}/api/workflows/ExpenseReimbursement/respond/{runId}
|
||||
Content-Type: application/json
|
||||
|
||||
{"eventName": "ManagerApproval", "response": {"Approved": false, "Comments": "Insufficient documentation. Please resubmit."}}
|
||||
|
||||
### Step 4: Check workflow status after manager approval (now waiting for parallel finance approvals)
|
||||
GET {{authority}}/api/workflows/ExpenseReimbursement/status/{runId}
|
||||
|
||||
### Step 5a: Send budget approval (replace {runId} with actual run ID from Step 1)
|
||||
POST {{authority}}/api/workflows/ExpenseReimbursement/respond/{runId}
|
||||
Content-Type: application/json
|
||||
|
||||
{"eventName": "BudgetApproval", "response": {"Approved": true, "Comments": "Budget approved."}}
|
||||
|
||||
### Step 5b: Send compliance approval (replace {runId} with actual run ID from Step 1)
|
||||
POST {{authority}}/api/workflows/ExpenseReimbursement/respond/{runId}
|
||||
Content-Type: application/json
|
||||
|
||||
{"eventName": "ComplianceApproval", "response": {"Approved": true, "Comments": "Compliance approved."}}
|
||||
|
||||
### Step 5b (alternative): Deny the expense at compliance level
|
||||
POST {{authority}}/api/workflows/ExpenseReimbursement/respond/{runId}
|
||||
Content-Type: application/json
|
||||
|
||||
{"eventName": "ComplianceApproval", "response": {"Approved": false, "Comments": "Compliance requirements not met."}}
|
||||
|
||||
### Step 6: Check final workflow status after all approvals
|
||||
GET {{authority}}/api/workflows/ExpenseReimbursement/status/{runId}
|
||||
@@ -0,0 +1,20 @@
|
||||
{
|
||||
"version": "2.0",
|
||||
"logging": {
|
||||
"logLevel": {
|
||||
"Microsoft.Agents.AI.DurableTask": "Information",
|
||||
"Microsoft.Agents.AI.Hosting.AzureFunctions": "Information",
|
||||
"DurableTask": "Information",
|
||||
"Microsoft.DurableTask": "Information"
|
||||
}
|
||||
},
|
||||
"extensions": {
|
||||
"durableTask": {
|
||||
"hubName": "default",
|
||||
"storageProvider": {
|
||||
"type": "AzureManaged",
|
||||
"connectionStringName": "DURABLE_TASK_SCHEDULER_CONNECTION_STRING"
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,28 @@
|
||||
<Project Sdk="Microsoft.NET.Sdk">
|
||||
<PropertyGroup>
|
||||
<TargetFrameworks>net10.0</TargetFrameworks>
|
||||
<OutputType>Exe</OutputType>
|
||||
<ImplicitUsings>enable</ImplicitUsings>
|
||||
<Nullable>enable</Nullable>
|
||||
<AssemblyName>WorkflowHITL</AssemblyName>
|
||||
<RootNamespace>WorkflowHITL</RootNamespace>
|
||||
</PropertyGroup>
|
||||
|
||||
<ItemGroup>
|
||||
<PackageReference Include="Azure.Identity" />
|
||||
<PackageReference Include="Microsoft.DurableTask.Client.AzureManaged" />
|
||||
<PackageReference Include="Microsoft.DurableTask.Worker.AzureManaged" />
|
||||
<PackageReference Include="Microsoft.Extensions.Hosting" />
|
||||
</ItemGroup>
|
||||
|
||||
<!-- Local projects that should be switched to package references when using the sample outside of this MAF repo -->
|
||||
<!--
|
||||
<ItemGroup>
|
||||
<PackageReference Include="Microsoft.Agents.AI.DurableTask" />
|
||||
<PackageReference Include="Microsoft.Agents.AI.Workflows" />
|
||||
</ItemGroup>
|
||||
-->
|
||||
<ItemGroup>
|
||||
<ProjectReference Include="..\..\..\..\..\src\Microsoft.Agents.AI.DurableTask\Microsoft.Agents.AI.DurableTask.csproj" />
|
||||
</ItemGroup>
|
||||
</Project>
|
||||
@@ -0,0 +1,81 @@
|
||||
// Copyright (c) Microsoft. All rights reserved.
|
||||
|
||||
using Microsoft.Agents.AI.Workflows;
|
||||
|
||||
namespace WorkflowHITL;
|
||||
|
||||
/// <summary>
|
||||
/// Represents an expense approval request.
|
||||
/// </summary>
|
||||
/// <param name="ExpenseId">The unique identifier of the expense.</param>
|
||||
/// <param name="Amount">The amount of the expense.</param>
|
||||
/// <param name="EmployeeName">The name of the employee submitting the expense.</param>
|
||||
public record ApprovalRequest(string ExpenseId, decimal Amount, string EmployeeName);
|
||||
|
||||
/// <summary>
|
||||
/// Represents the response to an approval request.
|
||||
/// </summary>
|
||||
/// <param name="Approved">Whether the expense was approved.</param>
|
||||
/// <param name="Comments">Optional comments from the approver.</param>
|
||||
public record ApprovalResponse(bool Approved, string? Comments);
|
||||
|
||||
/// <summary>
|
||||
/// Retrieves expense details and creates an approval request.
|
||||
/// </summary>
|
||||
internal sealed class CreateApprovalRequest() : Executor<string, ApprovalRequest>("RetrieveRequest")
|
||||
{
|
||||
/// <inheritdoc/>
|
||||
public override ValueTask<ApprovalRequest> HandleAsync(
|
||||
string message,
|
||||
IWorkflowContext context,
|
||||
CancellationToken cancellationToken = default)
|
||||
{
|
||||
// In a real scenario, this would look up expense details from a database
|
||||
return new ValueTask<ApprovalRequest>(new ApprovalRequest(message, 1500.00m, "Jerry"));
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Prepares the approval request for finance review after manager approval.
|
||||
/// </summary>
|
||||
internal sealed class PrepareFinanceReview() : Executor<ApprovalResponse, ApprovalRequest>("PrepareFinanceReview")
|
||||
{
|
||||
/// <inheritdoc/>
|
||||
public override ValueTask<ApprovalRequest> HandleAsync(
|
||||
ApprovalResponse message,
|
||||
IWorkflowContext context,
|
||||
CancellationToken cancellationToken = default)
|
||||
{
|
||||
if (!message.Approved)
|
||||
{
|
||||
throw new InvalidOperationException("Cannot proceed to finance review — manager denied the expense.");
|
||||
}
|
||||
|
||||
// In a real scenario, this would retrieve the original expense details
|
||||
return new ValueTask<ApprovalRequest>(new ApprovalRequest("EXP-2025-001", 1500.00m, "Jerry"));
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Processes the expense reimbursement based on the parallel approval responses from budget and compliance.
|
||||
/// </summary>
|
||||
internal sealed class ExpenseReimburse() : Executor<ApprovalResponse[], string>("Reimburse")
|
||||
{
|
||||
/// <inheritdoc/>
|
||||
public override async ValueTask<string> HandleAsync(
|
||||
ApprovalResponse[] message,
|
||||
IWorkflowContext context,
|
||||
CancellationToken cancellationToken = default)
|
||||
{
|
||||
// Check that all parallel approvals passed
|
||||
ApprovalResponse? denied = Array.Find(message, r => !r.Approved);
|
||||
if (denied is not null)
|
||||
{
|
||||
return $"Expense reimbursement denied. Comments: {denied.Comments}";
|
||||
}
|
||||
|
||||
// Simulate payment processing
|
||||
await Task.Delay(1000, cancellationToken);
|
||||
return $"Expense reimbursed at {DateTime.UtcNow:O}";
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,98 @@
|
||||
// Copyright (c) Microsoft. All rights reserved.
|
||||
|
||||
// This sample demonstrates a Human-in-the-Loop (HITL) workflow using Durable Tasks.
|
||||
//
|
||||
// ┌──────────────────────┐ ┌────────────────┐ ┌─────────────────────┐ ┌────────────────────┐
|
||||
// │ CreateApprovalRequest│──►│ManagerApproval │──►│PrepareFinanceReview │──┬►│ BudgetApproval │──┐
|
||||
// └──────────────────────┘ │ (RequestPort) │ └─────────────────────┘ │ │ (RequestPort) │ │
|
||||
// └────────────────┘ │ └────────────────────┘ │ ┌─────────────────┐
|
||||
// │ ├─►│ExpenseReimburse │
|
||||
// │ ┌────────────────────┐ │ └─────────────────┘
|
||||
// └►│ComplianceApproval │──┘
|
||||
// │ (RequestPort) │
|
||||
// └────────────────────┘
|
||||
//
|
||||
// The workflow pauses at three RequestPorts — one for the manager, then two in parallel for finance.
|
||||
// After manager approval, BudgetApproval and ComplianceApproval run concurrently via fan-out/fan-in.
|
||||
|
||||
using Microsoft.Agents.AI.DurableTask;
|
||||
using Microsoft.Agents.AI.DurableTask.Workflows;
|
||||
using Microsoft.Agents.AI.Workflows;
|
||||
using Microsoft.DurableTask.Client.AzureManaged;
|
||||
using Microsoft.DurableTask.Worker.AzureManaged;
|
||||
using Microsoft.Extensions.DependencyInjection;
|
||||
using Microsoft.Extensions.Hosting;
|
||||
using Microsoft.Extensions.Logging;
|
||||
using WorkflowHITL;
|
||||
|
||||
string dtsConnectionString = Environment.GetEnvironmentVariable("DURABLE_TASK_SCHEDULER_CONNECTION_STRING")
|
||||
?? "Endpoint=http://localhost:8080;TaskHub=default;Authentication=None";
|
||||
|
||||
// Define executors and RequestPorts for the three HITL pause points
|
||||
CreateApprovalRequest createRequest = new();
|
||||
RequestPort<ApprovalRequest, ApprovalResponse> managerApproval = RequestPort.Create<ApprovalRequest, ApprovalResponse>("ManagerApproval");
|
||||
PrepareFinanceReview prepareFinanceReview = new();
|
||||
RequestPort<ApprovalRequest, ApprovalResponse> budgetApproval = RequestPort.Create<ApprovalRequest, ApprovalResponse>("BudgetApproval");
|
||||
RequestPort<ApprovalRequest, ApprovalResponse> complianceApproval = RequestPort.Create<ApprovalRequest, ApprovalResponse>("ComplianceApproval");
|
||||
ExpenseReimburse reimburse = new();
|
||||
|
||||
// Build the workflow: CreateApprovalRequest -> ManagerApproval -> PrepareFinanceReview -> [BudgetApproval AND ComplianceApproval] -> ExpenseReimburse
|
||||
Workflow expenseApproval = new WorkflowBuilder(createRequest)
|
||||
.WithName("ExpenseReimbursement")
|
||||
.WithDescription("Expense reimbursement with manager and parallel finance approvals")
|
||||
.AddEdge(createRequest, managerApproval)
|
||||
.AddEdge(managerApproval, prepareFinanceReview)
|
||||
.AddFanOutEdge(prepareFinanceReview, [budgetApproval, complianceApproval])
|
||||
.AddFanInEdge([budgetApproval, complianceApproval], reimburse)
|
||||
.Build();
|
||||
|
||||
IHost host = Host.CreateDefaultBuilder(args)
|
||||
.ConfigureLogging(logging => logging.SetMinimumLevel(LogLevel.Warning))
|
||||
.ConfigureServices(services =>
|
||||
{
|
||||
services.ConfigureDurableWorkflows(
|
||||
options => options.AddWorkflow(expenseApproval),
|
||||
workerBuilder: builder => builder.UseDurableTaskScheduler(dtsConnectionString),
|
||||
clientBuilder: builder => builder.UseDurableTaskScheduler(dtsConnectionString));
|
||||
})
|
||||
.Build();
|
||||
|
||||
await host.StartAsync();
|
||||
|
||||
IWorkflowClient workflowClient = host.Services.GetRequiredService<IWorkflowClient>();
|
||||
|
||||
// Start the workflow with streaming to observe events including HITL pauses
|
||||
string expenseId = "EXP-2025-001";
|
||||
Console.WriteLine($"Starting expense reimbursement workflow for expense: {expenseId}");
|
||||
IStreamingWorkflowRun run = await workflowClient.StreamAsync(expenseApproval, expenseId);
|
||||
Console.WriteLine($"Workflow started with instance ID: {run.RunId}\n");
|
||||
|
||||
// Watch for workflow events — handle HITL requests as they arrive
|
||||
await foreach (WorkflowEvent evt in run.WatchStreamAsync())
|
||||
{
|
||||
switch (evt)
|
||||
{
|
||||
case DurableWorkflowWaitingForInputEvent requestEvent:
|
||||
Console.WriteLine($"Workflow paused at RequestPort: {requestEvent.RequestPort.Id}");
|
||||
Console.WriteLine($" Input: {requestEvent.Input}");
|
||||
|
||||
// In a real scenario, this would involve human interaction (UI, email, Teams, etc.)
|
||||
ApprovalRequest? request = requestEvent.GetInputAs<ApprovalRequest>();
|
||||
Console.WriteLine($" Approval for: {request?.EmployeeName}, Amount: {request?.Amount:C}");
|
||||
|
||||
ApprovalResponse approvalResponse = new(Approved: true, Comments: "Approved by manager.");
|
||||
await run.SendResponseAsync(requestEvent, approvalResponse);
|
||||
Console.WriteLine($" Response sent: Approved={approvalResponse.Approved}\n");
|
||||
break;
|
||||
|
||||
case DurableWorkflowCompletedEvent completedEvent:
|
||||
Console.WriteLine($"Workflow completed: {completedEvent.Result}");
|
||||
break;
|
||||
|
||||
case DurableWorkflowFailedEvent failedEvent:
|
||||
Console.WriteLine($"Workflow failed: {failedEvent.ErrorMessage}");
|
||||
break;
|
||||
}
|
||||
}
|
||||
|
||||
await host.StopAsync();
|
||||
@@ -0,0 +1,106 @@
|
||||
# Workflow Human-in-the-Loop (HITL) Sample
|
||||
|
||||
This sample demonstrates a **Human-in-the-Loop** pattern in durable workflows using `RequestPort`. The workflow pauses execution at a manager approval point, then fans out to two parallel finance approval points — budget and compliance — before resuming.
|
||||
|
||||
## Key Concepts Demonstrated
|
||||
|
||||
- Using `RequestPort` to define external input points in a workflow
|
||||
- Sequential and parallel HITL pause points in a single workflow using fan-out/fan-in
|
||||
- Streaming workflow events with `IStreamingWorkflowRun`
|
||||
- Handling `DurableWorkflowWaitingForInputEvent` to detect HITL pauses
|
||||
- Using `SendResponseAsync` to provide responses and resume the workflow
|
||||
- **Durability**: The workflow survives process restarts while waiting for human input
|
||||
|
||||
## Workflow
|
||||
|
||||
This sample implements the following workflow:
|
||||
|
||||
```
|
||||
┌──────────────────────┐ ┌────────────────┐ ┌─────────────────────┐ ┌────────────────────┐
|
||||
│ CreateApprovalRequest│──►│ManagerApproval │──►│PrepareFinanceReview │──┬►│ BudgetApproval │──┐
|
||||
└──────────────────────┘ │ (RequestPort) │ └─────────────────────┘ │ │ (RequestPort) │ │
|
||||
└────────────────┘ │ └────────────────────┘ │ ┌─────────────────┐
|
||||
│ ├─►│ExpenseReimburse │
|
||||
│ ┌────────────────────┐ │ └─────────────────┘
|
||||
└►│ComplianceApproval │──┘
|
||||
│ (RequestPort) │
|
||||
└────────────────────┘
|
||||
```
|
||||
|
||||
| Step | Description |
|
||||
|------|-------------|
|
||||
| CreateApprovalRequest | Retrieves expense details and creates an approval request |
|
||||
| ManagerApproval (RequestPort) | **PAUSES** the workflow and waits for manager approval |
|
||||
| PrepareFinanceReview | Prepares the request for finance review after manager approval |
|
||||
| BudgetApproval (RequestPort) | **PAUSES** the workflow and waits for budget approval (parallel) |
|
||||
| ComplianceApproval (RequestPort) | **PAUSES** the workflow and waits for compliance approval (parallel) |
|
||||
| ExpenseReimburse | Processes the reimbursement after all approvals pass |
|
||||
|
||||
## How It Works
|
||||
|
||||
A `RequestPort` defines a typed external input point in the workflow:
|
||||
|
||||
```csharp
|
||||
RequestPort<ApprovalRequest, ApprovalResponse> managerApproval =
|
||||
RequestPort.Create<ApprovalRequest, ApprovalResponse>("ManagerApproval");
|
||||
```
|
||||
|
||||
Use `WatchStreamAsync` to observe events. When the workflow reaches a `RequestPort`, a `DurableWorkflowWaitingForInputEvent` is emitted. Call `SendResponseAsync` to provide the response and resume the workflow:
|
||||
|
||||
```csharp
|
||||
await foreach (WorkflowEvent evt in run.WatchStreamAsync())
|
||||
{
|
||||
switch (evt)
|
||||
{
|
||||
case DurableWorkflowWaitingForInputEvent requestEvent:
|
||||
ApprovalRequest? request = requestEvent.GetInputAs<ApprovalRequest>();
|
||||
await run.SendResponseAsync(requestEvent, new ApprovalResponse(Approved: true, Comments: "Approved."));
|
||||
break;
|
||||
}
|
||||
}
|
||||
```
|
||||
|
||||
## Environment Setup
|
||||
|
||||
See the [README.md](../README.md) file in the parent directory for information on configuring the environment, including how to install and run the Durable Task Scheduler.
|
||||
|
||||
## Running the Sample
|
||||
|
||||
```bash
|
||||
cd dotnet/samples/Durable/Workflow/ConsoleApps/08_WorkflowHITL
|
||||
dotnet run --framework net10.0
|
||||
```
|
||||
|
||||
### Sample Output
|
||||
|
||||
```text
|
||||
Starting expense reimbursement workflow for expense: EXP-2025-001
|
||||
Workflow started with instance ID: abc123...
|
||||
|
||||
Workflow paused at RequestPort: ManagerApproval
|
||||
Input: {"expenseId":"EXP-2025-001","amount":1500.00,"employeeName":"Jerry"}
|
||||
Approval for: Jerry, Amount: $1,500.00
|
||||
Response sent: Approved=True
|
||||
|
||||
Workflow paused at RequestPort: BudgetApproval
|
||||
Input: {"expenseId":"EXP-2025-001","amount":1500.00,"employeeName":"Jerry"}
|
||||
Approval for: Jerry, Amount: $1,500.00
|
||||
Response sent: Approved=True
|
||||
|
||||
Workflow paused at RequestPort: ComplianceApproval
|
||||
Input: {"expenseId":"EXP-2025-001","amount":1500.00,"employeeName":"Jerry"}
|
||||
Approval for: Jerry, Amount: $1,500.00
|
||||
Response sent: Approved=True
|
||||
|
||||
Workflow completed: Expense reimbursed at 2025-01-23T17:30:00.0000000Z
|
||||
```
|
||||
|
||||
### Viewing Workflows in the DTS Dashboard
|
||||
|
||||
After running the sample, you can navigate to the Durable Task Scheduler (DTS) dashboard to visualize the completed orchestration and inspect its execution history.
|
||||
|
||||
If you are using the DTS emulator, the dashboard is available at `http://localhost:8082`.
|
||||
|
||||
1. Open the dashboard and look for the orchestration instance matching the instance ID logged in the console output (e.g., `abc123...`).
|
||||
2. Click into the instance to see the execution timeline, which shows each executor activity and the `WaitForExternalEvent` pauses where the workflow waited for human input — including the two parallel finance approvals.
|
||||
3. Expand individual activity steps to inspect inputs and outputs — for example, the `ManagerApproval`, `BudgetApproval`, and `ComplianceApproval` external events will show the approval request sent and the response received.
|
||||
@@ -1,4 +1,4 @@
|
||||
// Copyright (c) Microsoft. All rights reserved.
|
||||
// Copyright (c) Microsoft. All rights reserved.
|
||||
|
||||
using System.Diagnostics;
|
||||
using Microsoft.Agents.AI.DurableTask.Workflows;
|
||||
@@ -9,7 +9,7 @@ namespace Microsoft.Agents.AI.DurableTask;
|
||||
/// Provides configuration options for durable agents and workflows.
|
||||
/// </summary>
|
||||
[DebuggerDisplay("Workflows = {Workflows.Workflows.Count}, Agents = {Agents.AgentCount}")]
|
||||
public sealed class DurableOptions
|
||||
public class DurableOptions
|
||||
{
|
||||
/// <summary>
|
||||
/// Initializes a new instance of the <see cref="DurableOptions"/> class.
|
||||
|
||||
@@ -1,4 +1,4 @@
|
||||
// Copyright (c) Microsoft. All rights reserved.
|
||||
// Copyright (c) Microsoft. All rights reserved.
|
||||
|
||||
using Microsoft.Agents.AI.DurableTask;
|
||||
using Microsoft.Extensions.AI;
|
||||
@@ -211,4 +211,20 @@ internal static partial class Logs
|
||||
this ILogger logger,
|
||||
string source,
|
||||
string sink);
|
||||
|
||||
[LoggerMessage(
|
||||
EventId = 112,
|
||||
Level = LogLevel.Information,
|
||||
Message = "Workflow waiting for external input at RequestPort '{RequestPortId}'")]
|
||||
public static partial void LogWaitingForExternalEvent(
|
||||
this ILogger logger,
|
||||
string requestPortId);
|
||||
|
||||
[LoggerMessage(
|
||||
EventId = 113,
|
||||
Level = LogLevel.Information,
|
||||
Message = "Received external event for RequestPort '{RequestPortId}'")]
|
||||
public static partial void LogReceivedExternalEvent(
|
||||
this ILogger logger,
|
||||
string requestPortId);
|
||||
}
|
||||
|
||||
@@ -1,4 +1,4 @@
|
||||
// Copyright (c) Microsoft. All rights reserved.
|
||||
// Copyright (c) Microsoft. All rights reserved.
|
||||
|
||||
using Microsoft.Agents.AI.DurableTask.Workflows;
|
||||
using Microsoft.Agents.AI.Workflows;
|
||||
@@ -234,11 +234,12 @@ public static class ServiceCollectionExtensions
|
||||
HashSet<string> registeredActivities = [];
|
||||
HashSet<string> registeredOrchestrations = [];
|
||||
|
||||
foreach (Workflow workflow in durableOptions.Workflows.Workflows.Values.ToList())
|
||||
DurableWorkflowOptions workflowOptions = durableOptions.Workflows;
|
||||
foreach (Workflow workflow in workflowOptions.Workflows.Values.ToList())
|
||||
{
|
||||
BuildWorkflowRegistrationRecursive(
|
||||
workflow,
|
||||
durableOptions.Workflows,
|
||||
workflowOptions,
|
||||
registrations,
|
||||
registeredActivities,
|
||||
registeredOrchestrations);
|
||||
@@ -329,12 +330,14 @@ public static class ServiceCollectionExtensions
|
||||
|
||||
/// <summary>
|
||||
/// Returns <see langword="true"/> for bindings that should be registered as Durable Task activities.
|
||||
/// <see cref="AIAgentBinding"/> (Durable Entities) and <see cref="SubworkflowBinding"/> (sub-orchestrations)
|
||||
/// use specialized dispatch and are excluded.
|
||||
/// <see cref="AIAgentBinding"/> (Durable Entities), <see cref="SubworkflowBinding"/> (sub-orchestrations),
|
||||
/// and <see cref="RequestPortBinding"/> (human-in-the-loop via external events) use specialized dispatch
|
||||
/// and are excluded.
|
||||
/// </summary>
|
||||
private static bool IsActivityBinding(ExecutorBinding binding)
|
||||
=> binding is not AIAgentBinding
|
||||
and not SubworkflowBinding;
|
||||
and not SubworkflowBinding
|
||||
and not RequestPortBinding;
|
||||
|
||||
private static async Task<DurableWorkflowResult> RunWorkflowOrchestrationAsync(
|
||||
TaskOrchestrationContext context,
|
||||
|
||||
@@ -111,18 +111,38 @@ internal static class DurableActivityExecutor
|
||||
}
|
||||
}
|
||||
|
||||
private static object DeserializeInput(string input, Type targetType)
|
||||
internal static object DeserializeInput(string input, Type targetType)
|
||||
{
|
||||
if (targetType == typeof(string))
|
||||
{
|
||||
return input;
|
||||
}
|
||||
|
||||
// Fan-in aggregation serializes results as a JSON array of strings (e.g., ["{...}", "{...}"]).
|
||||
// When the target type is a non-string array, deserialize each element individually.
|
||||
if (targetType.IsArray && targetType != typeof(string[]))
|
||||
{
|
||||
Type elementType = targetType.GetElementType()!;
|
||||
string[]? stringArray = JsonSerializer.Deserialize<string[]>(input, DurableSerialization.Options);
|
||||
if (stringArray is not null)
|
||||
{
|
||||
Array result = Array.CreateInstance(elementType, stringArray.Length);
|
||||
for (int i = 0; i < stringArray.Length; i++)
|
||||
{
|
||||
object element = JsonSerializer.Deserialize(stringArray[i], elementType, DurableSerialization.Options)
|
||||
?? throw new InvalidOperationException($"Failed to deserialize element {i} to type '{elementType.Name}'.");
|
||||
result.SetValue(element, i);
|
||||
}
|
||||
|
||||
return result;
|
||||
}
|
||||
}
|
||||
|
||||
return JsonSerializer.Deserialize(input, targetType, DurableSerialization.Options)
|
||||
?? throw new InvalidOperationException($"Failed to deserialize input to type '{targetType.Name}'.");
|
||||
}
|
||||
|
||||
private static Type ResolveInputType(string? inputTypeName, ISet<Type> supportedTypes)
|
||||
internal static Type ResolveInputType(string? inputTypeName, ISet<Type> supportedTypes)
|
||||
{
|
||||
if (string.IsNullOrEmpty(inputTypeName))
|
||||
{
|
||||
@@ -141,10 +161,13 @@ internal static class DurableActivityExecutor
|
||||
|
||||
Type? loadedType = Type.GetType(inputTypeName);
|
||||
|
||||
// Fall back if type is string but executor doesn't support string
|
||||
if (loadedType == typeof(string) && !supportedTypes.Contains(typeof(string)))
|
||||
// Fall back if type is string or string[] but executor doesn't support it
|
||||
if (loadedType is not null && !supportedTypes.Contains(loadedType))
|
||||
{
|
||||
return supportedTypes.FirstOrDefault() ?? typeof(string);
|
||||
if (loadedType == typeof(string) || loadedType == typeof(string[]))
|
||||
{
|
||||
return supportedTypes.FirstOrDefault() ?? typeof(string);
|
||||
}
|
||||
}
|
||||
|
||||
return loadedType ?? supportedTypes.FirstOrDefault() ?? typeof(string);
|
||||
|
||||
@@ -1,4 +1,4 @@
|
||||
// Copyright (c) Microsoft. All rights reserved.
|
||||
// Copyright (c) Microsoft. All rights reserved.
|
||||
|
||||
// ConfigureAwait Usage in Orchestration Code:
|
||||
// This file uses ConfigureAwait(true) because it runs within orchestration context.
|
||||
@@ -8,31 +8,34 @@
|
||||
// Using ConfigureAwait(false) here could cause non-deterministic behavior during replay.
|
||||
|
||||
using System.Text.Json;
|
||||
using Microsoft.Agents.AI.Workflows;
|
||||
using Microsoft.DurableTask;
|
||||
using Microsoft.Extensions.Logging;
|
||||
|
||||
namespace Microsoft.Agents.AI.DurableTask.Workflows;
|
||||
|
||||
/// <summary>
|
||||
/// Dispatches workflow executors to activities, AI agents, or sub-orchestrations.
|
||||
/// Dispatches workflow executors to activities, AI agents, sub-orchestrations, or external events (human-in-the-loop).
|
||||
/// </summary>
|
||||
/// <remarks>
|
||||
/// Called during the dispatch phase of each superstep by
|
||||
/// <c>DurableWorkflowRunner.DispatchExecutorsInParallelAsync</c>. For each executor that has
|
||||
/// pending input, this dispatcher determines whether the executor is an AI agent (stateful,
|
||||
/// backed by Durable Entities), a sub-workflow (dispatched as a sub-orchestration), or a
|
||||
/// regular activity, and invokes the appropriate Durable Task API.
|
||||
/// backed by Durable Entities), a request port (human-in-the-loop, backed by external events),
|
||||
/// a sub-workflow (dispatched as a sub-orchestration), or a regular activity, and invokes the
|
||||
/// appropriate Durable Task API.
|
||||
/// The serialised string result is returned to the runner for the routing phase.
|
||||
/// </remarks>
|
||||
internal static class DurableExecutorDispatcher
|
||||
{
|
||||
/// <summary>
|
||||
/// Dispatches an executor based on its type (activity, AI agent, or sub-workflow).
|
||||
/// Dispatches an executor based on its type (activity, AI agent, request port, or sub-workflow).
|
||||
/// </summary>
|
||||
/// <param name="context">The task orchestration context.</param>
|
||||
/// <param name="executorInfo">Information about the executor to dispatch.</param>
|
||||
/// <param name="envelope">The message envelope containing input and type information.</param>
|
||||
/// <param name="sharedState">The shared state dictionary to pass to the executor.</param>
|
||||
/// <param name="liveStatus">The live workflow status used to publish events and pending request port state.</param>
|
||||
/// <param name="logger">The logger for tracing.</param>
|
||||
/// <returns>The result from the executor.</returns>
|
||||
internal static async Task<string> DispatchAsync(
|
||||
@@ -40,10 +43,16 @@ internal static class DurableExecutorDispatcher
|
||||
WorkflowExecutorInfo executorInfo,
|
||||
DurableMessageEnvelope envelope,
|
||||
Dictionary<string, string> sharedState,
|
||||
DurableWorkflowLiveStatus liveStatus,
|
||||
ILogger logger)
|
||||
{
|
||||
logger.LogDispatchingExecutor(executorInfo.ExecutorId, executorInfo.IsAgenticExecutor);
|
||||
|
||||
if (executorInfo.IsRequestPortExecutor)
|
||||
{
|
||||
return await ExecuteRequestPortAsync(context, executorInfo, envelope.Message, liveStatus, logger).ConfigureAwait(true);
|
||||
}
|
||||
|
||||
if (executorInfo.IsAgenticExecutor)
|
||||
{
|
||||
return await ExecuteAgentAsync(context, executorInfo, logger, envelope.Message).ConfigureAwait(true);
|
||||
@@ -79,6 +88,47 @@ internal static class DurableExecutorDispatcher
|
||||
return await context.CallActivityAsync<string>(activityName, serializedInput).ConfigureAwait(true);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Executes a request port executor by waiting for an external event (human-in-the-loop).
|
||||
/// </summary>
|
||||
/// <remarks>
|
||||
/// When the workflow reaches a <see cref="RequestPort"/> executor, the orchestration publishes
|
||||
/// the pending request to <see cref="DurableWorkflowLiveStatus"/> and waits for an external actor
|
||||
/// (e.g., a UI or API) to raise the corresponding event via
|
||||
/// <see cref="IStreamingWorkflowRun.SendResponseAsync{TResponse}(DurableWorkflowWaitingForInputEvent, TResponse, CancellationToken)"/>.
|
||||
/// Multiple RequestPorts may be dispatched in parallel during a fan-out superstep.
|
||||
/// Each adds its pending request to <see cref="DurableWorkflowLiveStatus.PendingEvents"/>.
|
||||
/// The wait has no built-in timeout; for time-limited approvals, callers can combine
|
||||
/// <c>context.CreateTimer</c> with <c>Task.WhenAny</c> in a wrapper executor.
|
||||
/// </remarks>
|
||||
private static async Task<string> ExecuteRequestPortAsync(
|
||||
TaskOrchestrationContext context,
|
||||
WorkflowExecutorInfo executorInfo,
|
||||
string input,
|
||||
DurableWorkflowLiveStatus liveStatus,
|
||||
ILogger logger)
|
||||
{
|
||||
RequestPort requestPort = executorInfo.RequestPort!;
|
||||
string eventName = requestPort.Id;
|
||||
|
||||
logger.LogWaitingForExternalEvent(eventName);
|
||||
|
||||
// Publish pending request so external clients can discover what input is needed
|
||||
liveStatus.PendingEvents.Add(new PendingRequestPortStatus(EventName: eventName, Input: input));
|
||||
context.SetCustomStatus(liveStatus);
|
||||
|
||||
// Wait until the external actor raises the event
|
||||
string response = await context.WaitForExternalEvent<string>(eventName).ConfigureAwait(true);
|
||||
|
||||
// Remove this pending request after receiving the response
|
||||
liveStatus.PendingEvents.RemoveAll(p => p.EventName == eventName);
|
||||
context.SetCustomStatus(liveStatus.Events.Count > 0 || liveStatus.PendingEvents.Count > 0 ? liveStatus : null);
|
||||
|
||||
logger.LogReceivedExternalEvent(eventName);
|
||||
|
||||
return response;
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Executes an AI agent executor through Durable Entities.
|
||||
/// </summary>
|
||||
|
||||
+76
-22
@@ -14,15 +14,24 @@ namespace Microsoft.Agents.AI.DurableTask.Workflows;
|
||||
/// Represents a durable workflow run that supports streaming workflow events as they occur.
|
||||
/// </summary>
|
||||
/// <remarks>
|
||||
/// <para>
|
||||
/// Events are detected by monitoring the orchestration's custom status at regular intervals.
|
||||
/// When executors emit events via <see cref="IWorkflowContext.AddEventAsync"/> or
|
||||
/// <see cref="IWorkflowContext.YieldOutputAsync"/>, they are written to the orchestration's
|
||||
/// custom status and picked up by this streaming run.
|
||||
/// </para>
|
||||
/// <para>
|
||||
/// When the workflow reaches a <see cref="RequestPort"/> executor, a <see cref="DurableWorkflowWaitingForInputEvent"/>
|
||||
/// is yielded containing the request data. The caller should then call
|
||||
/// <see cref="SendResponseAsync{TResponse}(DurableWorkflowWaitingForInputEvent, TResponse, CancellationToken)"/>
|
||||
/// to provide the response and resume the workflow.
|
||||
/// </para>
|
||||
/// </remarks>
|
||||
[DebuggerDisplay("{WorkflowName} ({RunId})")]
|
||||
internal sealed class DurableStreamingWorkflowRun : IStreamingWorkflowRun
|
||||
{
|
||||
private readonly DurableTaskClient _client;
|
||||
private readonly Dictionary<string, RequestPort> _requestPorts;
|
||||
|
||||
/// <summary>
|
||||
/// Initializes a new instance of the <see cref="DurableStreamingWorkflowRun"/> class.
|
||||
@@ -35,6 +44,7 @@ internal sealed class DurableStreamingWorkflowRun : IStreamingWorkflowRun
|
||||
this._client = client;
|
||||
this.RunId = instanceId;
|
||||
this.WorkflowName = workflow.Name ?? string.Empty;
|
||||
this._requestPorts = ExtractRequestPorts(workflow);
|
||||
}
|
||||
|
||||
/// <inheritdoc/>
|
||||
@@ -92,9 +102,12 @@ internal sealed class DurableStreamingWorkflowRun : IStreamingWorkflowRun
|
||||
TimeSpan maxInterval = TimeSpan.FromSeconds(2);
|
||||
TimeSpan currentInterval = minInterval;
|
||||
|
||||
// Track how many events we've already read from custom status
|
||||
// Track how many events we've already read from the durable workflow status
|
||||
int lastReadEventIndex = 0;
|
||||
|
||||
// Track which pending events we've already yielded to avoid duplicates
|
||||
HashSet<string> yieldedPendingEvents = [];
|
||||
|
||||
while (!cancellationToken.IsCancellationRequested)
|
||||
{
|
||||
// Poll with getInputsAndOutputs: true because SerializedCustomStatus
|
||||
@@ -111,26 +124,54 @@ internal sealed class DurableStreamingWorkflowRun : IStreamingWorkflowRun
|
||||
|
||||
bool hasNewEvents = false;
|
||||
|
||||
// Always drain any unread events from custom status before checking terminal states.
|
||||
// Always drain any unread events from the durable workflow status before checking terminal states.
|
||||
// The orchestration may complete before the next poll, so events would be lost if we
|
||||
// check terminal status first.
|
||||
if (metadata.SerializedCustomStatus is not null)
|
||||
{
|
||||
if (TryParseCustomStatus(metadata.SerializedCustomStatus, out DurableWorkflowCustomStatus customStatus))
|
||||
if (DurableWorkflowLiveStatus.TryParse(metadata.SerializedCustomStatus, out DurableWorkflowLiveStatus liveStatus))
|
||||
{
|
||||
(List<WorkflowEvent> events, lastReadEventIndex) = DrainNewEvents(customStatus.Events, lastReadEventIndex);
|
||||
(List<WorkflowEvent> events, lastReadEventIndex) = DrainNewEvents(liveStatus.Events, lastReadEventIndex);
|
||||
foreach (WorkflowEvent evt in events)
|
||||
{
|
||||
hasNewEvents = true;
|
||||
yield return evt;
|
||||
}
|
||||
|
||||
// Yield a DurableWorkflowWaitingForInputEvent for each new pending request port
|
||||
foreach (PendingRequestPortStatus pending in liveStatus.PendingEvents)
|
||||
{
|
||||
if (yieldedPendingEvents.Add(pending.EventName))
|
||||
{
|
||||
if (!this._requestPorts.TryGetValue(pending.EventName, out RequestPort? matchingPort))
|
||||
{
|
||||
// RequestPort may not exist in the current workflow definition (e.g., during rolling deployments).
|
||||
continue;
|
||||
}
|
||||
|
||||
hasNewEvents = true;
|
||||
yield return new DurableWorkflowWaitingForInputEvent(
|
||||
pending.Input,
|
||||
matchingPort);
|
||||
}
|
||||
}
|
||||
|
||||
// Sync tracking with current pending events so re-used RequestPort names can be yielded again
|
||||
if (liveStatus.PendingEvents.Count == 0)
|
||||
{
|
||||
yieldedPendingEvents.Clear();
|
||||
}
|
||||
else
|
||||
{
|
||||
yieldedPendingEvents.IntersectWith(liveStatus.PendingEvents.Select(p => p.EventName));
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Check terminal states after draining events from custom status
|
||||
// Check terminal states after draining events from the durable workflow status
|
||||
if (metadata.RuntimeStatus == OrchestrationRuntimeStatus.Completed)
|
||||
{
|
||||
// The framework clears custom status on completion, so events may be in
|
||||
// The framework clears the durable workflow status on completion, so events may be in
|
||||
// SerializedOutput as a DurableWorkflowResult wrapper.
|
||||
if (TryParseWorkflowResult(metadata.SerializedOutput, out DurableWorkflowResult? outputResult))
|
||||
{
|
||||
@@ -183,6 +224,28 @@ internal sealed class DurableStreamingWorkflowRun : IStreamingWorkflowRun
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Sends a response to a <see cref="DurableWorkflowWaitingForInputEvent"/> to resume the workflow.
|
||||
/// </summary>
|
||||
/// <typeparam name="TResponse">The type of the response data.</typeparam>
|
||||
/// <param name="requestEvent">The request event to respond to.</param>
|
||||
/// <param name="response">The response data to send.</param>
|
||||
/// <param name="cancellationToken">A cancellation token to observe.</param>
|
||||
/// <returns>A <see cref="ValueTask"/> representing the asynchronous operation.</returns>
|
||||
[UnconditionalSuppressMessage("AOT", "IL3050", Justification = "Serializing workflow types provided by the caller.")]
|
||||
[UnconditionalSuppressMessage("Trimming", "IL2026", Justification = "Serializing workflow types provided by the caller.")]
|
||||
public async ValueTask SendResponseAsync<TResponse>(DurableWorkflowWaitingForInputEvent requestEvent, TResponse response, CancellationToken cancellationToken = default)
|
||||
{
|
||||
ArgumentNullException.ThrowIfNull(requestEvent);
|
||||
|
||||
string serializedResponse = JsonSerializer.Serialize(response, DurableSerialization.Options);
|
||||
await this._client.RaiseEventAsync(
|
||||
this.RunId,
|
||||
requestEvent.RequestPort.Id,
|
||||
serializedResponse,
|
||||
cancellationToken).ConfigureAwait(false);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Waits for the workflow to complete and returns the result.
|
||||
/// </summary>
|
||||
@@ -242,22 +305,6 @@ internal sealed class DurableStreamingWorkflowRun : IStreamingWorkflowRun
|
||||
return (events, lastReadIndex);
|
||||
}
|
||||
|
||||
[UnconditionalSuppressMessage("AOT", "IL3050", Justification = "Deserializing workflow custom status.")]
|
||||
[UnconditionalSuppressMessage("Trimming", "IL2026", Justification = "Deserializing workflow custom status.")]
|
||||
private static bool TryParseCustomStatus(string serializedStatus, out DurableWorkflowCustomStatus result)
|
||||
{
|
||||
try
|
||||
{
|
||||
result = JsonSerializer.Deserialize(serializedStatus, DurableWorkflowJsonContext.Default.DurableWorkflowCustomStatus)!;
|
||||
return result is not null;
|
||||
}
|
||||
catch (JsonException)
|
||||
{
|
||||
result = default!;
|
||||
return false;
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Attempts to parse the orchestration output as a <see cref="DurableWorkflowResult"/> wrapper.
|
||||
/// </summary>
|
||||
@@ -395,4 +442,11 @@ internal sealed class DurableStreamingWorkflowRun : IStreamingWorkflowRun
|
||||
|
||||
return dataElement.ValueKind == JsonValueKind.Null ? null : dataElement.Clone();
|
||||
}
|
||||
|
||||
private static Dictionary<string, RequestPort> ExtractRequestPorts(Workflow workflow)
|
||||
{
|
||||
return WorkflowAnalyzer.GetExecutorsFromWorkflowInOrder(workflow)
|
||||
.Where(e => e.RequestPort is not null)
|
||||
.ToDictionary(e => e.RequestPort!.Id, e => e.RequestPort!);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -1,22 +0,0 @@
|
||||
// Copyright (c) Microsoft. All rights reserved.
|
||||
|
||||
namespace Microsoft.Agents.AI.DurableTask.Workflows;
|
||||
|
||||
/// <summary>
|
||||
/// Represents the custom status written by the orchestration for streaming consumption.
|
||||
/// </summary>
|
||||
/// <remarks>
|
||||
/// The Durable Task framework exposes <c>SerializedCustomStatus</c> on orchestration metadata,
|
||||
/// which is the only orchestration state readable by external clients while the orchestration
|
||||
/// is still running. The orchestrator writes this object via <c>SetCustomStatus</c> after each
|
||||
/// superstep so that <see cref="DurableStreamingWorkflowRun"/> can poll for new events.
|
||||
/// On orchestration completion the framework clears custom status, so events are also
|
||||
/// embedded in the output via <see cref="DurableWorkflowResult"/>.
|
||||
/// </remarks>
|
||||
internal sealed class DurableWorkflowCustomStatus
|
||||
{
|
||||
/// <summary>
|
||||
/// Gets or sets the serialized workflow events emitted so far.
|
||||
/// </summary>
|
||||
public List<string> Events { get; set; } = [];
|
||||
}
|
||||
@@ -1,4 +1,4 @@
|
||||
// Copyright (c) Microsoft. All rights reserved.
|
||||
// Copyright (c) Microsoft. All rights reserved.
|
||||
|
||||
using System.Text.Json.Serialization;
|
||||
|
||||
@@ -16,7 +16,7 @@ namespace Microsoft.Agents.AI.DurableTask.Workflows;
|
||||
/// <item><description><see cref="DurableActivityInput"/>: Activity input wrapper with state</description></item>
|
||||
/// <item><description><see cref="DurableExecutorOutput"/>: Executor output wrapper with results, events, and state updates</description></item>
|
||||
/// <item><description><see cref="TypedPayload"/>: Serialized payload wrapper with type info (events and messages)</description></item>
|
||||
/// <item><description><see cref="DurableWorkflowCustomStatus"/>: Custom status for streaming consumption</description></item>
|
||||
/// <item><description><see cref="DurableWorkflowLiveStatus"/>: Live status payload (streaming events and pending request ports)</description></item>
|
||||
/// </list>
|
||||
/// <para>
|
||||
/// Note: User-defined executor input/output types still use reflection-based serialization
|
||||
@@ -31,8 +31,10 @@ namespace Microsoft.Agents.AI.DurableTask.Workflows;
|
||||
[JsonSerializable(typeof(DurableExecutorOutput))]
|
||||
[JsonSerializable(typeof(TypedPayload))]
|
||||
[JsonSerializable(typeof(List<TypedPayload>))]
|
||||
[JsonSerializable(typeof(DurableWorkflowCustomStatus))]
|
||||
[JsonSerializable(typeof(DurableWorkflowLiveStatus))]
|
||||
[JsonSerializable(typeof(DurableWorkflowResult))]
|
||||
[JsonSerializable(typeof(PendingRequestPortStatus))]
|
||||
[JsonSerializable(typeof(List<PendingRequestPortStatus>))]
|
||||
[JsonSerializable(typeof(List<string>))]
|
||||
[JsonSerializable(typeof(Dictionary<string, string>))]
|
||||
[JsonSerializable(typeof(Dictionary<string, string?>))]
|
||||
|
||||
@@ -0,0 +1,59 @@
|
||||
// Copyright (c) Microsoft. All rights reserved.
|
||||
|
||||
using Microsoft.Agents.AI.Workflows;
|
||||
|
||||
namespace Microsoft.Agents.AI.DurableTask.Workflows;
|
||||
|
||||
/// <summary>
|
||||
/// Live status payload written to the orchestration via <c>SetCustomStatus</c>.
|
||||
/// </summary>
|
||||
/// <remarks>
|
||||
/// <para>
|
||||
/// This is the only orchestration state readable by external clients while the workflow
|
||||
/// is still running. It is written after each superstep so that
|
||||
/// <see cref="DurableStreamingWorkflowRun"/> can poll for new events.
|
||||
/// On completion the framework clears it, so events are also
|
||||
/// embedded in the output via <see cref="DurableWorkflowResult"/>.
|
||||
/// </para>
|
||||
/// <para>
|
||||
/// When the workflow is paused at one or more <see cref="RequestPort"/> nodes,
|
||||
/// <see cref="PendingEvents"/> contains the request data for each.
|
||||
/// </para>
|
||||
/// </remarks>
|
||||
internal sealed class DurableWorkflowLiveStatus
|
||||
{
|
||||
/// <summary>
|
||||
/// Gets or sets the pending request ports the workflow is waiting on. Empty when no input is needed.
|
||||
/// </summary>
|
||||
public List<PendingRequestPortStatus> PendingEvents { get; set; } = [];
|
||||
|
||||
/// <summary>
|
||||
/// Gets or sets the serialized workflow events emitted so far.
|
||||
/// </summary>
|
||||
public List<string> Events { get; set; } = [];
|
||||
|
||||
/// <summary>
|
||||
/// Attempts to deserialize a serialized custom status string into a <see cref="DurableWorkflowLiveStatus"/>.
|
||||
/// </summary>
|
||||
[System.Diagnostics.CodeAnalysis.UnconditionalSuppressMessage("AOT", "IL3050", Justification = "Deserializing durable workflow status.")]
|
||||
[System.Diagnostics.CodeAnalysis.UnconditionalSuppressMessage("Trimming", "IL2026", Justification = "Deserializing durable workflow status.")]
|
||||
internal static bool TryParse(string? serializedStatus, out DurableWorkflowLiveStatus result)
|
||||
{
|
||||
if (serializedStatus is null)
|
||||
{
|
||||
result = default!;
|
||||
return false;
|
||||
}
|
||||
|
||||
try
|
||||
{
|
||||
result = System.Text.Json.JsonSerializer.Deserialize<DurableWorkflowLiveStatus>(serializedStatus, DurableSerialization.Options)!;
|
||||
return result is not null;
|
||||
}
|
||||
catch (System.Text.Json.JsonException)
|
||||
{
|
||||
result = default!;
|
||||
return false;
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -1,4 +1,4 @@
|
||||
// Copyright (c) Microsoft. All rights reserved.
|
||||
// Copyright (c) Microsoft. All rights reserved.
|
||||
|
||||
using System.Diagnostics;
|
||||
using Microsoft.Agents.AI.Workflows;
|
||||
@@ -12,7 +12,6 @@ namespace Microsoft.Agents.AI.DurableTask.Workflows;
|
||||
public sealed class DurableWorkflowOptions
|
||||
{
|
||||
private readonly Dictionary<string, Workflow> _workflows = new(StringComparer.OrdinalIgnoreCase);
|
||||
private readonly DurableOptions? _parentOptions;
|
||||
|
||||
/// <summary>
|
||||
/// Initializes a new instance of the <see cref="DurableWorkflowOptions"/> class.
|
||||
@@ -20,9 +19,14 @@ public sealed class DurableWorkflowOptions
|
||||
/// <param name="parentOptions">Optional parent options container for accessing related configuration.</param>
|
||||
internal DurableWorkflowOptions(DurableOptions? parentOptions = null)
|
||||
{
|
||||
this._parentOptions = parentOptions;
|
||||
this.ParentOptions = parentOptions;
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Gets the parent <see cref="DurableOptions"/> container, if available.
|
||||
/// </summary>
|
||||
internal DurableOptions? ParentOptions { get; }
|
||||
|
||||
/// <summary>
|
||||
/// Gets the collection of workflows available in the current context, keyed by their unique names.
|
||||
/// </summary>
|
||||
@@ -77,7 +81,7 @@ public sealed class DurableWorkflowOptions
|
||||
/// </summary>
|
||||
private void RegisterWorkflowExecutors(Workflow workflow)
|
||||
{
|
||||
DurableAgentsOptions? agentOptions = this._parentOptions?.Agents;
|
||||
DurableAgentsOptions? agentOptions = this.ParentOptions?.Agents;
|
||||
|
||||
foreach ((string executorId, ExecutorBinding binding) in workflow.ReflectExecutors())
|
||||
{
|
||||
|
||||
@@ -1,4 +1,4 @@
|
||||
// Copyright (c) Microsoft. All rights reserved.
|
||||
// Copyright (c) Microsoft. All rights reserved.
|
||||
|
||||
// ConfigureAwait Usage in Orchestration Code:
|
||||
// This file uses ConfigureAwait(true) because it runs within orchestration context.
|
||||
@@ -173,7 +173,7 @@ internal sealed class DurableWorkflowRunner
|
||||
logger.LogSuperstepExecutors(superstep, string.Join(", ", executorInputs.Select(e => e.ExecutorId)));
|
||||
}
|
||||
|
||||
string[] results = await DispatchExecutorsInParallelAsync(context, executorInputs, state.SharedState, logger).ConfigureAwait(true);
|
||||
string[] results = await DispatchExecutorsInParallelAsync(context, executorInputs, state, logger).ConfigureAwait(true);
|
||||
|
||||
haltRequested = ProcessSuperstepResults(executorInputs, results, state, context, logger);
|
||||
|
||||
@@ -193,7 +193,7 @@ internal sealed class DurableWorkflowRunner
|
||||
// Publish final events for live streaming (skip during replay)
|
||||
if (!context.IsReplaying)
|
||||
{
|
||||
PublishEventsToCustomStatus(context, state);
|
||||
PublishEventsToLiveStatus(context, state);
|
||||
}
|
||||
|
||||
string finalResult = GetFinalResult(state.LastResults);
|
||||
@@ -226,11 +226,11 @@ internal sealed class DurableWorkflowRunner
|
||||
private static async Task<string[]> DispatchExecutorsInParallelAsync(
|
||||
TaskOrchestrationContext context,
|
||||
List<ExecutorInput> executorInputs,
|
||||
Dictionary<string, string> sharedState,
|
||||
SuperstepState state,
|
||||
ILogger logger)
|
||||
{
|
||||
Task<string>[] dispatchTasks = executorInputs
|
||||
.Select(input => DurableExecutorDispatcher.DispatchAsync(context, input.Info, input.Envelope, sharedState, logger))
|
||||
.Select(input => DurableExecutorDispatcher.DispatchAsync(context, input.Info, input.Envelope, state.SharedState, state.LiveStatus, logger))
|
||||
.ToArray();
|
||||
|
||||
return await Task.WhenAll(dispatchTasks).ConfigureAwait(true);
|
||||
@@ -273,9 +273,14 @@ internal sealed class DurableWorkflowRunner
|
||||
public Dictionary<string, string> SharedState { get; } = [];
|
||||
|
||||
/// <summary>
|
||||
/// Accumulated workflow events for custom status (streaming consumption).
|
||||
/// Accumulated workflow events for the durable workflow status (streaming consumption).
|
||||
/// </summary>
|
||||
public List<string> AccumulatedEvents { get; } = [];
|
||||
|
||||
/// <summary>
|
||||
/// Workflow status published via <c>SetCustomStatus</c> so external clients can poll for streaming events and pending HITL requests.
|
||||
/// </summary>
|
||||
public DurableWorkflowLiveStatus LiveStatus { get; } = new();
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
@@ -378,7 +383,7 @@ internal sealed class DurableWorkflowRunner
|
||||
// Merge state updates from activity into shared state
|
||||
MergeStateUpdates(state, resultInfo.StateUpdates, resultInfo.ClearedScopes);
|
||||
|
||||
// Accumulate events for custom status (streaming)
|
||||
// Accumulate events for the durable workflow status (streaming)
|
||||
state.AccumulatedEvents.AddRange(resultInfo.Events);
|
||||
|
||||
// Check for halt request
|
||||
@@ -387,7 +392,7 @@ internal sealed class DurableWorkflowRunner
|
||||
// Publish events for live streaming (skip during replay)
|
||||
if (!context.IsReplaying)
|
||||
{
|
||||
PublishEventsToCustomStatus(context, state);
|
||||
PublishEventsToLiveStatus(context, state);
|
||||
}
|
||||
|
||||
RouteOutputToSuccessors(executorId, resultInfo.Result, resultInfo.SentMessages, state, logger);
|
||||
@@ -464,24 +469,23 @@ internal sealed class DurableWorkflowRunner
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Publishes accumulated workflow events to the orchestration's custom status,
|
||||
/// Publishes accumulated workflow events to the durable workflow's custom status,
|
||||
/// making them available to <see cref="DurableStreamingWorkflowRun"/> for live streaming.
|
||||
/// </summary>
|
||||
/// <remarks>
|
||||
/// Custom status is the only orchestration metadata readable by external clients while
|
||||
/// Custom status is the only orchestration state readable by external clients while
|
||||
/// the orchestration is still running. It is cleared by the framework on completion,
|
||||
/// so events are also included in <see cref="DurableWorkflowResult"/> for final retrieval.
|
||||
/// </remarks>
|
||||
private static void PublishEventsToCustomStatus(TaskOrchestrationContext context, SuperstepState state)
|
||||
private static void PublishEventsToLiveStatus(
|
||||
TaskOrchestrationContext context,
|
||||
SuperstepState state)
|
||||
{
|
||||
DurableWorkflowCustomStatus customStatus = new()
|
||||
{
|
||||
Events = state.AccumulatedEvents
|
||||
};
|
||||
state.LiveStatus.Events = state.AccumulatedEvents;
|
||||
|
||||
// Pass the object directly — the framework's DataConverter handles serialization.
|
||||
// Pre-serializing would cause double-serialization (string wrapped in JSON quotes).
|
||||
context.SetCustomStatus(customStatus);
|
||||
context.SetCustomStatus(state.LiveStatus);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
|
||||
+42
@@ -0,0 +1,42 @@
|
||||
// Copyright (c) Microsoft. All rights reserved.
|
||||
|
||||
using System.Diagnostics;
|
||||
using System.Diagnostics.CodeAnalysis;
|
||||
using System.Text.Json;
|
||||
using Microsoft.Agents.AI.Workflows;
|
||||
|
||||
namespace Microsoft.Agents.AI.DurableTask.Workflows;
|
||||
|
||||
/// <summary>
|
||||
/// Event raised when the durable workflow is waiting for external input at a <see cref="RequestPort"/>.
|
||||
/// </summary>
|
||||
/// <param name="Input">The serialized input data that was passed to the RequestPort.</param>
|
||||
/// <param name="RequestPort">The request port definition.</param>
|
||||
[DebuggerDisplay("RequestPort = {RequestPort.Id}")]
|
||||
public sealed class DurableWorkflowWaitingForInputEvent(
|
||||
string Input,
|
||||
RequestPort RequestPort) : WorkflowEvent
|
||||
{
|
||||
/// <summary>
|
||||
/// Gets the serialized input data that was passed to the RequestPort.
|
||||
/// </summary>
|
||||
public string Input { get; } = Input;
|
||||
|
||||
/// <summary>
|
||||
/// Gets the request port definition.
|
||||
/// </summary>
|
||||
public RequestPort RequestPort { get; } = RequestPort;
|
||||
|
||||
/// <summary>
|
||||
/// Attempts to deserialize the input data to the specified type.
|
||||
/// </summary>
|
||||
/// <typeparam name="T">The type to deserialize to.</typeparam>
|
||||
/// <returns>The deserialized input.</returns>
|
||||
/// <exception cref="JsonException">Thrown when the input cannot be deserialized to the specified type.</exception>
|
||||
[UnconditionalSuppressMessage("AOT", "IL3050", Justification = "Deserializing workflow types provided by the caller.")]
|
||||
[UnconditionalSuppressMessage("Trimming", "IL2026", Justification = "Deserializing workflow types provided by the caller.")]
|
||||
public T? GetInputAs<T>()
|
||||
{
|
||||
return JsonSerializer.Deserialize<T>(this.Input, DurableSerialization.Options);
|
||||
}
|
||||
}
|
||||
@@ -1,4 +1,4 @@
|
||||
// Copyright (c) Microsoft. All rights reserved.
|
||||
// Copyright (c) Microsoft. All rights reserved.
|
||||
|
||||
using Microsoft.Agents.AI.Workflows;
|
||||
|
||||
@@ -39,4 +39,17 @@ public interface IStreamingWorkflowRun
|
||||
/// workflow state changes.
|
||||
/// </returns>
|
||||
IAsyncEnumerable<WorkflowEvent> WatchStreamAsync(CancellationToken cancellationToken = default);
|
||||
|
||||
/// <summary>
|
||||
/// Sends a response to a <see cref="DurableWorkflowWaitingForInputEvent"/> to resume the workflow.
|
||||
/// </summary>
|
||||
/// <typeparam name="TResponse">The type of the response data.</typeparam>
|
||||
/// <param name="requestEvent">The request event to respond to.</param>
|
||||
/// <param name="response">The response data to send.</param>
|
||||
/// <param name="cancellationToken">A cancellation token to observe.</param>
|
||||
/// <returns>A <see cref="ValueTask"/> representing the asynchronous operation.</returns>
|
||||
ValueTask SendResponseAsync<TResponse>(
|
||||
DurableWorkflowWaitingForInputEvent requestEvent,
|
||||
TResponse response,
|
||||
CancellationToken cancellationToken = default);
|
||||
}
|
||||
|
||||
@@ -0,0 +1,12 @@
|
||||
// Copyright (c) Microsoft. All rights reserved.
|
||||
|
||||
namespace Microsoft.Agents.AI.DurableTask.Workflows;
|
||||
|
||||
/// <summary>
|
||||
/// Represents a RequestPort the workflow is paused at, waiting for a response.
|
||||
/// </summary>
|
||||
/// <param name="EventName">The RequestPort ID identifying which input is needed.</param>
|
||||
/// <param name="Input">The serialized request data passed to the RequestPort.</param>
|
||||
internal sealed record PendingRequestPortStatus(
|
||||
string EventName,
|
||||
string Input);
|
||||
@@ -1,4 +1,4 @@
|
||||
// Copyright (c) Microsoft. All rights reserved.
|
||||
// Copyright (c) Microsoft. All rights reserved.
|
||||
|
||||
using Microsoft.Azure.Functions.Worker;
|
||||
using Microsoft.Azure.Functions.Worker.Context.Features;
|
||||
@@ -85,6 +85,34 @@ internal sealed class BuiltInFunctionExecutor : IFunctionExecutor
|
||||
return;
|
||||
}
|
||||
|
||||
if (context.FunctionDefinition.EntryPoint == BuiltInFunctions.GetWorkflowStatusHttpFunctionEntryPoint)
|
||||
{
|
||||
if (httpRequestData == null)
|
||||
{
|
||||
throw new InvalidOperationException($"HTTP request data binding is missing for the invocation {context.InvocationId}.");
|
||||
}
|
||||
|
||||
context.GetInvocationResult().Value = await BuiltInFunctions.GetWorkflowStatusAsync(
|
||||
httpRequestData,
|
||||
durableTaskClient,
|
||||
context);
|
||||
return;
|
||||
}
|
||||
|
||||
if (context.FunctionDefinition.EntryPoint == BuiltInFunctions.RespondToWorkflowHttpFunctionEntryPoint)
|
||||
{
|
||||
if (httpRequestData == null)
|
||||
{
|
||||
throw new InvalidOperationException($"HTTP request data binding is missing for the invocation {context.InvocationId}.");
|
||||
}
|
||||
|
||||
context.GetInvocationResult().Value = await BuiltInFunctions.RespondToWorkflowAsync(
|
||||
httpRequestData,
|
||||
durableTaskClient,
|
||||
context);
|
||||
return;
|
||||
}
|
||||
|
||||
if (context.FunctionDefinition.EntryPoint == BuiltInFunctions.InvokeWorkflowActivityFunctionEntryPoint)
|
||||
{
|
||||
if (encodedEntityRequest is null)
|
||||
|
||||
@@ -1,6 +1,7 @@
|
||||
// Copyright (c) Microsoft. All rights reserved.
|
||||
|
||||
using System.Net;
|
||||
using System.Text.Json;
|
||||
using System.Text.Json.Serialization;
|
||||
using Microsoft.Agents.AI.DurableTask;
|
||||
using Microsoft.Agents.AI.DurableTask.Workflows;
|
||||
@@ -26,6 +27,8 @@ internal static class BuiltInFunctions
|
||||
internal static readonly string RunWorkflowOrchestrationHttpFunctionEntryPoint = $"{typeof(BuiltInFunctions).FullName!}.{nameof(RunWorkflowOrchestrationHttpTriggerAsync)}";
|
||||
internal static readonly string RunWorkflowOrchestrationFunctionEntryPoint = $"{typeof(BuiltInFunctions).FullName!}.{nameof(RunWorkflowOrchestration)}";
|
||||
internal static readonly string InvokeWorkflowActivityFunctionEntryPoint = $"{typeof(BuiltInFunctions).FullName!}.{nameof(InvokeWorkflowActivityAsync)}";
|
||||
internal static readonly string GetWorkflowStatusHttpFunctionEntryPoint = $"{typeof(BuiltInFunctions).FullName!}.{nameof(GetWorkflowStatusAsync)}";
|
||||
internal static readonly string RespondToWorkflowHttpFunctionEntryPoint = $"{typeof(BuiltInFunctions).FullName!}.{nameof(RespondToWorkflowAsync)}";
|
||||
|
||||
#pragma warning disable IL3000 // Avoid accessing Assembly file path when publishing as a single file - Azure Functions does not use single-file publishing
|
||||
internal static readonly string ScriptFile = Path.GetFileName(typeof(BuiltInFunctions).Assembly.Location);
|
||||
@@ -63,6 +66,122 @@ internal static class BuiltInFunctions
|
||||
return response;
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Returns the workflow status including any pending HITL requests.
|
||||
/// The run ID is extracted from the route parameter <c>{runId}</c>.
|
||||
/// </summary>
|
||||
public static async Task<HttpResponseData> GetWorkflowStatusAsync(
|
||||
[HttpTrigger] HttpRequestData req,
|
||||
[DurableClient] DurableTaskClient client,
|
||||
FunctionContext context)
|
||||
{
|
||||
string? runId = context.BindingContext.BindingData.TryGetValue("runId", out object? value) ? value?.ToString() : null;
|
||||
if (string.IsNullOrEmpty(runId))
|
||||
{
|
||||
return await CreateErrorResponseAsync(req, context, HttpStatusCode.BadRequest, "Run ID is required.");
|
||||
}
|
||||
|
||||
OrchestrationMetadata? metadata = await client.GetInstanceAsync(runId, getInputsAndOutputs: true);
|
||||
if (metadata is null)
|
||||
{
|
||||
return await CreateErrorResponseAsync(req, context, HttpStatusCode.NotFound, $"Workflow run '{runId}' not found.");
|
||||
}
|
||||
|
||||
// Parse HITL inputs the workflow is waiting for from the durable workflow status
|
||||
List<PendingRequestPortStatus>? waitingForInput = null;
|
||||
if (DurableWorkflowLiveStatus.TryParse(metadata.SerializedCustomStatus, out DurableWorkflowLiveStatus liveStatus)
|
||||
&& liveStatus.PendingEvents.Count > 0)
|
||||
{
|
||||
waitingForInput = liveStatus.PendingEvents;
|
||||
}
|
||||
|
||||
HttpResponseData response = req.CreateResponse(HttpStatusCode.OK);
|
||||
await response.WriteAsJsonAsync(new
|
||||
{
|
||||
runId,
|
||||
status = metadata.RuntimeStatus.ToString(),
|
||||
waitingForInput = waitingForInput?.Select(p => new { eventName = p.EventName, input = JsonDocument.Parse(p.Input).RootElement })
|
||||
});
|
||||
return response;
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Sends a response to a pending RequestPort, resuming the workflow.
|
||||
/// Expects a JSON body: <c>{ "eventName": "...", "response": { ... } }</c>.
|
||||
/// </summary>
|
||||
public static async Task<HttpResponseData> RespondToWorkflowAsync(
|
||||
[HttpTrigger] HttpRequestData req,
|
||||
[DurableClient] DurableTaskClient client,
|
||||
FunctionContext context)
|
||||
{
|
||||
string? runId = context.BindingContext.BindingData.TryGetValue("runId", out object? value) ? value?.ToString() : null;
|
||||
if (string.IsNullOrEmpty(runId))
|
||||
{
|
||||
return await CreateErrorResponseAsync(req, context, HttpStatusCode.BadRequest, "Run ID is required.");
|
||||
}
|
||||
|
||||
WorkflowRespondRequest? request;
|
||||
try
|
||||
{
|
||||
request = await req.ReadFromJsonAsync<WorkflowRespondRequest>(context.CancellationToken);
|
||||
}
|
||||
catch (JsonException)
|
||||
{
|
||||
return await CreateErrorResponseAsync(req, context, HttpStatusCode.BadRequest, "Request body is not valid JSON.");
|
||||
}
|
||||
|
||||
if (request is null || string.IsNullOrEmpty(request.EventName)
|
||||
|| request.Response.ValueKind == JsonValueKind.Undefined)
|
||||
{
|
||||
return await CreateErrorResponseAsync(req, context, HttpStatusCode.BadRequest, "Body must contain a non-empty 'eventName' and a 'response' property.");
|
||||
}
|
||||
|
||||
// Verify the orchestration exists and is in a valid state
|
||||
OrchestrationMetadata? metadata = await client.GetInstanceAsync(runId, getInputsAndOutputs: true);
|
||||
if (metadata is null)
|
||||
{
|
||||
return await CreateErrorResponseAsync(req, context, HttpStatusCode.NotFound, $"Workflow run '{runId}' not found.");
|
||||
}
|
||||
|
||||
if (metadata.RuntimeStatus is OrchestrationRuntimeStatus.Completed
|
||||
or OrchestrationRuntimeStatus.Failed
|
||||
or OrchestrationRuntimeStatus.Terminated)
|
||||
{
|
||||
return await CreateErrorResponseAsync(req, context, HttpStatusCode.BadRequest,
|
||||
$"Workflow run '{runId}' is in terminal state '{metadata.RuntimeStatus}'.");
|
||||
}
|
||||
|
||||
// Verify the workflow is waiting for the specified event.
|
||||
// If status can't be parsed (e.g., not yet set during early execution), allow the event through —
|
||||
// Durable Task safely queues it until the orchestration reaches WaitForExternalEvent.
|
||||
bool eventValidated = false;
|
||||
if (DurableWorkflowLiveStatus.TryParse(metadata.SerializedCustomStatus, out DurableWorkflowLiveStatus liveStatus))
|
||||
{
|
||||
if (!liveStatus.PendingEvents.Exists(p => string.Equals(p.EventName, request.EventName, StringComparison.Ordinal)))
|
||||
{
|
||||
return await CreateErrorResponseAsync(req, context, HttpStatusCode.BadRequest,
|
||||
$"Workflow is not waiting for event '{request.EventName}'.");
|
||||
}
|
||||
|
||||
eventValidated = true;
|
||||
}
|
||||
|
||||
// Raise the external event to unblock the orchestration's WaitForExternalEvent call
|
||||
await client.RaiseEventAsync(runId, request.EventName, request.Response.GetRawText());
|
||||
|
||||
HttpResponseData response = req.CreateResponse(HttpStatusCode.Accepted);
|
||||
await response.WriteAsJsonAsync(new
|
||||
{
|
||||
message = eventValidated
|
||||
? "Response sent to workflow."
|
||||
: "Response sent to workflow. Event could not be validated against pending requests.",
|
||||
runId,
|
||||
eventName = request.EventName,
|
||||
validated = eventValidated,
|
||||
});
|
||||
return response;
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Executes a workflow activity by looking up the registered executor and delegating to it.
|
||||
/// The executor name is derived from the activity function name via <see cref="WorkflowNamingHelper"/>.
|
||||
@@ -413,6 +532,15 @@ internal static class BuiltInFunctions
|
||||
[property: JsonPropertyName("status")] int Status,
|
||||
[property: JsonPropertyName("thread_id")] string ThreadId);
|
||||
|
||||
/// <summary>
|
||||
/// Represents a request to respond to a pending RequestPort in a workflow.
|
||||
/// </summary>
|
||||
/// <param name="EventName">The name of the event to raise (the RequestPort ID).</param>
|
||||
/// <param name="Response">The response payload to send to the workflow.</param>
|
||||
private sealed record WorkflowRespondRequest(
|
||||
[property: JsonPropertyName("eventName")] string? EventName,
|
||||
[property: JsonPropertyName("response")] JsonElement Response);
|
||||
|
||||
/// <summary>
|
||||
/// A service provider that combines the original service provider with an additional DurableTaskClient instance.
|
||||
/// </summary>
|
||||
|
||||
@@ -1,4 +1,4 @@
|
||||
// Copyright (c) Microsoft. All rights reserved.
|
||||
// Copyright (c) Microsoft. All rights reserved.
|
||||
|
||||
using Microsoft.Agents.AI.DurableTask;
|
||||
using Microsoft.Azure.Functions.Worker.Core.FunctionMetadata;
|
||||
@@ -38,8 +38,9 @@ internal static class FunctionMetadataFactory
|
||||
/// <param name="name">The base name used to derive the HTTP function name.</param>
|
||||
/// <param name="route">The HTTP route for the trigger.</param>
|
||||
/// <param name="entryPoint">The entry point method for the HTTP trigger.</param>
|
||||
/// <param name="methods">The allowed HTTP methods as a JSON array fragment (e.g., <c>"\"get\""</c>). Defaults to POST.</param>
|
||||
/// <returns>A <see cref="DefaultFunctionMetadata"/> configured for an HTTP trigger.</returns>
|
||||
internal static DefaultFunctionMetadata CreateHttpTrigger(string name, string route, string entryPoint)
|
||||
internal static DefaultFunctionMetadata CreateHttpTrigger(string name, string route, string entryPoint, string methods = "\"post\"")
|
||||
{
|
||||
return new DefaultFunctionMetadata()
|
||||
{
|
||||
@@ -47,7 +48,7 @@ internal static class FunctionMetadataFactory
|
||||
Language = "dotnet-isolated",
|
||||
RawBindings =
|
||||
[
|
||||
$"{{\"name\":\"req\",\"type\":\"httpTrigger\",\"direction\":\"In\",\"authLevel\":\"function\",\"methods\": [\"post\"],\"route\":\"{route}\"}}",
|
||||
$"{{\"name\":\"req\",\"type\":\"httpTrigger\",\"direction\":\"In\",\"authLevel\":\"function\",\"methods\": [{methods}],\"route\":\"{route}\"}}",
|
||||
"{\"name\":\"$return\",\"type\":\"http\",\"direction\":\"Out\"}",
|
||||
"{\"name\":\"client\",\"type\":\"durableClient\",\"direction\":\"In\"}"
|
||||
],
|
||||
|
||||
+11
-8
@@ -1,4 +1,4 @@
|
||||
// Copyright (c) Microsoft. All rights reserved.
|
||||
// Copyright (c) Microsoft. All rights reserved.
|
||||
|
||||
using Microsoft.Agents.AI.DurableTask;
|
||||
using Microsoft.Agents.AI.DurableTask.Workflows;
|
||||
@@ -62,10 +62,10 @@ public static class FunctionsApplicationBuilderExtensions
|
||||
ArgumentNullException.ThrowIfNull(builder);
|
||||
ArgumentNullException.ThrowIfNull(configure);
|
||||
|
||||
builder.Services.ConfigureDurableOptions(configure);
|
||||
// Ensure FunctionsDurableOptions is registered BEFORE the core extension creates a plain DurableOptions
|
||||
FunctionsDurableOptions sharedOptions = GetOrCreateSharedOptions(builder.Services);
|
||||
|
||||
// Read the shared options to check if workflows were added
|
||||
DurableOptions sharedOptions = GetOrCreateSharedOptions(builder.Services);
|
||||
builder.Services.ConfigureDurableOptions(configure);
|
||||
|
||||
if (sharedOptions.Workflows.Workflows.Count > 0)
|
||||
{
|
||||
@@ -105,7 +105,9 @@ public static class FunctionsApplicationBuilderExtensions
|
||||
string.Equals(context.FunctionDefinition.EntryPoint, BuiltInFunctions.RunAgentEntityFunctionEntryPoint, StringComparison.Ordinal) ||
|
||||
string.Equals(context.FunctionDefinition.EntryPoint, BuiltInFunctions.RunWorkflowOrchestrationHttpFunctionEntryPoint, StringComparison.Ordinal) ||
|
||||
string.Equals(context.FunctionDefinition.EntryPoint, BuiltInFunctions.RunWorkflowOrchestrationFunctionEntryPoint, StringComparison.Ordinal) ||
|
||||
string.Equals(context.FunctionDefinition.EntryPoint, BuiltInFunctions.InvokeWorkflowActivityFunctionEntryPoint, StringComparison.Ordinal)
|
||||
string.Equals(context.FunctionDefinition.EntryPoint, BuiltInFunctions.InvokeWorkflowActivityFunctionEntryPoint, StringComparison.Ordinal) ||
|
||||
string.Equals(context.FunctionDefinition.EntryPoint, BuiltInFunctions.GetWorkflowStatusHttpFunctionEntryPoint, StringComparison.Ordinal) ||
|
||||
string.Equals(context.FunctionDefinition.EntryPoint, BuiltInFunctions.RespondToWorkflowHttpFunctionEntryPoint, StringComparison.Ordinal)
|
||||
);
|
||||
builder.Services.TryAddSingleton<BuiltInFunctionExecutor>();
|
||||
}
|
||||
@@ -113,17 +115,18 @@ public static class FunctionsApplicationBuilderExtensions
|
||||
/// <summary>
|
||||
/// Gets or creates a shared <see cref="DurableOptions"/> instance from the service collection.
|
||||
/// </summary>
|
||||
private static DurableOptions GetOrCreateSharedOptions(IServiceCollection services)
|
||||
private static FunctionsDurableOptions GetOrCreateSharedOptions(IServiceCollection services)
|
||||
{
|
||||
ServiceDescriptor? existingDescriptor = services.FirstOrDefault(
|
||||
d => d.ServiceType == typeof(DurableOptions) && d.ImplementationInstance is not null);
|
||||
|
||||
if (existingDescriptor?.ImplementationInstance is DurableOptions existing)
|
||||
if (existingDescriptor?.ImplementationInstance is FunctionsDurableOptions existing)
|
||||
{
|
||||
return existing;
|
||||
}
|
||||
|
||||
DurableOptions options = new();
|
||||
FunctionsDurableOptions options = new();
|
||||
services.AddSingleton<DurableOptions>(options);
|
||||
services.AddSingleton(options);
|
||||
return options;
|
||||
}
|
||||
|
||||
@@ -0,0 +1,29 @@
|
||||
// Copyright (c) Microsoft. All rights reserved.
|
||||
|
||||
using Microsoft.Agents.AI.DurableTask;
|
||||
|
||||
namespace Microsoft.Agents.AI.Hosting.AzureFunctions;
|
||||
|
||||
/// <summary>
|
||||
/// Provides Azure Functions–specific configuration for durable workflows.
|
||||
/// </summary>
|
||||
internal sealed class FunctionsDurableOptions : DurableOptions
|
||||
{
|
||||
private readonly HashSet<string> _statusEndpointWorkflows = new(StringComparer.OrdinalIgnoreCase);
|
||||
|
||||
/// <summary>
|
||||
/// Enables the status HTTP endpoint for the specified workflow.
|
||||
/// </summary>
|
||||
internal void EnableStatusEndpoint(string workflowName)
|
||||
{
|
||||
this._statusEndpointWorkflows.Add(workflowName);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Returns whether the status endpoint is enabled for the specified workflow.
|
||||
/// </summary>
|
||||
internal bool IsStatusEndpointEnabled(string workflowName)
|
||||
{
|
||||
return this._statusEndpointWorkflows.Contains(workflowName);
|
||||
}
|
||||
}
|
||||
+30
@@ -0,0 +1,30 @@
|
||||
// Copyright (c) Microsoft. All rights reserved.
|
||||
|
||||
using Microsoft.Agents.AI.DurableTask.Workflows;
|
||||
using Microsoft.Agents.AI.Workflows;
|
||||
|
||||
namespace Microsoft.Agents.AI.Hosting.AzureFunctions;
|
||||
|
||||
/// <summary>
|
||||
/// Extension methods for <see cref="DurableWorkflowOptions"/> to configure Azure Functions HTTP trigger options.
|
||||
/// </summary>
|
||||
public static class DurableWorkflowOptionsExtensions
|
||||
{
|
||||
/// <summary>
|
||||
/// Adds a workflow and optionally exposes a status HTTP endpoint for querying pending HITL requests.
|
||||
/// </summary>
|
||||
/// <param name="options">The workflow options to add the workflow to.</param>
|
||||
/// <param name="workflow">The workflow instance to add.</param>
|
||||
/// <param name="exposeStatusEndpoint">If <see langword="true"/>, a GET endpoint is generated at <c>workflows/{name}/status/{runId}</c>.</param>
|
||||
public static void AddWorkflow(this DurableWorkflowOptions options, Workflow workflow, bool exposeStatusEndpoint)
|
||||
{
|
||||
ArgumentNullException.ThrowIfNull(options);
|
||||
|
||||
options.AddWorkflow(workflow);
|
||||
|
||||
if (exposeStatusEndpoint && options.ParentOptions is FunctionsDurableOptions functionsOptions)
|
||||
{
|
||||
functionsOptions.EnableStatusEndpoint(workflow.Name!);
|
||||
}
|
||||
}
|
||||
}
|
||||
+39
-6
@@ -25,18 +25,20 @@ namespace Microsoft.Agents.AI.Hosting.AzureFunctions;
|
||||
internal sealed class DurableWorkflowsFunctionMetadataTransformer : IFunctionMetadataTransformer
|
||||
{
|
||||
private readonly ILogger<DurableWorkflowsFunctionMetadataTransformer> _logger;
|
||||
private readonly DurableWorkflowOptions _options;
|
||||
private readonly FunctionsDurableOptions _options;
|
||||
|
||||
/// <summary>
|
||||
/// Initializes a new instance of the <see cref="DurableWorkflowsFunctionMetadataTransformer"/> class.
|
||||
/// </summary>
|
||||
/// <param name="logger">The logger instance for diagnostic output.</param>
|
||||
/// <param name="durableOptions">The durable options containing workflow configurations.</param>
|
||||
public DurableWorkflowsFunctionMetadataTransformer(ILogger<DurableWorkflowsFunctionMetadataTransformer> logger, DurableOptions durableOptions)
|
||||
public DurableWorkflowsFunctionMetadataTransformer(
|
||||
ILogger<DurableWorkflowsFunctionMetadataTransformer> logger,
|
||||
FunctionsDurableOptions durableOptions)
|
||||
{
|
||||
this._logger = logger ?? throw new ArgumentNullException(nameof(logger));
|
||||
ArgumentNullException.ThrowIfNull(durableOptions);
|
||||
this._options = durableOptions.Workflows;
|
||||
this._options = durableOptions;
|
||||
}
|
||||
|
||||
/// <inheritdoc />
|
||||
@@ -51,7 +53,8 @@ internal sealed class DurableWorkflowsFunctionMetadataTransformer : IFunctionMet
|
||||
// Track registered function names to avoid duplicates when workflows share executors.
|
||||
HashSet<string> registeredFunctions = [];
|
||||
|
||||
foreach (var workflow in this._options.Workflows)
|
||||
DurableWorkflowOptions workflowOptions = this._options.Workflows;
|
||||
foreach (var workflow in workflowOptions.Workflows)
|
||||
{
|
||||
string httpFunctionName = $"{BuiltInFunctions.HttpPrefix}{workflow.Key}";
|
||||
|
||||
@@ -80,12 +83,42 @@ internal sealed class DurableWorkflowsFunctionMetadataTransformer : IFunctionMet
|
||||
BuiltInFunctions.RunWorkflowOrchestrationHttpFunctionEntryPoint));
|
||||
}
|
||||
|
||||
// Register a status endpoint if opted in via AddWorkflow(exposeStatusEndpoint: true).
|
||||
if (this._options.IsStatusEndpointEnabled(workflow.Key))
|
||||
{
|
||||
string statusFunctionName = $"{BuiltInFunctions.HttpPrefix}{workflow.Key}-status";
|
||||
if (registeredFunctions.Add(statusFunctionName))
|
||||
{
|
||||
this._logger.LogRegisteringWorkflowTrigger(workflow.Key, statusFunctionName, "http-status");
|
||||
original.Add(FunctionMetadataFactory.CreateHttpTrigger(
|
||||
$"{workflow.Key}-status",
|
||||
$"workflows/{workflow.Key}/status/{{runId}}",
|
||||
BuiltInFunctions.GetWorkflowStatusHttpFunctionEntryPoint,
|
||||
methods: "\"get\""));
|
||||
}
|
||||
}
|
||||
|
||||
// Register a respond endpoint when the workflow contains RequestPort nodes.
|
||||
bool hasRequestPorts = workflow.Value.ReflectExecutors().Values.Any(b => b is RequestPortBinding);
|
||||
if (hasRequestPorts)
|
||||
{
|
||||
string respondFunctionName = $"{BuiltInFunctions.HttpPrefix}{workflow.Key}-respond";
|
||||
if (registeredFunctions.Add(respondFunctionName))
|
||||
{
|
||||
this._logger.LogRegisteringWorkflowTrigger(workflow.Key, respondFunctionName, "http-respond");
|
||||
original.Add(FunctionMetadataFactory.CreateHttpTrigger(
|
||||
$"{workflow.Key}-respond",
|
||||
$"workflows/{workflow.Key}/respond/{{runId}}",
|
||||
BuiltInFunctions.RespondToWorkflowHttpFunctionEntryPoint));
|
||||
}
|
||||
}
|
||||
|
||||
// Register activity or entity functions for each executor in the workflow.
|
||||
// ReflectExecutors() returns all executors across the graph; no need to manually traverse edges.
|
||||
foreach (KeyValuePair<string, ExecutorBinding> entry in workflow.Value.ReflectExecutors())
|
||||
{
|
||||
// Sub-workflow bindings are handled as separate orchestrations, not activities.
|
||||
if (entry.Value is SubworkflowBinding)
|
||||
// Sub-workflow and RequestPort bindings use specialized dispatch, not activities.
|
||||
if (entry.Value is SubworkflowBinding or RequestPortBinding)
|
||||
{
|
||||
continue;
|
||||
}
|
||||
|
||||
+53
@@ -451,6 +451,59 @@ public sealed class WorkflowConsoleAppSamplesValidation(ITestOutputHelper output
|
||||
});
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task WorkflowHITLSampleValidationAsync()
|
||||
{
|
||||
using CancellationTokenSource testTimeoutCts = this.CreateTestTimeoutCts();
|
||||
string samplePath = Path.Combine(s_samplesPath, "08_WorkflowHITL");
|
||||
|
||||
await this.RunSampleTestAsync(samplePath, (process, logs) =>
|
||||
{
|
||||
bool foundStarted = false;
|
||||
bool foundManagerApprovalPause = false;
|
||||
bool foundManagerApprovalInput = false;
|
||||
bool foundManagerResponseSent = false;
|
||||
bool foundBudgetApprovalPause = false;
|
||||
bool foundBudgetResponseSent = false;
|
||||
bool foundComplianceApprovalPause = false;
|
||||
bool foundComplianceResponseSent = false;
|
||||
bool foundWorkflowCompleted = false;
|
||||
|
||||
string? line;
|
||||
while ((line = this.ReadLogLine(logs, testTimeoutCts.Token)) != null)
|
||||
{
|
||||
foundStarted |= line.Contains("Starting expense reimbursement workflow", StringComparison.Ordinal);
|
||||
foundManagerApprovalPause |= line.Contains("Workflow paused at RequestPort: ManagerApproval", StringComparison.Ordinal);
|
||||
foundManagerApprovalInput |= line.Contains("Approval for: Jerry", StringComparison.Ordinal);
|
||||
foundManagerResponseSent |= line.Contains("Response sent: Approved=True", StringComparison.Ordinal) && foundManagerApprovalPause && !foundBudgetApprovalPause && !foundComplianceApprovalPause;
|
||||
foundBudgetApprovalPause |= line.Contains("Workflow paused at RequestPort: BudgetApproval", StringComparison.Ordinal);
|
||||
foundBudgetResponseSent |= line.Contains("Response sent: Approved=True", StringComparison.Ordinal) && foundBudgetApprovalPause;
|
||||
foundComplianceApprovalPause |= line.Contains("Workflow paused at RequestPort: ComplianceApproval", StringComparison.Ordinal);
|
||||
foundComplianceResponseSent |= line.Contains("Response sent: Approved=True", StringComparison.Ordinal) && foundComplianceApprovalPause;
|
||||
|
||||
if (line.Contains("Workflow completed: Expense reimbursed at", StringComparison.Ordinal))
|
||||
{
|
||||
foundWorkflowCompleted = true;
|
||||
break;
|
||||
}
|
||||
|
||||
this.AssertNoError(line);
|
||||
}
|
||||
|
||||
Assert.True(foundStarted, "Workflow start message not found.");
|
||||
Assert.True(foundManagerApprovalPause, "Manager approval pause not found.");
|
||||
Assert.True(foundManagerApprovalInput, "Manager approval input (Jerry) not found.");
|
||||
Assert.True(foundManagerResponseSent, "Manager approval response not sent.");
|
||||
Assert.True(foundBudgetApprovalPause, "Budget approval pause not found.");
|
||||
Assert.True(foundBudgetResponseSent, "Budget approval response not sent.");
|
||||
Assert.True(foundComplianceApprovalPause, "Compliance approval pause not found.");
|
||||
Assert.True(foundComplianceResponseSent, "Compliance approval response not sent.");
|
||||
Assert.True(foundWorkflowCompleted, "Workflow did not complete successfully.");
|
||||
|
||||
return Task.CompletedTask;
|
||||
});
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task WorkflowAndAgentsSampleValidationAsync()
|
||||
{
|
||||
|
||||
+235
@@ -0,0 +1,235 @@
|
||||
// Copyright (c) Microsoft. All rights reserved.
|
||||
|
||||
using System.Text.Json;
|
||||
using Microsoft.Agents.AI.DurableTask.Workflows;
|
||||
|
||||
namespace Microsoft.Agents.AI.DurableTask.UnitTests.Workflows;
|
||||
|
||||
public sealed class DurableActivityExecutorTests
|
||||
{
|
||||
private static readonly JsonSerializerOptions s_camelCaseOptions = new()
|
||||
{
|
||||
PropertyNamingPolicy = JsonNamingPolicy.CamelCase,
|
||||
PropertyNameCaseInsensitive = true
|
||||
};
|
||||
|
||||
#region DeserializeInput
|
||||
|
||||
[Fact]
|
||||
public void DeserializeInput_StringType_ReturnsInputAsIs()
|
||||
{
|
||||
// Arrange
|
||||
const string Input = "hello world";
|
||||
|
||||
// Act
|
||||
object result = DurableActivityExecutor.DeserializeInput(Input, typeof(string));
|
||||
|
||||
// Assert
|
||||
Assert.Equal("hello world", result);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void DeserializeInput_SimpleObject_DeserializesCorrectly()
|
||||
{
|
||||
// Arrange
|
||||
string input = JsonSerializer.Serialize(new TestRecord("EXP-001", 100.50m), s_camelCaseOptions);
|
||||
|
||||
// Act
|
||||
object result = DurableActivityExecutor.DeserializeInput(input, typeof(TestRecord));
|
||||
|
||||
// Assert
|
||||
TestRecord record = Assert.IsType<TestRecord>(result);
|
||||
Assert.Equal("EXP-001", record.Id);
|
||||
Assert.Equal(100.50m, record.Amount);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void DeserializeInput_StringArray_DeserializesDirectly()
|
||||
{
|
||||
// Arrange
|
||||
string input = JsonSerializer.Serialize((string[])["a", "b", "c"]);
|
||||
|
||||
// Act
|
||||
object result = DurableActivityExecutor.DeserializeInput(input, typeof(string[]));
|
||||
|
||||
// Assert
|
||||
string[] array = Assert.IsType<string[]>(result);
|
||||
Assert.Equal(["a", "b", "c"], array);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void DeserializeInput_TypedArrayFromFanIn_DeserializesEachElement()
|
||||
{
|
||||
// Arrange — fan-in produces a JSON array of serialized strings
|
||||
TestRecord r1 = new("EXP-001", 100m);
|
||||
TestRecord r2 = new("EXP-002", 200m);
|
||||
string[] serializedElements =
|
||||
[
|
||||
JsonSerializer.Serialize(r1, s_camelCaseOptions),
|
||||
JsonSerializer.Serialize(r2, s_camelCaseOptions)
|
||||
];
|
||||
string input = JsonSerializer.Serialize(serializedElements);
|
||||
|
||||
// Act
|
||||
object result = DurableActivityExecutor.DeserializeInput(input, typeof(TestRecord[]));
|
||||
|
||||
// Assert
|
||||
TestRecord[] records = Assert.IsType<TestRecord[]>(result);
|
||||
Assert.Equal(2, records.Length);
|
||||
Assert.Equal("EXP-001", records[0].Id);
|
||||
Assert.Equal(100m, records[0].Amount);
|
||||
Assert.Equal("EXP-002", records[1].Id);
|
||||
Assert.Equal(200m, records[1].Amount);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void DeserializeInput_TypedArrayWithSingleElement_DeserializesCorrectly()
|
||||
{
|
||||
// Arrange
|
||||
TestRecord r1 = new("EXP-001", 50m);
|
||||
string[] serializedElements = [JsonSerializer.Serialize(r1, s_camelCaseOptions)];
|
||||
string input = JsonSerializer.Serialize(serializedElements);
|
||||
|
||||
// Act
|
||||
object result = DurableActivityExecutor.DeserializeInput(input, typeof(TestRecord[]));
|
||||
|
||||
// Assert
|
||||
TestRecord[] records = Assert.IsType<TestRecord[]>(result);
|
||||
Assert.Single(records);
|
||||
Assert.Equal("EXP-001", records[0].Id);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void DeserializeInput_TypedArrayWithNullElement_ThrowsInvalidOperationException()
|
||||
{
|
||||
// Arrange — one element is "null"
|
||||
string input = JsonSerializer.Serialize((string[])["null"]);
|
||||
|
||||
// Act & Assert
|
||||
Assert.Throws<InvalidOperationException>(
|
||||
() => DurableActivityExecutor.DeserializeInput(input, typeof(TestRecord[])));
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void DeserializeInput_InvalidJson_ThrowsJsonException()
|
||||
{
|
||||
// Arrange
|
||||
const string Input = "not valid json";
|
||||
|
||||
// Act & Assert
|
||||
Assert.ThrowsAny<JsonException>(
|
||||
() => DurableActivityExecutor.DeserializeInput(Input, typeof(TestRecord)));
|
||||
}
|
||||
|
||||
#endregion
|
||||
|
||||
#region ResolveInputType
|
||||
|
||||
[Fact]
|
||||
public void ResolveInputType_NullTypeName_ReturnsFirstSupportedType()
|
||||
{
|
||||
// Arrange
|
||||
HashSet<Type> supportedTypes = [typeof(TestRecord), typeof(string)];
|
||||
|
||||
// Act
|
||||
Type result = DurableActivityExecutor.ResolveInputType(null, supportedTypes);
|
||||
|
||||
// Assert
|
||||
Assert.Equal(typeof(TestRecord), result);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void ResolveInputType_EmptyTypeName_ReturnsFirstSupportedType()
|
||||
{
|
||||
// Arrange
|
||||
HashSet<Type> supportedTypes = [typeof(TestRecord)];
|
||||
|
||||
// Act
|
||||
Type result = DurableActivityExecutor.ResolveInputType(string.Empty, supportedTypes);
|
||||
|
||||
// Assert
|
||||
Assert.Equal(typeof(TestRecord), result);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void ResolveInputType_EmptySupportedTypes_DefaultsToString()
|
||||
{
|
||||
// Arrange
|
||||
HashSet<Type> supportedTypes = [];
|
||||
|
||||
// Act
|
||||
Type result = DurableActivityExecutor.ResolveInputType(null, supportedTypes);
|
||||
|
||||
// Assert
|
||||
Assert.Equal(typeof(string), result);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void ResolveInputType_MatchesByFullName()
|
||||
{
|
||||
// Arrange
|
||||
HashSet<Type> supportedTypes = [typeof(TestRecord)];
|
||||
|
||||
// Act
|
||||
Type result = DurableActivityExecutor.ResolveInputType(typeof(TestRecord).FullName, supportedTypes);
|
||||
|
||||
// Assert
|
||||
Assert.Equal(typeof(TestRecord), result);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void ResolveInputType_MatchesByName()
|
||||
{
|
||||
// Arrange
|
||||
HashSet<Type> supportedTypes = [typeof(TestRecord)];
|
||||
|
||||
// Act
|
||||
Type result = DurableActivityExecutor.ResolveInputType("TestRecord", supportedTypes);
|
||||
|
||||
// Assert
|
||||
Assert.Equal(typeof(TestRecord), result);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void ResolveInputType_StringArrayFallsBackToSupportedType()
|
||||
{
|
||||
// Arrange — fan-in sends string[] but executor expects TestRecord[]
|
||||
HashSet<Type> supportedTypes = [typeof(TestRecord[])];
|
||||
|
||||
// Act
|
||||
Type result = DurableActivityExecutor.ResolveInputType(typeof(string[]).FullName, supportedTypes);
|
||||
|
||||
// Assert
|
||||
Assert.Equal(typeof(TestRecord[]), result);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void ResolveInputType_StringFallsBackToSupportedType()
|
||||
{
|
||||
// Arrange — executor doesn't support string
|
||||
HashSet<Type> supportedTypes = [typeof(TestRecord)];
|
||||
|
||||
// Act
|
||||
Type result = DurableActivityExecutor.ResolveInputType(typeof(string).FullName, supportedTypes);
|
||||
|
||||
// Assert
|
||||
Assert.Equal(typeof(TestRecord), result);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void ResolveInputType_StringArrayRetainedWhenSupported()
|
||||
{
|
||||
// Arrange — executor explicitly supports string[]
|
||||
HashSet<Type> supportedTypes = [typeof(string[])];
|
||||
|
||||
// Act
|
||||
Type result = DurableActivityExecutor.ResolveInputType(typeof(string[]).FullName, supportedTypes);
|
||||
|
||||
// Assert
|
||||
Assert.Equal(typeof(string[]), result);
|
||||
}
|
||||
|
||||
#endregion
|
||||
|
||||
private sealed record TestRecord(string Id, decimal Amount);
|
||||
}
|
||||
+143
-2
@@ -36,8 +36,28 @@ public sealed class DurableStreamingWorkflowRunTests
|
||||
|
||||
private static string SerializeCustomStatus(List<string> events)
|
||||
{
|
||||
DurableWorkflowCustomStatus status = new() { Events = events };
|
||||
return JsonSerializer.Serialize(status, DurableWorkflowJsonContext.Default.DurableWorkflowCustomStatus);
|
||||
DurableWorkflowLiveStatus status = new() { Events = events };
|
||||
return JsonSerializer.Serialize(status, DurableSerialization.Options);
|
||||
}
|
||||
|
||||
private static string SerializeCustomStatusWithPendingEvents(
|
||||
List<string> events,
|
||||
List<PendingRequestPortStatus> pendingEvents)
|
||||
{
|
||||
DurableWorkflowLiveStatus status = new() { Events = events, PendingEvents = pendingEvents };
|
||||
return JsonSerializer.Serialize(status, DurableSerialization.Options);
|
||||
}
|
||||
|
||||
private static Workflow CreateTestWorkflowWithRequestPort(string requestPortId)
|
||||
{
|
||||
FunctionExecutor<string> start = new("start", (_, _, _) => default);
|
||||
RequestPort<string, string> requestPort = RequestPort.Create<string, string>(requestPortId);
|
||||
FunctionExecutor<string> end = new("end", (_, _, _) => default);
|
||||
return new WorkflowBuilder(start)
|
||||
.WithName(WorkflowTestName)
|
||||
.AddEdge(start, requestPort)
|
||||
.AddEdge(requestPort, end)
|
||||
.Build();
|
||||
}
|
||||
|
||||
private static string SerializeWorkflowResult(string? result, List<string> events)
|
||||
@@ -486,6 +506,127 @@ public sealed class DurableStreamingWorkflowRunTests
|
||||
Assert.Empty(events);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task WatchStreamAsync_PendingRequestPort_YieldsWaitingForInputEventAsync()
|
||||
{
|
||||
// Arrange
|
||||
string customStatus = SerializeCustomStatusWithPendingEvents(
|
||||
[],
|
||||
[new PendingRequestPortStatus("ApprovalPort", """{"amount":100}""")]);
|
||||
string serializedOutput = SerializeWorkflowResult("approved", []);
|
||||
|
||||
int callCount = 0;
|
||||
Mock<DurableTaskClient> mockClient = new("test");
|
||||
mockClient.Setup(c => c.GetInstanceAsync(InstanceId, true, It.IsAny<CancellationToken>()))
|
||||
.ReturnsAsync(() =>
|
||||
{
|
||||
callCount++;
|
||||
return callCount == 1
|
||||
? CreateMetadata(OrchestrationRuntimeStatus.Running, serializedCustomStatus: customStatus)
|
||||
: CreateMetadata(OrchestrationRuntimeStatus.Completed, serializedOutput: serializedOutput);
|
||||
});
|
||||
|
||||
Workflow workflow = CreateTestWorkflowWithRequestPort("ApprovalPort");
|
||||
DurableStreamingWorkflowRun run = new(mockClient.Object, InstanceId, workflow);
|
||||
|
||||
// Act
|
||||
List<WorkflowEvent> events = [];
|
||||
await foreach (WorkflowEvent evt in run.WatchStreamAsync())
|
||||
{
|
||||
events.Add(evt);
|
||||
}
|
||||
|
||||
// Assert
|
||||
Assert.Equal(2, events.Count);
|
||||
DurableWorkflowWaitingForInputEvent waitingEvent = Assert.IsType<DurableWorkflowWaitingForInputEvent>(events[0]);
|
||||
Assert.Equal("ApprovalPort", waitingEvent.RequestPort.Id);
|
||||
Assert.Contains("amount", waitingEvent.Input);
|
||||
DurableWorkflowCompletedEvent completedEvent = Assert.IsType<DurableWorkflowCompletedEvent>(events[1]);
|
||||
Assert.Equal("approved", completedEvent.Result);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task WatchStreamAsync_PendingRequestPort_DoesNotDuplicateOnSubsequentPollsAsync()
|
||||
{
|
||||
// Arrange — same pending event across 2 polls, then completion
|
||||
string customStatus = SerializeCustomStatusWithPendingEvents(
|
||||
[],
|
||||
[new PendingRequestPortStatus("ApprovalPort", """{"amount":100}""")]);
|
||||
string serializedOutput = SerializeWorkflowResult("done", []);
|
||||
|
||||
int callCount = 0;
|
||||
Mock<DurableTaskClient> mockClient = new("test");
|
||||
mockClient.Setup(c => c.GetInstanceAsync(InstanceId, true, It.IsAny<CancellationToken>()))
|
||||
.ReturnsAsync(() =>
|
||||
{
|
||||
callCount++;
|
||||
return callCount switch
|
||||
{
|
||||
<= 2 => CreateMetadata(OrchestrationRuntimeStatus.Running, serializedCustomStatus: customStatus),
|
||||
_ => CreateMetadata(OrchestrationRuntimeStatus.Completed, serializedOutput: serializedOutput),
|
||||
};
|
||||
});
|
||||
|
||||
Workflow workflow = CreateTestWorkflowWithRequestPort("ApprovalPort");
|
||||
DurableStreamingWorkflowRun run = new(mockClient.Object, InstanceId, workflow);
|
||||
|
||||
// Act
|
||||
List<WorkflowEvent> events = [];
|
||||
await foreach (WorkflowEvent evt in run.WatchStreamAsync())
|
||||
{
|
||||
events.Add(evt);
|
||||
}
|
||||
|
||||
// Assert — WaitingForInputEvent yielded only once despite 2 polls
|
||||
Assert.Equal(2, events.Count);
|
||||
Assert.IsType<DurableWorkflowWaitingForInputEvent>(events[0]);
|
||||
Assert.IsType<DurableWorkflowCompletedEvent>(events[1]);
|
||||
}
|
||||
|
||||
#endregion
|
||||
|
||||
#region SendResponseAsync
|
||||
|
||||
[Fact]
|
||||
public async Task SendResponseAsync_SerializesAndRaisesEventAsync()
|
||||
{
|
||||
// Arrange
|
||||
Mock<DurableTaskClient> mockClient = new("test");
|
||||
mockClient.Setup(c => c.RaiseEventAsync(
|
||||
InstanceId,
|
||||
"ApprovalPort",
|
||||
It.IsAny<string>(),
|
||||
It.IsAny<CancellationToken>()))
|
||||
.Returns(Task.CompletedTask);
|
||||
|
||||
RequestPort approvalPort = RequestPort.Create<string, string>("ApprovalPort");
|
||||
DurableWorkflowWaitingForInputEvent requestEvent = new("""{"amount":100}""", approvalPort);
|
||||
Workflow workflow = CreateTestWorkflowWithRequestPort("ApprovalPort");
|
||||
DurableStreamingWorkflowRun run = new(mockClient.Object, InstanceId, workflow);
|
||||
|
||||
// Act
|
||||
await run.SendResponseAsync(requestEvent, new { approved = true, comments = "Looks good" });
|
||||
|
||||
// Assert
|
||||
mockClient.Verify(c => c.RaiseEventAsync(
|
||||
InstanceId,
|
||||
"ApprovalPort",
|
||||
It.Is<string>(s => s.Contains("approved") && s.Contains("true")),
|
||||
It.IsAny<CancellationToken>()), Times.Once);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task SendResponseAsync_NullRequestEvent_ThrowsAsync()
|
||||
{
|
||||
// Arrange
|
||||
Mock<DurableTaskClient> mockClient = new("test");
|
||||
DurableStreamingWorkflowRun run = new(mockClient.Object, InstanceId, CreateTestWorkflow());
|
||||
|
||||
// Act & Assert
|
||||
await Assert.ThrowsAsync<ArgumentNullException>(() =>
|
||||
run.SendResponseAsync<string>(null!, "response").AsTask());
|
||||
}
|
||||
|
||||
#endregion
|
||||
|
||||
#region WaitForCompletionAsync
|
||||
|
||||
+110
-1
@@ -1,4 +1,4 @@
|
||||
// Copyright (c) Microsoft. All rights reserved.
|
||||
// Copyright (c) Microsoft. All rights reserved.
|
||||
|
||||
using System.Diagnostics;
|
||||
using System.Reflection;
|
||||
@@ -117,6 +117,115 @@ public sealed class WorkflowSamplesValidation(ITestOutputHelper outputHelper) :
|
||||
});
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task HITLWorkflowSampleValidationAsync()
|
||||
{
|
||||
string samplePath = Path.Combine(s_samplesPath, "03_WorkflowHITL");
|
||||
await this.RunSampleTestAsync(samplePath, requiresOpenAI: false, async (logs) =>
|
||||
{
|
||||
// Use a unique run ID to avoid conflicts with previous test runs
|
||||
string runId = $"hitl-test-{Guid.NewGuid():N}";
|
||||
|
||||
// Step 1: Start the expense reimbursement workflow
|
||||
Uri runUri = new($"http://localhost:{AzureFunctionsPort}/api/workflows/ExpenseReimbursement/run?runId={runId}");
|
||||
this._outputHelper.WriteLine($"Starting ExpenseReimbursement workflow via POST request to {runUri}...");
|
||||
|
||||
using HttpContent runContent = new StringContent("EXP-2025-001", Encoding.UTF8, "text/plain");
|
||||
using HttpResponseMessage runResponse = await s_sharedHttpClient.PostAsync(runUri, runContent);
|
||||
|
||||
Assert.True(runResponse.IsSuccessStatusCode, $"Run request failed with status: {runResponse.StatusCode}");
|
||||
string runResponseText = await runResponse.Content.ReadAsStringAsync();
|
||||
Assert.Contains("ExpenseReimbursement", runResponseText);
|
||||
this._outputHelper.WriteLine($"Run response: {runResponseText}");
|
||||
|
||||
// Step 2: Wait for the workflow to pause at the ManagerApproval RequestPort
|
||||
await this.WaitForConditionAsync(
|
||||
condition: () =>
|
||||
{
|
||||
lock (logs)
|
||||
{
|
||||
bool exists = logs.Any(log => log.Message.Contains("Workflow waiting for external input at RequestPort 'ManagerApproval'"));
|
||||
return Task.FromResult(exists);
|
||||
}
|
||||
},
|
||||
message: "Workflow paused at ManagerApproval RequestPort",
|
||||
timeout: s_orchestrationTimeout);
|
||||
|
||||
// Step 3: Send approval response to resume the workflow
|
||||
Uri respondUri = new($"http://localhost:{AzureFunctionsPort}/api/workflows/ExpenseReimbursement/respond/{runId}");
|
||||
this._outputHelper.WriteLine($"Sending approval response via POST request to {respondUri}...");
|
||||
|
||||
using HttpContent respondContent = new StringContent(
|
||||
"""{"eventName": "ManagerApproval", "response": {"Approved": true, "Comments": "Approved by test."}}""",
|
||||
Encoding.UTF8, "application/json");
|
||||
using HttpResponseMessage respondResponse = await s_sharedHttpClient.PostAsync(respondUri, respondContent);
|
||||
|
||||
Assert.True(respondResponse.IsSuccessStatusCode, $"Respond request failed with status: {respondResponse.StatusCode}");
|
||||
string respondResponseText = await respondResponse.Content.ReadAsStringAsync();
|
||||
Assert.Contains("Response sent to workflow", respondResponseText);
|
||||
this._outputHelper.WriteLine($"Respond response: {respondResponseText}");
|
||||
|
||||
// Step 4: Wait for the workflow to pause at the parallel BudgetApproval and ComplianceApproval RequestPorts
|
||||
await this.WaitForConditionAsync(
|
||||
condition: () =>
|
||||
{
|
||||
lock (logs)
|
||||
{
|
||||
bool exists = logs.Any(log => log.Message.Contains("Workflow waiting for external input at RequestPort 'BudgetApproval'"));
|
||||
return Task.FromResult(exists);
|
||||
}
|
||||
},
|
||||
message: "Workflow paused at BudgetApproval RequestPort",
|
||||
timeout: s_orchestrationTimeout);
|
||||
|
||||
// Step 5a: Send budget approval response
|
||||
this._outputHelper.WriteLine("Sending BudgetApproval response...");
|
||||
|
||||
using HttpContent budgetContent = new StringContent(
|
||||
"""{"eventName": "BudgetApproval", "response": {"Approved": true, "Comments": "Budget approved by test."}}""",
|
||||
Encoding.UTF8, "application/json");
|
||||
using HttpResponseMessage budgetResponse = await s_sharedHttpClient.PostAsync(respondUri, budgetContent);
|
||||
|
||||
Assert.True(budgetResponse.IsSuccessStatusCode, $"BudgetApproval request failed with status: {budgetResponse.StatusCode}");
|
||||
this._outputHelper.WriteLine($"BudgetApproval response: {await budgetResponse.Content.ReadAsStringAsync()}");
|
||||
|
||||
// Step 5b: Send compliance approval response
|
||||
this._outputHelper.WriteLine("Sending ComplianceApproval response...");
|
||||
|
||||
using HttpContent complianceContent = new StringContent(
|
||||
"""{"eventName": "ComplianceApproval", "response": {"Approved": true, "Comments": "Compliance approved by test."}}""",
|
||||
Encoding.UTF8, "application/json");
|
||||
using HttpResponseMessage complianceResponse = await s_sharedHttpClient.PostAsync(respondUri, complianceContent);
|
||||
|
||||
Assert.True(complianceResponse.IsSuccessStatusCode, $"ComplianceApproval request failed with status: {complianceResponse.StatusCode}");
|
||||
this._outputHelper.WriteLine($"ComplianceApproval response: {await complianceResponse.Content.ReadAsStringAsync()}");
|
||||
|
||||
// Step 6: Wait for the workflow to complete
|
||||
await this.WaitForConditionAsync(
|
||||
condition: () =>
|
||||
{
|
||||
lock (logs)
|
||||
{
|
||||
bool exists = logs.Any(log => log.Message.Contains("Workflow completed"));
|
||||
return Task.FromResult(exists);
|
||||
}
|
||||
},
|
||||
message: "HITL workflow completed",
|
||||
timeout: s_orchestrationTimeout);
|
||||
|
||||
// Verify executor activities ran
|
||||
lock (logs)
|
||||
{
|
||||
Assert.True(logs.Any(log => log.Message.Contains("Received external event for RequestPort 'ManagerApproval'")),
|
||||
"ManagerApproval external event receipt not found in logs.");
|
||||
Assert.True(logs.Any(log => log.Message.Contains("Received external event for RequestPort 'BudgetApproval'")),
|
||||
"BudgetApproval external event receipt not found in logs.");
|
||||
Assert.True(logs.Any(log => log.Message.Contains("Received external event for RequestPort 'ComplianceApproval'")),
|
||||
"ComplianceApproval external event receipt not found in logs.");
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task ConcurrentWorkflowSampleValidationAsync()
|
||||
{
|
||||
|
||||
Reference in New Issue
Block a user