diff --git a/dotnet/src/Microsoft.Agents.AI.Workflows.Declarative/DeclarativeWorkflowBuilder.cs b/dotnet/src/Microsoft.Agents.AI.Workflows.Declarative/DeclarativeWorkflowBuilder.cs index aeb3d2b6e95..054aa38b237 100644 --- a/dotnet/src/Microsoft.Agents.AI.Workflows.Declarative/DeclarativeWorkflowBuilder.cs +++ b/dotnet/src/Microsoft.Agents.AI.Workflows.Declarative/DeclarativeWorkflowBuilder.cs @@ -74,6 +74,7 @@ public static Workflow Build( WorkflowFormulaState state = new(options.CreateRecalcEngine()); state.Initialize(workflowElement.WrapWithBot(), options.Configuration); + state.CaptureInitialState(); DeclarativeWorkflowExecutor rootExecutor = new(rootId, options, diff --git a/dotnet/src/Microsoft.Agents.AI.Workflows.Declarative/Interpreter/DeclarativeActionExecutor.cs b/dotnet/src/Microsoft.Agents.AI.Workflows.Declarative/Interpreter/DeclarativeActionExecutor.cs index 776caebda60..5d052c64d3d 100644 --- a/dotnet/src/Microsoft.Agents.AI.Workflows.Declarative/Interpreter/DeclarativeActionExecutor.cs +++ b/dotnet/src/Microsoft.Agents.AI.Workflows.Declarative/Interpreter/DeclarativeActionExecutor.cs @@ -64,6 +64,7 @@ protected override ProtocolBuilder ConfigureProtocol(ProtocolBuilder protocolBui /// public virtual ValueTask ResetAsync() { + this._state.Reset(); return default; } diff --git a/dotnet/src/Microsoft.Agents.AI.Workflows.Declarative/Interpreter/DeclarativeWorkflowExecutor.cs b/dotnet/src/Microsoft.Agents.AI.Workflows.Declarative/Interpreter/DeclarativeWorkflowExecutor.cs index 053f28c89ac..1126b43d8ff 100644 --- a/dotnet/src/Microsoft.Agents.AI.Workflows.Declarative/Interpreter/DeclarativeWorkflowExecutor.cs +++ b/dotnet/src/Microsoft.Agents.AI.Workflows.Declarative/Interpreter/DeclarativeWorkflowExecutor.cs @@ -42,9 +42,14 @@ internal sealed class DeclarativeWorkflowExecutor( /// public ValueTask ResetAsync() { + state.Reset(); return default; } + /// + protected override ValueTask OnCheckpointRestoredAsync(IWorkflowContext context, CancellationToken cancellationToken = default) => + state.RestoreAsync(context, cancellationToken); + /// [SendsMessage(typeof(ActionExecutorResult))] public override ValueTask HandleAsync(TInput message, IWorkflowContext context, CancellationToken cancellationToken = default) diff --git a/dotnet/src/Microsoft.Agents.AI.Workflows.Declarative/Interpreter/DelegateActionExecutor.cs b/dotnet/src/Microsoft.Agents.AI.Workflows.Declarative/Interpreter/DelegateActionExecutor.cs index 0f51eacc7c1..716a3656dac 100644 --- a/dotnet/src/Microsoft.Agents.AI.Workflows.Declarative/Interpreter/DelegateActionExecutor.cs +++ b/dotnet/src/Microsoft.Agents.AI.Workflows.Declarative/Interpreter/DelegateActionExecutor.cs @@ -52,6 +52,7 @@ protected override ProtocolBuilder ConfigureProtocol(ProtocolBuilder protocolBui /// public ValueTask ResetAsync() { + this._state.Reset(); return default; } diff --git a/dotnet/src/Microsoft.Agents.AI.Workflows.Declarative/PowerFx/WorkflowFormulaState.cs b/dotnet/src/Microsoft.Agents.AI.Workflows.Declarative/PowerFx/WorkflowFormulaState.cs index c739cb3bf95..aaff60b08b4 100644 --- a/dotnet/src/Microsoft.Agents.AI.Workflows.Declarative/PowerFx/WorkflowFormulaState.cs +++ b/dotnet/src/Microsoft.Agents.AI.Workflows.Declarative/PowerFx/WorkflowFormulaState.cs @@ -29,6 +29,8 @@ internal sealed class WorkflowFormulaState private readonly Dictionary _scopes; + private Dictionary _initialScopes; + private int _isInitialized; public RecalcEngine Engine { get; } @@ -38,6 +40,7 @@ internal sealed class WorkflowFormulaState public WorkflowFormulaState(RecalcEngine engine) { this._scopes = VariableScopeNames.AllScopes.ToDictionary(scopeName => GetScopeName(scopeName), _ => new WorkflowScope()); + this._initialScopes = this.CreateScopeSnapshot(); this.Engine = engine; this.Evaluator = new WorkflowExpressionEngine(engine); @@ -61,6 +64,18 @@ public void Set(string variableName, FormulaValue value, string? scopeName = nul public bool SetInitialized() => Interlocked.CompareExchange(ref this._isInitialized, 1, 0) == 0; + public void CaptureInitialState() + { + this._initialScopes = this.CreateScopeSnapshot(); + } + + public void Reset() + { + this.RestoreInitialState(); + Interlocked.Exchange(ref this._isInitialized, 0); + this.Bind(); + } + public async ValueTask RestoreAsync(IWorkflowContext context, CancellationToken cancellationToken) { if (!this.SetInitialized()) @@ -68,6 +83,8 @@ public async ValueTask RestoreAsync(IWorkflowContext context, CancellationToken return; } + this.RestoreInitialState(); + Stopwatch timer = Stopwatch.StartNew(); Debug.WriteLine("RESTORE CHECKPOINT - BEGIN"); await Task.WhenAll(RestorableScopes.Select(scopeName => ReadScopeAsync(scopeName))).ConfigureAwait(false); @@ -93,6 +110,22 @@ async Task ReadScopeAsync(string scopeName) } } + private Dictionary CreateScopeSnapshot() => + this._scopes.ToDictionary(scope => scope.Key, scope => new WorkflowScope(scope.Value)); + + private void RestoreInitialState() + { + foreach (KeyValuePair initialScopeEntry in this._initialScopes) + { + WorkflowScope scope = this._scopes[initialScopeEntry.Key]; + scope.Clear(); + foreach (KeyValuePair initialValueEntry in initialScopeEntry.Value) + { + scope[initialValueEntry.Key] = initialValueEntry.Value; + } + } + } + public void Bind(string? scopeNameToBind = null) { if (scopeNameToBind is not null) @@ -143,5 +176,15 @@ public static string GetScopeName(string? scopeName) /// /// The set of variables for a specific action scope. /// - private sealed class WorkflowScope : Dictionary; + private sealed class WorkflowScope : Dictionary + { + public WorkflowScope() + { + } + + public WorkflowScope(IDictionary values) + : base(values) + { + } + } } diff --git a/dotnet/tests/Microsoft.Agents.AI.Workflows.Declarative.UnitTests/DeclarativeWorkflowTest.cs b/dotnet/tests/Microsoft.Agents.AI.Workflows.Declarative.UnitTests/DeclarativeWorkflowTest.cs index f01b91e5e1a..7bb92f913c2 100644 --- a/dotnet/tests/Microsoft.Agents.AI.Workflows.Declarative.UnitTests/DeclarativeWorkflowTest.cs +++ b/dotnet/tests/Microsoft.Agents.AI.Workflows.Declarative.UnitTests/DeclarativeWorkflowTest.cs @@ -79,6 +79,52 @@ public async Task EndConversationActionAsync() this.AssertNotExecuted("sendActivity_1"); } + [Fact] + public async Task HostedWorkflowAgentIsolatesDeclarativeStateBySessionAsync() + { + // Arrange + RecordingAgentProvider provider = new(); + AIAgent agent = CreateStateEchoWorkflow(provider).AsAIAgent(id: "host", name: "host"); + AgentSession aliceSession = await agent.CreateSessionAsync(); + AgentSession mallorySession = await agent.CreateSessionAsync(); + + // Act + AgentResponse aliceSeedResponse = await agent.RunAsync("EMBER-QUARTZ-7319", aliceSession); + AgentResponse aliceResumeResponse = await agent.RunAsync("inspect-alice", aliceSession); + AgentResponse malloryInspectResponse = await agent.RunAsync("inspect-mallory", mallorySession); + _ = await agent.RunAsync("ONYX-CEDAR-4826", mallorySession); + AgentResponse aliceFinalResponse = await agent.RunAsync("inspect-alice-again", aliceSession); + + // Assert + Assert.NotSame(aliceSession, mallorySession); + Assert.Contains("Marker: \"\"", aliceSeedResponse.Text, StringComparison.Ordinal); + Assert.Contains("EMBER-QUARTZ-7319", aliceResumeResponse.Text, StringComparison.Ordinal); + Assert.DoesNotContain("EMBER-QUARTZ-7319", malloryInspectResponse.Text, StringComparison.Ordinal); + Assert.DoesNotContain("ONYX-CEDAR-4826", aliceFinalResponse.Text, StringComparison.Ordinal); + Assert.Contains("inspect-alice", aliceFinalResponse.Text, StringComparison.Ordinal); + + Assert.True(provider.MessageConversations.Count >= 3); + Assert.Equal(provider.MessageConversations[0], provider.MessageConversations[1]); + Assert.NotEqual(provider.MessageConversations[0], provider.MessageConversations[2]); + } + + [Fact] + public async Task HostedWorkflowAgentIsolatesDeclarativeStateForImplicitSessionsAsync() + { + // Arrange + RecordingAgentProvider provider = new(); + AIAgent agent = CreateStateEchoWorkflow(provider).AsAIAgent(id: "host", name: "host"); + + // Act + _ = await agent.RunAsync("EMBER-QUARTZ-7319"); + AgentResponse secondImplicitResponse = await agent.RunAsync("inspect-implicit"); + + // Assert + Assert.DoesNotContain("EMBER-QUARTZ-7319", secondImplicitResponse.Text, StringComparison.Ordinal); + Assert.True(provider.MessageConversations.Count >= 2); + Assert.NotEqual(provider.MessageConversations[0], provider.MessageConversations[1]); + } + [Fact] public async Task GotoActionAsync() { @@ -386,6 +432,81 @@ private Workflow CreateWorkflow(string workflowPath, TInput workflowInpu return DeclarativeWorkflowBuilder.Build(yamlReader, workflowContext); } + private static Workflow CreateStateEchoWorkflow(ResponseAgentProvider provider) + { + using StringReader yamlReader = new( + """ + kind: Workflow + trigger: + + kind: OnConversationStart + id: state_echo_workflow + actions: + + - kind: SendActivity + id: show_marker + activity: |- + Marker: "{Local.Marker}" + + - kind: SetVariable + id: set_marker + variable: Local.Marker + value: =System.LastMessageText + """); + DeclarativeWorkflowOptions options = new(provider); + + return DeclarativeWorkflowBuilder.Build(yamlReader, options); + } + + private sealed class RecordingAgentProvider : ResponseAgentProvider + { + public List MessageConversations { get; } = []; + + private int _conversationCount; + + public override Task CreateConversationAsync(CancellationToken cancellationToken = default) => + Task.FromResult($"conversation-{Interlocked.Increment(ref this._conversationCount):D2}"); + + public override Task CreateMessageAsync( + string conversationId, + ChatMessage conversationMessage, + CancellationToken cancellationToken = default) + { + this.MessageConversations.Add(conversationId); + return Task.FromResult(conversationMessage); + } + + public override Task GetMessageAsync( + string conversationId, + string messageId, + CancellationToken cancellationToken = default) => + Task.FromResult(new ChatMessage(ChatRole.Assistant, string.Empty) { MessageId = messageId }); + + public override async IAsyncEnumerable InvokeAgentAsync( + string agentId, + string? agentVersion, + string? conversationId, + IEnumerable? messages, + IDictionary? inputArguments, + [System.Runtime.CompilerServices.EnumeratorCancellation] CancellationToken cancellationToken = default) + { + await Task.CompletedTask; + yield break; + } + + public override async IAsyncEnumerable GetMessagesAsync( + string conversationId, + int? limit = null, + string? after = null, + string? before = null, + bool newestFirst = false, + [System.Runtime.CompilerServices.EnumeratorCancellation] CancellationToken cancellationToken = default) + { + await Task.CompletedTask; + yield break; + } + } + private static Mock CreateMockProvider(string input) { Mock mockAgentProvider = new(MockBehavior.Strict);