diff --git a/docs/decisions/0048-dotnet-declarative-state-ownership.md b/docs/decisions/0048-dotnet-declarative-state-ownership.md new file mode 100644 index 00000000000..0a41e8366f0 --- /dev/null +++ b/docs/decisions/0048-dotnet-declarative-state-ownership.md @@ -0,0 +1,77 @@ +--- +status: proposed +contact: baywet +date: 2026-10-28 +deciders: westey-m +--- + +# Explicit ownership for .NET declarative Local state + +## Context and Problem Statement + +Declarative `Local` and `Topic` variables share canonical `Local` storage and a sensitivity sidecar. Older +checkpoints can instead contain declarative values in physical `Topic` storage. Code-first executors can also use +physical `Topic` independently, including the same key as a declarative variable. + +Sensitivity metadata alone cannot identify which physical value it describes. A preflight check cannot prevent +this ambiguity when different fan-out executors queue raw `Topic` and canonical `Local` writes in the same +superstep: each executor sees only its own pending updates. + +## Decision Drivers + +- Preserve unrelated code-first state and the native representation of canonical values. +- Continue reading and selectively migrating older checkpoints without guessing assignment order. +- Keep declarative storage conventions out of the language-neutral workflow state manager. +- Avoid full-scope scans for individual reads and writes. +- Preserve read-only expression snapshots and the existing checkpoint restoration publication boundary. + +## Considered Options + +### Infer ownership from sensitivity metadata and visible keys + +Good: no additional persisted metadata or core API changes. +Bad: executor-private pending updates defeat the preflight, and committed sensitivity metadata cannot distinguish +canonical ownership from an older declarative `Topic` value. + +### Validate combined updates before publication + +Good: a validator can inspect updates from every executor before they are committed. +Bad: this needs a new language-neutral validation extension point or declarative scope knowledge in core +publication. Rejecting every same-key pair would also prevent independently owned raw and canonical values. + +### Persist versioned canonical ownership separately + +Good: plain helpers, wrapped contexts, and restoration can distinguish ownership after concurrent publication, +without changing native values or public workflow APIs. +Bad: writes and checkpoints include one additional private entry per canonical Local key. Older runtimes do not +understand that ownership when both physical scopes contain the same key. + +## Decision Outcome + +Chosen option: "Persist versioned canonical ownership separately", because the ambiguity belongs to declarative +storage interpretation and does not require changing core publication semantics. + +For each new canonical `Local` assignment or migrated legacy entry, persist integer version `1` under the same key +in the private `__Microsoft_Agents_AI_Workflows_Declarative_Ownership:Local` scope, alongside the existing value and +sensitivity sidecar. No ownership entry is needed for other scopes. + +A recognized ownership entry means the sensitivity sidecar describes canonical `Local` state. Any same-key raw +`Topic` value remains independent and must not be read as a legacy value, migrated, or deleted by declarative +helpers or restoration. This permits mixed fan-out writes to coexist. Unsupported ownership versions are rejected +when consulted to disambiguate a `Topic` entry. + +Without ownership metadata, preserve the existing conservative compatibility rules: a metadata-marked `Topic` key +without a `Local` value is legacy; overlapping marked values are rejected before migration writes; unmarked raw +`Topic` entries remain untouched. Visible unmarked collisions still reject new canonical assignments unless +canonical ownership is already recorded. Missing ownership alone does not trigger a write during restoration. + +Migration records canonical ownership before removing only the migrated legacy key. Plain helper null updates +physically delete the canonical value, sensitivity, and ownership entries. Wrapped declarative clears retain the +existing stored blank representation with its sensitivity and ownership. + +Point operations use key-level `PortableValue` existence probes, including for metadata with an explicit `None` +value. Scope enumeration remains appropriate for whole-state restoration and read-only snapshots, but not for +each point operation in a growing batch. Ad hoc expression snapshots never persist migration or ownership updates. + +Checkpoint stores must preserve these private entries together with workflow values. This format is +backward-readable by the new runtime; resuming mixed-scope checkpoints on older runtimes is not supported. diff --git a/dotnet/src/Microsoft.Agents.AI.Workflows.Declarative/Extensions/PortableValueExtensions.cs b/dotnet/src/Microsoft.Agents.AI.Workflows.Declarative/Extensions/PortableValueExtensions.cs index 3e883a1f6c4..18c1f73f92b 100644 --- a/dotnet/src/Microsoft.Agents.AI.Workflows.Declarative/Extensions/PortableValueExtensions.cs +++ b/dotnet/src/Microsoft.Agents.AI.Workflows.Declarative/Extensions/PortableValueExtensions.cs @@ -14,26 +14,32 @@ namespace Microsoft.Agents.AI.Workflows.Declarative.Extensions; internal static class PortableValueExtensions { public static FormulaValue ToFormula(this PortableValue value) => + value.ToPortableObject() switch + { + Dictionary recordValue => recordValue.ToRecord(), + IDictionary recordValue => recordValue.ToRecord(), + PortableValue[] tableValue => tableValue.ToTable(), + var nativeValue => nativeValue.ToFormula(), + }; + + public static object ToPortableObject(this PortableValue value) => value.TypeId switch { - null => FormulaValue.NewBlank(), - _ when value.TypeId.IsMatch() => FormulaValue.NewBlank(), - _ when value.IsType(out string? stringValue) => FormulaValue.New(stringValue), - _ when value.IsSystemType(out bool? boolValue) => FormulaValue.New(boolValue.Value), - _ when value.IsSystemType(out int? intValue) => FormulaValue.New(intValue.Value), - _ when value.IsSystemType(out long? longValue) => FormulaValue.New(longValue.Value), - _ when value.IsSystemType(out decimal? decimalValue) => FormulaValue.New(decimalValue.Value), - _ when value.IsSystemType(out float? floatValue) => FormulaValue.New(floatValue.Value), - _ when value.IsSystemType(out double? doubleValue) => FormulaValue.New(doubleValue.Value), - _ when value.IsParentType(out Dictionary? recordValue) => recordValue.ToRecord(), - _ when value.IsParentType(out IDictionary? recordValue) => recordValue.ToRecord(), - _ when value.IsType(out PortableValue[]? tableValue) => tableValue.ToTable(), - _ when value.IsType(out ChatMessage? messageValue) => messageValue.ToRecord(), - _ when value.IsType(out DateTime dateValue) => - dateValue.TimeOfDay == TimeSpan.Zero ? - FormulaValue.NewDateOnly(dateValue.Date) : - FormulaValue.New(dateValue), - _ when value.IsType(out TimeSpan timeValue) => FormulaValue.New(timeValue), + null => UnassignedValue.Instance, + _ when value.TypeId.IsMatch() => UnassignedValue.Instance, + _ when value.IsType(out string? stringValue) => stringValue, + _ when value.IsSystemType(out bool? boolValue) => boolValue.Value, + _ when value.IsSystemType(out int? intValue) => intValue.Value, + _ when value.IsSystemType(out long? longValue) => longValue.Value, + _ when value.IsSystemType(out decimal? decimalValue) => decimalValue.Value, + _ when value.IsSystemType(out float? floatValue) => floatValue.Value, + _ when value.IsSystemType(out double? doubleValue) => doubleValue.Value, + _ when value.IsParentType(out Dictionary? recordValue) => recordValue, + _ when value.IsParentType(out IDictionary? recordValue) => recordValue, + _ when value.IsType(out PortableValue[]? tableValue) => tableValue, + _ when value.IsType(out ChatMessage? messageValue) => messageValue, + _ when value.IsType(out DateTime dateValue) => dateValue, + _ when value.IsType(out TimeSpan timeValue) => timeValue, _ => throw new DeclarativeModelException($"Unsupported portable type: {value.TypeId.TypeName}"), }; 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 4902589514d..b60c804ce1e 100644 --- a/dotnet/src/Microsoft.Agents.AI.Workflows.Declarative/Interpreter/DeclarativeActionExecutor.cs +++ b/dotnet/src/Microsoft.Agents.AI.Workflows.Declarative/Interpreter/DeclarativeActionExecutor.cs @@ -40,6 +40,7 @@ protected DeclarativeActionExecutor(DialogAction model, WorkflowFormulaState sta } this.State = state; + this.RegisterCheckpointRestoringCallback(state, state.RestoreCheckpointAsync); this.Model = model; } @@ -125,13 +126,6 @@ public override async ValueTask HandleAsync(ActionExecutorResult message, IWorkf protected abstract ValueTask ExecuteAsync(IWorkflowContext context, CancellationToken cancellationToken = default); - /// - /// Restore the state of the executor from a checkpoint. - /// This must be overridden to restore any state that was saved during checkpointing. - /// - protected override ValueTask OnCheckpointRestoredAsync(IWorkflowContext context, CancellationToken cancellationToken = default) => - this.State.RestoreAsync(context, cancellationToken); - protected async ValueTask AssignAsync(PropertyPath? targetPath, FormulaValue result, IWorkflowContext context, SensitivityLevel sensitivity = SensitivityLevel.None) { if (targetPath is null) diff --git a/dotnet/src/Microsoft.Agents.AI.Workflows.Declarative/Interpreter/DeclarativeWorkflowContext.cs b/dotnet/src/Microsoft.Agents.AI.Workflows.Declarative/Interpreter/DeclarativeWorkflowContext.cs index 183b2e785e7..8232b350c85 100644 --- a/dotnet/src/Microsoft.Agents.AI.Workflows.Declarative/Interpreter/DeclarativeWorkflowContext.cs +++ b/dotnet/src/Microsoft.Agents.AI.Workflows.Declarative/Interpreter/DeclarativeWorkflowContext.cs @@ -93,6 +93,8 @@ internal async ValueTask QueueStateUpdateAsync( /// public async ValueTask ReadStateAsync(string key, string? scopeName = null, CancellationToken cancellationToken = default) { + scopeName = IsManagedScope(scopeName) ? WorkflowFormulaState.GetScopeName(scopeName) : scopeName; + return typeof(TValue) switch { // Not a managed scope, just pass through. This is valid when a declarative @@ -107,6 +109,8 @@ _ when typeof(TValue) == typeof(FormulaValue) => (TValue?)(object?)this.State.Ge public async ValueTask ReadOrInitStateAsync(string key, Func initialStateFactory, string? scopeName = null, CancellationToken cancellationToken = default) { + scopeName = IsManagedScope(scopeName) ? WorkflowFormulaState.GetScopeName(scopeName) : scopeName; + return typeof(TValue) switch { // Not a managed scope, just pass through. This is valid when a declarative @@ -115,9 +119,32 @@ public async ValueTask ReadOrInitStateAsync(string key, Func await EnsureFormulaValueAsync().ConfigureAwait(false), // Retrieve native types from the source context to avoid conversion. - _ => await this.Source.ReadOrInitStateAsync(key, initialStateFactory, scopeName, cancellationToken).ConfigureAwait(false), + _ => await EnsureNativeValueAsync().ConfigureAwait(false), }; + async ValueTask EnsureNativeValueAsync() + { + PortableValue? existing = await this.Source.ReadStateAsync(key, scopeName, cancellationToken).ConfigureAwait(false); + bool legacyEntry = existing is null && + await Kit.IWorkflowContextExtensions.HasLegacyTopicEntryAsync(this.Source, key, scopeName!, forWrite: true, cancellationToken).ConfigureAwait(false); + bool initialized = false; + TValue result = await this.Source.ReadOrInitStateAsync(key, () => + { + initialized = true; + return initialStateFactory(); + }, legacyEntry ? VariableScopeNames.Topic : scopeName, cancellationToken).ConfigureAwait(false); + if (legacyEntry) + { + PortableValue legacyValue = await ReadLegacyValueAsync().ConfigureAwait(false); + await this.Source.QueueStateUpdateAsync(key, legacyValue.ToPortableObject(), scopeName, cancellationToken).ConfigureAwait(false); + } + if (initialized || legacyEntry) + { + await CompleteInitializationAsync(legacyEntry).ConfigureAwait(false); + } + return result; + } + async ValueTask EnsureFormulaValueAsync() { Debug.Assert(typeof(TValue) == typeof(FormulaValue), "It is a bug to call this method with TValue not === FormulaValue"); @@ -125,30 +152,53 @@ async ValueTask EnsureFormulaValueAsync() if (result is null or BlankValue) { - result = initialStateFactory() as FormulaValue; + bool legacyEntry = await Kit.IWorkflowContextExtensions.HasLegacyTopicEntryAsync(this.Source, key, scopeName!, forWrite: true, cancellationToken).ConfigureAwait(false); + PortableValue? legacyValue = legacyEntry ? await ReadLegacyValueAsync().ConfigureAwait(false) : null; + result = legacyValue is null ? initialStateFactory() as FormulaValue : legacyValue.ToFormula(); if (result is null) { throw new InvalidOperationException($"The initial state factory for key '{key}' in scope '{scopeName}' did not return a FormulaValue."); } this.State.Set(key, result, scopeName); - await this.Source.QueueStateUpdateAsync(key, result.AsPortable(), scopeName, cancellationToken) + await this.Source.QueueStateUpdateAsync(key, legacyValue is null ? result.AsPortable() : legacyValue.ToPortableObject(), scopeName, cancellationToken) .ConfigureAwait(false); + await CompleteInitializationAsync(legacyEntry).ConfigureAwait(false); } return (TValue)(object)result!; // The null analyzer is confused here, but it is impossible to hit this line with result is null } + + async ValueTask ReadLegacyValueAsync() => + await this.Source.ReadStateAsync(key, VariableScopeNames.Topic, cancellationToken).ConfigureAwait(false) + ?? throw new InvalidOperationException($"The existing state for key '{key}' in scope '{VariableScopeNames.Topic}' is missing."); + + async ValueTask CompleteInitializationAsync(bool legacyEntry) + { + SensitivityLevel sensitivity = legacyEntry + ? await this.Source.ReadStateAsync(key, WorkflowFormulaState.GetSensitivityScopeName(scopeName!), cancellationToken).ConfigureAwait(false) + : SensitivityLevel.None; + await this.QueueSensitivityUpdateAsync(key, scopeName, sensitivity, cancellationToken).ConfigureAwait(false); + this.State.SetSensitivity(key, scopeName, sensitivity); + if (legacyEntry) + { + await this.Source.QueueStateUpdateAsync(key, null, VariableScopeNames.Topic, cancellationToken).ConfigureAwait(false); + } + } } /// public ValueTask> ReadStateKeysAsync(string? scopeName = null, CancellationToken cancellationToken = default) - => this.Source.ReadStateKeysAsync(scopeName, cancellationToken); + { + scopeName = IsManagedScope(scopeName) ? WorkflowFormulaState.GetScopeName(scopeName) : scopeName; + return this.Source.ReadStateKeysAsync(scopeName, cancellationToken); + } /// public ValueTask SendMessageAsync(object message, string? targetId = null, CancellationToken cancellationToken = default) => this.Source.SendMessageAsync(message, targetId, cancellationToken); - public ValueTask UpdateStateAsync( + public async ValueTask UpdateStateAsync( string key, T? value, string? scopeName, @@ -164,7 +214,8 @@ scopeName is not null && // null scope cannot be managed { // Not a managed scope, just pass through. This is valid when a declarative // workflow has been ejected to code (where DeclarativeWorkflowContext is also utilized). - return this.Source.QueueStateUpdateAsync(key, value, scopeName, cancellationToken); + await this.Source.QueueStateUpdateAsync(key, value, scopeName, cancellationToken).ConfigureAwait(false); + return; } if (!ManagedScopes.Contains(scopeName!) && !allowSystem) @@ -172,15 +223,30 @@ scopeName is not null && // null scope cannot be managed throw new DeclarativeActionException($"Cannot manage variable definitions in scope: '{scopeName}'."); } - return value switch + scopeName = WorkflowFormulaState.GetScopeName(scopeName); + + bool legacyEntry = await Kit.IWorkflowContextExtensions.HasLegacyTopicEntryAsync(this.Source, key, scopeName, forWrite: true, cancellationToken).ConfigureAwait(false); + switch (value) { - null => QueueEmptyStateAsync(), - UnassignedValue => QueueEmptyStateAsync(), - BlankValue => QueueEmptyStateAsync(), - FormulaValue formulaValue => QueueFormulaStateAsync(formulaValue), - DataValue dataValue => QueueDataValueStateAsync(dataValue), - _ => QueueNativeStateAsync(value), - }; + case null: + case UnassignedValue: + case BlankValue: + await QueueEmptyStateAsync().ConfigureAwait(false); + break; + case FormulaValue formulaValue: + await QueueFormulaStateAsync(formulaValue).ConfigureAwait(false); + break; + case DataValue dataValue: + await QueueDataValueStateAsync(dataValue).ConfigureAwait(false); + break; + default: + await QueueNativeStateAsync(value).ConfigureAwait(false); + break; + } + if (legacyEntry) + { + await this.Source.QueueStateUpdateAsync(key, null, VariableScopeNames.Topic, cancellationToken).ConfigureAwait(false); + } async ValueTask QueueEmptyStateAsync() { @@ -237,6 +303,6 @@ private ValueTask QueueSensitivityUpdateAsync(string key, string? scopeName, Sen return default; } - return this.Source.QueueStateUpdateAsync(key, sensitivity, WorkflowFormulaState.GetSensitivityScopeName(scopeName), cancellationToken); + return WorkflowFormulaState.QueueSensitivityUpdateAsync(this.Source, key, scopeName, sensitivity, delete: false, cancellationToken); } } 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 1126b43d8ff..834262d2c07 100644 --- a/dotnet/src/Microsoft.Agents.AI.Workflows.Declarative/Interpreter/DeclarativeWorkflowExecutor.cs +++ b/dotnet/src/Microsoft.Agents.AI.Workflows.Declarative/Interpreter/DeclarativeWorkflowExecutor.cs @@ -32,29 +32,37 @@ namespace Microsoft.Agents.AI.Workflows.Declarative.Interpreter; /// executors), so the start executor only ever sees a single inbound batch per turn. /// /// -internal sealed class DeclarativeWorkflowExecutor( - string workflowId, - DeclarativeWorkflowOptions options, - WorkflowFormulaState state, - Func inputTransform) : - Executor(workflowId), IResettableExecutor, IModeledAction where TInput : notnull +internal sealed class DeclarativeWorkflowExecutor : + Executor, IResettableExecutor, IModeledAction where TInput : notnull { + private readonly DeclarativeWorkflowOptions _options; + private readonly WorkflowFormulaState _state; + private readonly Func _inputTransform; + + public DeclarativeWorkflowExecutor( + string workflowId, + DeclarativeWorkflowOptions options, + WorkflowFormulaState state, + Func inputTransform) : base(workflowId) + { + this._options = options; + this._state = state; + this._inputTransform = inputTransform; + this.RegisterCheckpointRestoringCallback(state, state.RestoreCheckpointAsync); + } + /// public ValueTask ResetAsync() { - state.Reset(); + this._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) { - ChatMessage input = inputTransform.Invoke(message); + ChatMessage input = this._inputTransform.Invoke(message); return this.AdvanceAsync(input, context, cancellationToken); } @@ -152,9 +160,9 @@ private ValueTask HandleTurnTokenAsync(TurnToken token, IWorkflowContext context private async ValueTask AdvanceAsync(ChatMessage input, IWorkflowContext context, CancellationToken cancellationToken, bool finalizeTurn = true) { // No state to restore if we're starting from the beginning. - state.SetInitialized(); + this._state.SetInitialized(); - DeclarativeWorkflowContext declarativeContext = new(context, state); + DeclarativeWorkflowContext declarativeContext = new(context, this._state); // Conversation id resolution prefers state already persisted by a prior turn, // so multi-turn invocations reuse the same backend conversation rather than @@ -162,13 +170,13 @@ private async ValueTask AdvanceAsync(ChatMessage input, IWorkflowContext context string? conversationId = declarativeContext.GetWorkflowConversation(); if (string.IsNullOrWhiteSpace(conversationId)) { - conversationId = options.ConversationId; + conversationId = this._options.ConversationId; } bool conversationCreated = false; if (string.IsNullOrWhiteSpace(conversationId)) { - conversationId = await options.AgentProvider.CreateConversationAsync(cancellationToken).ConfigureAwait(false); + conversationId = await this._options.AgentProvider.CreateConversationAsync(cancellationToken).ConfigureAwait(false); conversationCreated = true; } @@ -177,7 +185,7 @@ private async ValueTask AdvanceAsync(ChatMessage input, IWorkflowContext context await declarativeContext.QueueConversationUpdateAsync(conversationId!, isExternal: true, cancellationToken).ConfigureAwait(false); } - ChatMessage inputMessage = await options.AgentProvider.CreateMessageAsync(conversationId!, input, cancellationToken).ConfigureAwait(false); + ChatMessage inputMessage = await this._options.AgentProvider.CreateMessageAsync(conversationId!, input, cancellationToken).ConfigureAwait(false); // Use the original input for System.LastMessage to ensure Text is preserved (the // service may strip text on round-trip), but substitute server-side media references diff --git a/dotnet/src/Microsoft.Agents.AI.Workflows.Declarative/Kit/IWorkflowContextExtensions.cs b/dotnet/src/Microsoft.Agents.AI.Workflows.Declarative/Kit/IWorkflowContextExtensions.cs index 498840c8a9a..10b189a5cdc 100644 --- a/dotnet/src/Microsoft.Agents.AI.Workflows.Declarative/Kit/IWorkflowContextExtensions.cs +++ b/dotnet/src/Microsoft.Agents.AI.Workflows.Declarative/Kit/IWorkflowContextExtensions.cs @@ -1,5 +1,6 @@ // Copyright (c) Microsoft. All rights reserved. +using System; using System.Collections.Generic; using System.Text; using System.Threading; @@ -200,8 +201,9 @@ public static async ValueTask> FormatTemplateWithSensit return new(declarativeValue, declarativeSensitivity); } - string plainScopeName = scopeName ?? WorkflowFormulaState.DefaultScopeName; - TValue? value = await context.ReadStateAsync(key, plainScopeName, cancellationToken).ConfigureAwait(false); + string plainScopeName = GetPlainScopeName(scopeName); + bool legacyEntry = await HasLegacyTopicEntryAsync(context, key, plainScopeName, forWrite: false, cancellationToken).ConfigureAwait(false); + TValue? value = await context.ReadStateAsync(key, legacyEntry ? VariableScopeNames.Topic : plainScopeName, cancellationToken).ConfigureAwait(false); SensitivityLevel sensitivity = ShouldPersistSensitivity(plainScopeName) ? await context.ReadStateAsync(key, WorkflowFormulaState.GetSensitivityScopeName(plainScopeName), cancellationToken).ConfigureAwait(false) : SensitivityLevel.None; @@ -232,11 +234,17 @@ public static async ValueTask QueueStateUpdateWithSensitivityAsync( return; } - string plainScopeName = scopeName ?? WorkflowFormulaState.DefaultScopeName; + string plainScopeName = GetPlainScopeName(scopeName); + bool legacyEntry = await HasLegacyTopicEntryAsync(context, key, plainScopeName, forWrite: true, cancellationToken).ConfigureAwait(false); await context.QueueStateUpdateAsync(key, value.Value, plainScopeName, cancellationToken).ConfigureAwait(false); if (ShouldPersistSensitivity(plainScopeName)) { - await context.QueueStateUpdateAsync(key, value.Sensitivity, WorkflowFormulaState.GetSensitivityScopeName(plainScopeName), cancellationToken).ConfigureAwait(false); + await WorkflowFormulaState.QueueSensitivityUpdateAsync(context, key, plainScopeName, value.Sensitivity, + delete: value.Value is null, cancellationToken).ConfigureAwait(false); + } + if (legacyEntry) + { + await context.QueueStateUpdateAsync(key, null, VariableScopeNames.Topic, cancellationToken).ConfigureAwait(false); } } @@ -314,7 +322,7 @@ private static async Task GetStateAsync(this IWorkflowCont WorkflowFormulaState state = new(RecalcEngineFactory.Create()); - await state.RestoreAsync(context, cancellationToken).ConfigureAwait(false); + await state.RestoreSnapshotAsync(context, cancellationToken).ConfigureAwait(false); return state; } @@ -330,6 +338,42 @@ private static void ThrowIfSensitive(SensitivityLevel sensitivity) private static SensitivityLevel MaxSensitivity(SensitivityLevel left, SensitivityLevel right) => left == SensitivityLevel.Sensitive || right == SensitivityLevel.Sensitive ? SensitivityLevel.Sensitive : SensitivityLevel.None; + private static string GetPlainScopeName(string? scopeName) => + scopeName is null || VariableScopeNames.IsValidName(scopeName) ? WorkflowFormulaState.GetScopeName(scopeName) : scopeName; + + internal static async ValueTask HasLegacyTopicEntryAsync(IWorkflowContext context, string key, string scopeName, bool forWrite, CancellationToken cancellationToken) + { + if (!string.Equals(scopeName, VariableScopeNames.Local, StringComparison.Ordinal)) + { + return false; + } + + PortableValue? legacyValue = await context.ReadStateAsync(key, VariableScopeNames.Topic, cancellationToken).ConfigureAwait(false); + if (legacyValue is null || await WorkflowFormulaState.HasCanonicalLocalOwnershipAsync(context, key, cancellationToken).ConfigureAwait(false)) + { + return false; + } + + PortableValue? sensitivity = await context.ReadStateAsync( + key, WorkflowFormulaState.GetSensitivityScopeName(scopeName), cancellationToken).ConfigureAwait(false); + if (sensitivity is null) + { + if (forWrite) + { + throw new DeclarativeActionException("Cannot update declarative state with an existing code-first Topic entry."); + } + return false; + } + + PortableValue? localValue = await context.ReadStateAsync(key, scopeName, cancellationToken).ConfigureAwait(false); + if (localValue is not null) + { + throw new DeclarativeActionException("Cannot access state with conflicting Local and Topic scope entries."); + } + + return true; + } + private static bool ShouldPersistSensitivity(string scopeName) => DeclarativeWorkflowContext.ManagedScopes.Contains(scopeName) || scopeName == VariableScopeNames.Environment || 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 19af8f95834..e1d71416758 100644 --- a/dotnet/src/Microsoft.Agents.AI.Workflows.Declarative/PowerFx/WorkflowFormulaState.cs +++ b/dotnet/src/Microsoft.Agents.AI.Workflows.Declarative/PowerFx/WorkflowFormulaState.cs @@ -8,6 +8,7 @@ using System.Threading; using System.Threading.Tasks; using Microsoft.Agents.AI.Workflows.Declarative.Extensions; +using Microsoft.Agents.AI.Workflows.Declarative.Kit; using Microsoft.Agents.ObjectModel; using Microsoft.PowerFx; using Microsoft.PowerFx.Types; @@ -29,6 +30,8 @@ internal sealed class WorkflowFormulaState ]; private const string SensitivityScopePrefix = "__Microsoft_Agents_AI_Workflows_Declarative_Sensitivity:"; + internal const string LocalOwnershipScopeName = "__Microsoft_Agents_AI_Workflows_Declarative_Ownership:Local"; + private const int LocalOwnershipVersion = 1; private readonly Dictionary _scopes; private readonly FrozenSet _restorableEnvironmentVariables; @@ -114,30 +117,57 @@ public void Reset() this.Bind(); } - public ValueTask RestoreAsync(IWorkflowContext context, CancellationToken cancellationToken) + public async ValueTask RestoreAsync(IWorkflowContext context, CancellationToken cancellationToken) { if (!this.SetInitialized()) { - return default; + return; } - return this.RestoreCheckpointAsync(context, cancellationToken); + bool restored = false; + try + { + await this.RestoreCheckpointAsync(context, cancellationToken).ConfigureAwait(false); + restored = true; + } + finally + { + if (!restored) + { + Interlocked.Exchange(ref this._isInitialized, 0); + } + } } - public async ValueTask RestoreCheckpointAsync(IWorkflowContext context, CancellationToken cancellationToken) + public ValueTask RestoreCheckpointAsync(IWorkflowContext context, CancellationToken cancellationToken) => + this.RestoreCheckpointAsync(context, migrateLegacyEntries: true, cancellationToken); + + public ValueTask RestoreSnapshotAsync(IWorkflowContext context, CancellationToken cancellationToken) => + this.RestoreCheckpointAsync(context, migrateLegacyEntries: false, cancellationToken); + + private async ValueTask RestoreCheckpointAsync(IWorkflowContext context, bool migrateLegacyEntries, CancellationToken cancellationToken) { this.RestoreInitialState(); Stopwatch timer = Stopwatch.StartNew(); Debug.WriteLine("RESTORE CHECKPOINT - BEGIN"); + HashSet legacyKeys = await GetLegacyTopicKeysAsync(context, cancellationToken).ConfigureAwait(false); + if (migrateLegacyEntries) + { + await MigrateLegacyTopicScopeAsync(context, legacyKeys, cancellationToken).ConfigureAwait(false); + } await Task.WhenAll(RestorableScopes.Select(scopeName => ReadScopeAsync(scopeName))).ConfigureAwait(false); + if (!migrateLegacyEntries && legacyKeys.Count > 0) + { + await ReadScopeAsync(VariableScopeNames.Topic, legacyKeys).ConfigureAwait(false); + } if (this._restorableEnvironmentVariables.Count > 0) { await ReadScopeAsync(VariableScopeNames.Environment, this._restorableEnvironmentVariables).ConfigureAwait(false); } Debug.WriteLine($"RESTORE CHECKPOINT - COMPLETE [{timer.Elapsed}]"); - async Task ReadScopeAsync(string scopeName, FrozenSet? allowedKeys = null) + async Task ReadScopeAsync(string scopeName, IReadOnlyCollection? allowedKeys = null) { HashSet keys = await context.ReadStateKeysAsync(scopeName, cancellationToken).ConfigureAwait(false); foreach (string key in keys) @@ -162,6 +192,99 @@ async Task ReadScopeAsync(string scopeName, FrozenSet? allowedKeys = nul } } + private static async ValueTask> GetLegacyTopicKeysAsync(IWorkflowContext context, CancellationToken cancellationToken) + { + HashSet legacyKeys = await context.ReadStateKeysAsync(VariableScopeNames.Topic, cancellationToken).ConfigureAwait(false); + if (legacyKeys.Count == 0) + { + return legacyKeys; + } + + string sensitivityScope = GetSensitivityScopeName(VariableScopeNames.Local); + HashSet declarativeKeys = await context.ReadStateKeysAsync(sensitivityScope, cancellationToken).ConfigureAwait(false); + legacyKeys.IntersectWith(declarativeKeys); + List canonicalKeys = []; + foreach (string key in legacyKeys) + { + if (await HasCanonicalLocalOwnershipAsync(context, key, cancellationToken).ConfigureAwait(false)) + { + canonicalKeys.Add(key); + } + } + legacyKeys.ExceptWith(canonicalKeys); + if (legacyKeys.Count == 0) + { + return legacyKeys; + } + + HashSet localKeys = await context.ReadStateKeysAsync(VariableScopeNames.Local, cancellationToken).ConfigureAwait(false); + if (localKeys.Overlaps(legacyKeys)) + { + throw new DeclarativeActionException("Cannot restore checkpoint with conflicting Local and Topic scope entries."); + } + + return legacyKeys; + } + + private static async ValueTask MigrateLegacyTopicScopeAsync(IWorkflowContext context, HashSet legacyKeys, CancellationToken cancellationToken) + { + string sensitivityScope = GetSensitivityScopeName(VariableScopeNames.Local); + List<(string Key, object Value, SensitivityLevel Sensitivity)> entries = []; + foreach (string key in legacyKeys) + { + PortableValue? value = await context.ReadStateAsync(key, VariableScopeNames.Topic, cancellationToken).ConfigureAwait(false); + SensitivityLevel sensitivity = await context.ReadStateAsync(key, sensitivityScope, cancellationToken).ConfigureAwait(false); + entries.Add((key, value?.ToPortableObject() ?? UnassignedValue.Instance, sensitivity)); + } + + foreach (var entry in entries) + { + await context.QueueStateUpdateAsync(entry.Key, entry.Value, VariableScopeNames.Local, cancellationToken).ConfigureAwait(false); + await QueueSensitivityUpdateAsync(context, entry.Key, VariableScopeNames.Local, entry.Sensitivity, delete: false, cancellationToken).ConfigureAwait(false); + await context.QueueStateUpdateAsync(entry.Key, null, VariableScopeNames.Topic, cancellationToken).ConfigureAwait(false); + } + } + + internal static async ValueTask HasCanonicalLocalOwnershipAsync(IWorkflowContext context, string key, CancellationToken cancellationToken) + { + PortableValue? ownership = await context.ReadStateAsync(key, LocalOwnershipScopeName, cancellationToken).ConfigureAwait(false); + if (ownership is null) + { + return false; + } + if (ownership.As() != LocalOwnershipVersion) + { + throw new DeclarativeActionException("Unsupported declarative state ownership version."); + } + return true; + } + + internal static async ValueTask QueueSensitivityUpdateAsync( + IWorkflowContext context, string key, string scopeName, SensitivityLevel sensitivity, bool delete, CancellationToken cancellationToken) + { + string sensitivityScope = GetSensitivityScopeName(scopeName); + if (delete) + { + await context.QueueStateUpdateAsync(key, null, sensitivityScope, cancellationToken).ConfigureAwait(false); + } + else + { + await context.QueueStateUpdateAsync(key, sensitivity, sensitivityScope, cancellationToken).ConfigureAwait(false); + } + + if (string.Equals(scopeName, VariableScopeNames.Local, StringComparison.Ordinal)) + { + if (delete) + { + await context.QueueStateUpdateAsync(key, null, LocalOwnershipScopeName, cancellationToken).ConfigureAwait(false); + } + else + { + await context.QueueStateUpdateAsync(key, LocalOwnershipVersion, LocalOwnershipScopeName, cancellationToken).ConfigureAwait(false); + } + } + } + private Dictionary CreateScopeSnapshot() => this._scopes.ToDictionary(scope => scope.Key, scope => new WorkflowScope(scope.Value)); @@ -206,7 +329,7 @@ public void Bind(string? scopeNameToBind = null) void Bind(string scopeName, string? targetScope = null) { - targetScope = GetScopeName(targetScope ?? scopeName); + targetScope ??= GetScopeName(scopeName); RecordValue scopeRecord = this.GetScope(scopeName).ToRecord(); this.Engine.DeleteFormula(targetScope); this.Engine.UpdateVariable(targetScope, scopeRecord); diff --git a/dotnet/src/Microsoft.Agents.AI.Workflows/Execution/StateManager.cs b/dotnet/src/Microsoft.Agents.AI.Workflows/Execution/StateManager.cs index 5b6fb3c0bd6..c5d68c8f182 100644 --- a/dotnet/src/Microsoft.Agents.AI.Workflows/Execution/StateManager.cs +++ b/dotnet/src/Microsoft.Agents.AI.Workflows/Execution/StateManager.cs @@ -251,6 +251,14 @@ internal async ValueTask> ExportStateAsync() return this._scopes.Values.SelectMany(ExportScope).ToDictionary(kvp => kvp.Key, kvp => kvp.Value); } + internal void DiscardQueuedUpdates() => this._queuedUpdates.Clear(); + + internal ValueTask RollbackStateAsync(Checkpoint checkpoint) + { + this.DiscardQueuedUpdates(); + return this.ImportStateAsync(checkpoint); + } + internal ValueTask ImportStateAsync(Checkpoint checkpoint) { // TODO: Should this be a warning instead? diff --git a/dotnet/src/Microsoft.Agents.AI.Workflows/InProc/InProcessRunner.cs b/dotnet/src/Microsoft.Agents.AI.Workflows/InProc/InProcessRunner.cs index 69b6c1e9bcc..a8ec8bb042d 100644 --- a/dotnet/src/Microsoft.Agents.AI.Workflows/InProc/InProcessRunner.cs +++ b/dotnet/src/Microsoft.Agents.AI.Workflows/InProc/InProcessRunner.cs @@ -416,11 +416,18 @@ private async ValueTask RestoreCheckpointCoreAsync(CheckpointInfo checkpointInfo await this.RunContext.StateManager.ImportStateAsync(checkpoint).ConfigureAwait(false); await this.RunContext.ImportStateAsync(checkpoint).ConfigureAwait(false); - Task executorNotifyTask = this.RunContext.NotifyCheckpointLoadedAsync(cancellationToken); - await this.RunContext.ImportEdgeStateAsync(checkpoint.EdgeStateData).ConfigureAwait(false); - await Task.WhenAll(executorNotifyTask, - restoreCheckpointIndexTask.AsTask()).ConfigureAwait(false); + try + { + await Task.WhenAll(this.RunContext.NotifyCheckpointLoadedAsync(checkpoint, cancellationToken), + restoreCheckpointIndexTask.AsTask()).ConfigureAwait(false); + } + catch + { + // Both tasks have completed; discard hook writes without reverting published restoration state. + this.RunContext.StateManager.DiscardQueuedUpdates(); + throw; + } this._lastCheckpointInfo = checkpointInfo; this.StepTracer.Reload(checkpoint.StepNumber); diff --git a/dotnet/src/Microsoft.Agents.AI.Workflows/InProc/InProcessRunnerContext.cs b/dotnet/src/Microsoft.Agents.AI.Workflows/InProc/InProcessRunnerContext.cs index dc74be1495d..f2e4f8408ba 100644 --- a/dotnet/src/Microsoft.Agents.AI.Workflows/InProc/InProcessRunnerContext.cs +++ b/dotnet/src/Microsoft.Agents.AI.Workflows/InProc/InProcessRunnerContext.cs @@ -402,7 +402,7 @@ async Task InvokeCheckpointingAsync(Task executorTask) } } - internal async Task NotifyCheckpointLoadedAsync(CancellationToken cancellationToken = default) + internal async Task NotifyCheckpointLoadedAsync(Checkpoint checkpoint, CancellationToken cancellationToken = default) { this.CheckEnded(); @@ -427,7 +427,17 @@ internal async Task NotifyCheckpointLoadedAsync(CancellationToken cancellationTo restorationTasks.Add(RestoreStateAsync(executor, callbacks)); } // Shared state must finish restoring before any subclass post-load hook runs. - await Task.WhenAll(restorationTasks).ConfigureAwait(false); + try + { + await Task.WhenAll(restorationTasks).ConfigureAwait(false); + await this.StateManager.PublishUpdatesAsync(tracer: null).ConfigureAwait(false); + } + catch + { + // All callbacks have completed; discard failed writes before any post-load hooks run. + await this.StateManager.RollbackStateAsync(checkpoint).ConfigureAwait(false); + throw; + } await Task.WhenAll(executors.Select(InvokeCheckpointRestoredAsync)).ConfigureAwait(false); async Task RestoreStateAsync(Executor executor, List> callbacks) 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 8f9fbf34cf4..b51b50b96ce 100644 --- a/dotnet/tests/Microsoft.Agents.AI.Workflows.Declarative.UnitTests/DeclarativeWorkflowTest.cs +++ b/dotnet/tests/Microsoft.Agents.AI.Workflows.Declarative.UnitTests/DeclarativeWorkflowTest.cs @@ -5,8 +5,11 @@ using System.IO; using System.Linq; using System.Reflection; +using System.Text.Json; +using System.Text.Json.Nodes; using System.Threading; using System.Threading.Tasks; +using Microsoft.Agents.AI.Workflows.Checkpointing; using Microsoft.Agents.AI.Workflows.Declarative.Interpreter; using Microsoft.Agents.AI.Workflows.Declarative.Kit; using Microsoft.Agents.AI.Workflows.Declarative.PowerFx; @@ -30,6 +33,117 @@ public sealed class DeclarativeWorkflowTest(ITestOutputHelper output) : Workflow private Dictionary WorkflowEventCounts { get; set; } = []; + [Theory] + [InlineData(false, false)] + [InlineData(false, true)] + [InlineData(true, false)] + [InlineData(true, true)] + public async Task ResumeAsync_LegacyCheckpointPreservesNextAssignmentAsync(bool offThread, bool clear) + { + // Arrange + (Workflow workflow, CheckpointManager checkpoints, CheckpointInfo checkpoint) = + await CreateLegacyCheckpointAsync(offThread, clear, completed: false); + var environment = (offThread ? InProcessExecution.OffThread : InProcessExecution.Lockstep).WithCheckpointing(checkpoints); + + // Act + await using StreamingRun resumed = await environment.ResumeStreamingAsync(workflow, checkpoint); + await foreach (WorkflowEvent evt in resumed.WatchStreamAsync()) + { + Assert.False(evt is WorkflowErrorEvent, (evt as WorkflowErrorEvent)?.Exception?.ToString()); + } + + // Assert + WorkflowFormulaState state = GetRootState(workflow); + Assert.Equal(clear ? null : "updated", state.Get("TestValue").ToObject()); + } + + [Theory] + [InlineData(false)] + [InlineData(true)] + public async Task ResumeAsync_LegacyCheckpointSupportsImmediateRestoresAsync(bool offThread) + { + // Arrange + (Workflow workflow, CheckpointManager checkpoints, CheckpointInfo checkpoint) = + await CreateLegacyCheckpointAsync(offThread, clear: false, completed: true); + var environment = (offThread ? InProcessExecution.OffThread : InProcessExecution.Lockstep).WithCheckpointing(checkpoints); + + // Act + await using StreamingRun resumed = await environment.ResumeStreamingAsync(workflow, checkpoint); + await resumed.RestoreCheckpointAsync(checkpoint); + await resumed.RestoreCheckpointAsync(checkpoint); + + // Assert + Assert.Equal("updated", GetRootState(workflow).Get("TestValue").ToObject()); + } + + private static async Task<(Workflow Workflow, CheckpointManager Checkpoints, CheckpointInfo Checkpoint)> CreateLegacyCheckpointAsync( + bool offThread, bool clear, bool completed) + { + Dictionary stored = []; + Mock> store = new(); + store.Setup(s => s.CreateCheckpointAsync(It.IsAny(), It.IsAny(), It.IsAny())) + .ReturnsAsync((string sessionId, JsonElement value, CheckpointInfo? _) => + { + CheckpointInfo info = new(sessionId, Guid.NewGuid().ToString("N")); + stored.Add(info, value); + return info; + }); + store.Setup(s => s.RetrieveCheckpointAsync(It.IsAny(), It.IsAny())) + .ReturnsAsync((string _, CheckpointInfo info) => stored[info]); + store.Setup(s => s.RetrieveIndexAsync(It.IsAny(), It.IsAny())) + .ReturnsAsync((string sessionId, CheckpointInfo? _) => + stored.Keys.Where(info => string.Equals(info.SessionId, sessionId, StringComparison.Ordinal)).ToArray().AsEnumerable()); + CheckpointManager checkpoints = CheckpointManager.CreateJson(store.Object, DeclarativeWorkflowJsonOptions.Default); + var environment = (offThread ? InProcessExecution.OffThread : InProcessExecution.Lockstep).WithCheckpointing(checkpoints); + List saved = []; + await using (StreamingRun run = await environment.RunStreamingAsync(BuildWorkflow(), "start")) + { + await foreach (WorkflowEvent evt in run.WatchStreamAsync()) + { + Assert.False(evt is WorkflowErrorEvent, (evt as WorkflowErrorEvent)?.Exception?.ToString()); + if (evt is SuperStepCompletedEvent { CompletionInfo.Checkpoint: { } checkpoint }) + { + saved.Add(checkpoint); + } + } + } + Assert.True(saved.Count >= 3); + CheckpointInfo legacy = completed ? saved[^1] : saved.First(info => + stored[info].GetProperty("stateData").EnumerateObject().Any(entry => + entry.Name.EndsWith("|@Local|TestValue", StringComparison.Ordinal))); + JsonNode? node = JsonNode.Parse(stored[legacy].GetRawText()); + Assert.NotNull(node); + JsonObject stateData = Assert.IsType(node["stateData"]); + string localKey = Assert.Single(stateData.Select(entry => entry.Key), + key => key.EndsWith("|@Local|TestValue", StringComparison.Ordinal)); + JsonNode? value = stateData[localKey]; + Assert.True(stateData.Remove(localKey)); + stateData.Add(localKey.Replace("|@Local|", "|@Topic|"), value); + Assert.True(stateData.Remove(localKey.Replace("|@Local|", $"|@{WorkflowFormulaState.LocalOwnershipScopeName}|"))); + stored[legacy] = JsonSerializer.SerializeToElement(node); + return (BuildWorkflow(), checkpoints, legacy); + + Workflow BuildWorkflow() + { + using StringReader reader = new($$""" + kind: Workflow + trigger: + kind: OnConversationStart + id: legacy_restore + actions: + - kind: SetVariable + id: seed_value + variable: Local.TestValue + value: legacy + - kind: SetVariable + id: update_value + variable: Topic.TestValue + value: {{(clear ? "=Blank()" : "updated")}} + """); + return DeclarativeWorkflowBuilder.Build(reader, new(new MockAgentProvider().Object)); + } + } + [Theory] [InlineData("BadEmpty.yaml")] [InlineData("BadId.yaml")] diff --git a/dotnet/tests/Microsoft.Agents.AI.Workflows.Declarative.UnitTests/Kit/IWorkflowContextExtensionsTests.cs b/dotnet/tests/Microsoft.Agents.AI.Workflows.Declarative.UnitTests/Kit/IWorkflowContextExtensionsTests.cs index d26a1b6a84b..1cd0cad6cfe 100644 --- a/dotnet/tests/Microsoft.Agents.AI.Workflows.Declarative.UnitTests/Kit/IWorkflowContextExtensionsTests.cs +++ b/dotnet/tests/Microsoft.Agents.AI.Workflows.Declarative.UnitTests/Kit/IWorkflowContextExtensionsTests.cs @@ -1,7 +1,9 @@ // Copyright (c) Microsoft. All rights reserved. using System.Collections; +using System.Collections.Generic; using System.Linq; +using System.Threading; using System.Threading.Tasks; using Microsoft.Agents.AI.Workflows.Declarative.Extensions; using Microsoft.Agents.AI.Workflows.Declarative.Interpreter; @@ -16,6 +18,150 @@ namespace Microsoft.Agents.AI.Workflows.Declarative.UnitTests.Kit; public sealed class IWorkflowContextExtensionsTests { + [Theory] + [InlineData(VariableScopeNames.Topic, VariableScopeNames.Local, true)] + [InlineData(VariableScopeNames.Local, VariableScopeNames.Local, true)] + [InlineData(VariableScopeNames.Global, VariableScopeNames.Global, true)] + [InlineData("custom", "custom", false)] + [InlineData(null, null, false)] + public async Task StateAccess_UsesCanonicalScopeAsync(string? scope, string? expectedScope, bool managed) + { + // Arrange + Mock source = new(MockBehavior.Strict); + source.Setup(c => c.ReadStateAsync("TestValue", expectedScope, It.IsAny())) + .ReturnsAsync("stored"); + if (managed) + { + source.Setup(c => c.ReadStateAsync("TestValue", expectedScope, It.IsAny())) + .ReturnsAsync(new PortableValue("stored")); + } + source.Setup(c => c.ReadOrInitStateAsync("TestValue", It.IsAny>(), expectedScope, It.IsAny())) + .ReturnsAsync("stored"); + source.Setup(c => c.ReadStateKeysAsync(expectedScope, It.IsAny())) + .ReturnsAsync(["TestValue"]); + DeclarativeWorkflowContext context = new(source.Object, new(RecalcEngineFactory.Create())); + + // Act + string? value = await context.ReadStateAsync("TestValue", scope); + string initialized = await context.ReadOrInitStateAsync("TestValue", () => "initial", scope); + HashSet keys = await context.ReadStateKeysAsync(scope); + + // Assert + Assert.Equal("stored", value); + Assert.Equal("stored", initialized); + Assert.Equal("TestValue", Assert.Single(keys)); + source.VerifyAll(); + } + + [Fact] + public async Task ReadOrInitStateAsync_FormulaAliasUsesCanonicalScopeAsync() + { + // Arrange + Mock source = new(MockBehavior.Strict); + source.Setup(c => c.QueueStateUpdateAsync("TestValue", It.IsAny(), VariableScopeNames.Local, It.IsAny())) + .Returns(default(ValueTask)); + source.Setup(c => c.ReadStateAsync("TestValue", VariableScopeNames.Topic, It.IsAny())) + .ReturnsAsync((PortableValue?)null); + source.Setup(c => c.QueueStateUpdateAsync("TestValue", SensitivityLevel.None, + WorkflowFormulaState.GetSensitivityScopeName(VariableScopeNames.Local), It.IsAny())) + .Returns(default(ValueTask)); + source.Setup(c => c.QueueStateUpdateAsync("TestValue", It.IsAny(), + WorkflowFormulaState.LocalOwnershipScopeName, It.IsAny())) + .Returns(default(ValueTask)); + WorkflowFormulaState state = new(RecalcEngineFactory.Create()); + DeclarativeWorkflowContext context = new(source.Object, state); + + // Act + FormulaValue result = await context.ReadOrInitStateAsync("TestValue", () => FormulaValue.New("initial"), VariableScopeNames.Topic); + + // Assert + Assert.Equal("initial", result.ToObject()); + Assert.Equal("initial", state.Get("TestValue", VariableScopeNames.Local).ToObject()); + source.VerifyAll(); + } + + [Theory] + [InlineData(false)] + [InlineData(true)] + public async Task ReadOrInitStateAsync_RejectsUnmarkedTopicCollisionAsync(bool formula) + { + // Arrange + Mock source = new(); + source.Setup(c => c.ReadStateAsync("TestValue", VariableScopeNames.Topic, It.IsAny())) + .ReturnsAsync(new PortableValue("native")); + source.Setup(c => c.ReadOrInitStateAsync("TestValue", It.IsAny>(), VariableScopeNames.Local, It.IsAny())) + .ReturnsAsync((string _, System.Func factory, string? _, CancellationToken _) => factory()); + WorkflowFormulaState state = new(RecalcEngineFactory.Create()); + DeclarativeWorkflowContext context = new(source.Object, state); + bool factoryInvoked = false; + + // Act + async Task InitializeAsync() + { + if (formula) + { + await context.ReadOrInitStateAsync("TestValue", () => + { + factoryInvoked = true; + return FormulaValue.New(42); + }, VariableScopeNames.Topic); + } + else + { + await context.ReadOrInitStateAsync("TestValue", () => + { + factoryInvoked = true; + return 42; + }, VariableScopeNames.Topic); + } + } + + // Assert + await Assert.ThrowsAsync(InitializeAsync); + Assert.False(factoryInvoked); + Assert.IsType(state.Get("TestValue")); + Assert.DoesNotContain(source.Invocations, invocation => + string.Equals(invocation.Method.Name, nameof(IWorkflowContext.QueueStateUpdateAsync), System.StringComparison.Ordinal)); + } + + [Theory] + [InlineData(false)] + [InlineData(true)] + public async Task ReadOrInitStateAsync_ExistingValuePreservesSensitivityAsync(bool formula) + { + // Arrange + Mock source = new(); + source.Setup(c => c.ReadStateAsync("TestValue", VariableScopeNames.Local, It.IsAny())) + .ReturnsAsync(new PortableValue(42)); + source.Setup(c => c.ReadStateAsync("TestValue", VariableScopeNames.Topic, It.IsAny())) + .ReturnsAsync(new PortableValue("native")); + source.Setup(c => c.ReadOrInitStateAsync("TestValue", It.IsAny>(), VariableScopeNames.Local, It.IsAny())) + .ReturnsAsync(42); + WorkflowFormulaState state = new(RecalcEngineFactory.Create()); + state.Set("TestValue", FormulaValue.New(42), VariableScopeNames.Local, SensitivityLevel.Sensitive); + DeclarativeWorkflowContext context = new(source.Object, state); + + // Act + if (formula) + { + FormulaValue result = await context.ReadOrInitStateAsync( + "TestValue", () => throw new System.InvalidOperationException("Existing values must not be initialized."), + VariableScopeNames.Topic); + Assert.Equal(42m, result.ToObject()); + } + else + { + Assert.Equal(42, await context.ReadOrInitStateAsync( + "TestValue", () => throw new System.InvalidOperationException("Existing values must not be initialized."), + VariableScopeNames.Topic)); + } + + // Assert + Assert.Equal(SensitivityLevel.Sensitive, state.GetSensitivity("TestValue")); + Assert.DoesNotContain(source.Invocations, invocation => + string.Equals(invocation.Method.Name, nameof(IWorkflowContext.QueueStateUpdateAsync), System.StringComparison.Ordinal)); + } + [Fact] public async Task FormatTemplateAsync_WithSensitiveValue_ThrowsAsync() { @@ -91,7 +237,9 @@ public async Task QueueStateUpdateAsync_WithSensitivity_RebindsStateAsync() WorkflowFormulaState state = new(RecalcEngineFactory.Create()); state.Set("TestValue", FormulaValue.New("old-value")); state.Bind(); - DeclarativeWorkflowContext context = new(new Mock().Object, state); + Mock source = new(); + source.Setup(c => c.ReadStateAsync(It.IsAny(), VariableScopeNames.Topic, default)).ReturnsAsync((PortableValue?)null); + DeclarativeWorkflowContext context = new(source.Object, state); // Act await context.QueueStateUpdateAsync(PropertyPath.Create("Local.TestValue"), FormulaValue.New("new-value"), SensitivityLevel.Sensitive); @@ -110,6 +258,7 @@ public async Task ReadStateWithSensitivityAsync_QueuesSensitiveAssignmentAsync() state.Bind(); Mock source = new(MockBehavior.Loose); + source.Setup(c => c.ReadStateAsync(It.IsAny(), VariableScopeNames.Topic, default)).ReturnsAsync((PortableValue?)null); source .Setup(c => c.ReadStateAsync(SystemScope.Names.LastMessageText, VariableScopeNames.System, default)) .Returns(new ValueTask("secret-value")); @@ -129,6 +278,7 @@ public async Task ReadStateWithSensitivityAsync_WithPlainContext_ReadsSensitivit { // Arrange Mock context = new(MockBehavior.Loose); + context.Setup(c => c.ReadStateAsync(It.IsAny(), VariableScopeNames.Topic, default)).ReturnsAsync((PortableValue?)null); context .Setup(c => c.ReadStateAsync("TestValue", VariableScopeNames.Local, default)) .Returns(new ValueTask("secret-value")); @@ -149,12 +299,16 @@ public async Task QueueStateUpdateWithSensitivityAsync_WithPlainContext_QueuesSe { // Arrange Mock context = new(MockBehavior.Strict); + context.Setup(c => c.ReadStateAsync(It.IsAny(), VariableScopeNames.Topic, default)).ReturnsAsync((PortableValue?)null); context .Setup(c => c.QueueStateUpdateAsync("TestValue", "secret-value", VariableScopeNames.Local, default)) .Returns(default(ValueTask)); context .Setup(c => c.QueueStateUpdateAsync("TestValue", SensitivityLevel.Sensitive, WorkflowFormulaState.GetSensitivityScopeName(VariableScopeNames.Local), default)) .Returns(default(ValueTask)); + context + .Setup(c => c.QueueStateUpdateAsync("TestValue", It.IsAny(), WorkflowFormulaState.LocalOwnershipScopeName, default)) + .Returns(default(ValueTask)); // Act await context.Object.QueueStateUpdateWithSensitivityAsync("TestValue", new EvaluationResult("secret-value", SensitivityLevel.Sensitive), VariableScopeNames.Local); @@ -177,7 +331,9 @@ public async Task GeneratedForeachPattern_WithSensitiveCollection_PreservesItemS VariableScopeNames.Environment, SensitivityLevel.Sensitive); state.Bind(); - DeclarativeWorkflowContext context = new(new Mock().Object, state); + Mock source = new(); + source.Setup(c => c.ReadStateAsync(It.IsAny(), VariableScopeNames.Topic, default)).ReturnsAsync((PortableValue?)null); + DeclarativeWorkflowContext context = new(source.Object, state); // Act EvaluationResult evaluatedValue = await context.EvaluateValueWithSensitivityAsync("Env.SensitiveItems"); @@ -199,6 +355,7 @@ public async Task ConvertValueWithSensitivityAsync_WithPlainContext_PreservesSen { // Arrange Mock context = new(MockBehavior.Loose); + context.Setup(c => c.ReadStateAsync(It.IsAny(), VariableScopeNames.Topic, default)).ReturnsAsync((PortableValue?)null); context .Setup(c => c.ReadStateAsync("TestValue", VariableScopeNames.Local, default)) .Returns(new ValueTask(new PortableValue("42"))); diff --git a/dotnet/tests/Microsoft.Agents.AI.Workflows.Declarative.UnitTests/PowerFx/WorkflowFormulaStateTests.cs b/dotnet/tests/Microsoft.Agents.AI.Workflows.Declarative.UnitTests/PowerFx/WorkflowFormulaStateTests.cs index 34db6f294eb..c1f1393f7c5 100644 --- a/dotnet/tests/Microsoft.Agents.AI.Workflows.Declarative.UnitTests/PowerFx/WorkflowFormulaStateTests.cs +++ b/dotnet/tests/Microsoft.Agents.AI.Workflows.Declarative.UnitTests/PowerFx/WorkflowFormulaStateTests.cs @@ -2,8 +2,13 @@ using System; using System.Collections.Generic; +using System.Linq; +using System.Text.Json; using System.Threading; using System.Threading.Tasks; +using Microsoft.Agents.AI.Workflows.Checkpointing; +using Microsoft.Agents.AI.Workflows.Declarative.Interpreter; +using Microsoft.Agents.AI.Workflows.Declarative.Kit; using Microsoft.Agents.AI.Workflows.Declarative.PowerFx; using Microsoft.Agents.ObjectModel; using Microsoft.Agents.ObjectModel.Abstractions; @@ -182,4 +187,1231 @@ public async Task RestoreAsync_RestoresPersistedSensitivityAsync() // Assert Assert.Equal(SensitivityLevel.Sensitive, this.State.GetSensitivity("secret")); } + + [Theory] + [InlineData(VariableScopeNames.Local, VariableScopeNames.Topic, false, false)] + [InlineData(VariableScopeNames.Topic, VariableScopeNames.Local, false, false)] + [InlineData(VariableScopeNames.Local, VariableScopeNames.Topic, true, false)] + [InlineData(VariableScopeNames.Topic, VariableScopeNames.Local, true, false)] + [InlineData(VariableScopeNames.Local, VariableScopeNames.Topic, false, true)] + [InlineData(VariableScopeNames.Topic, VariableScopeNames.Local, false, true)] + public async Task RestoreAsync_PreservesLatestAliasAssignmentAsync(string firstScope, string secondScope, bool sensitive, bool clear) + { + // Arrange + Dictionary<(string? Scope, string Key), PortableValue> values = []; + Dictionary<(string? Scope, string Key), SensitivityLevel> sensitivities = []; + Mock source = CreateStateContext(values, sensitivities); + DeclarativeWorkflowContext context = new(source.Object, this.State); + await context.QueueStateUpdateAsync("TestValue", FormulaValue.New("first"), firstScope, SensitivityLevel.Sensitive); + if (clear) + { + await context.QueueClearScopeAsync(secondScope); + } + else + { + await context.QueueStateUpdateAsync("TestValue", "second", secondScope, sensitive ? SensitivityLevel.Sensitive : SensitivityLevel.None); + } + object? expectedValue = this.State.Get("TestValue").ToObject(); + SensitivityLevel expectedSensitivity = this.State.GetSensitivity("TestValue"); + + // Act + WorkflowFormulaState restored = new(RecalcEngineFactory.Create()); + await restored.RestoreAsync(source.Object, CancellationToken.None); + + // Assert + Assert.Equal(clear ? null : "second", expectedValue); + Assert.Equal(sensitive ? SensitivityLevel.Sensitive : SensitivityLevel.None, expectedSensitivity); + Assert.Equal(expectedValue, restored.Get("TestValue", VariableScopeNames.Local).ToObject()); + Assert.Equal(expectedValue, restored.Get("TestValue", VariableScopeNames.Topic).ToObject()); + Assert.Equal(expectedSensitivity, restored.GetSensitivity("TestValue", VariableScopeNames.Local)); + Assert.Equal(expectedSensitivity, restored.GetSensitivity("TestValue", VariableScopeNames.Topic)); + } + + [Theory] + [InlineData(false, false)] + [InlineData(true, false)] + [InlineData(false, true)] + public async Task RestoreAsync_MigratesLegacyTopicStateAsync(bool sensitive, bool clear) + { + // Arrange + string sensitivityScope = WorkflowFormulaState.GetSensitivityScopeName(VariableScopeNames.Local); + SensitivityLevel sensitivity = sensitive ? SensitivityLevel.Sensitive : SensitivityLevel.None; + Dictionary<(string? Scope, string Key), PortableValue> values = new() + { + [(VariableScopeNames.Local, "OtherValue")] = new("other"), + [(VariableScopeNames.Topic, "TestValue")] = new(clear ? UnassignedValue.Instance : "legacy"), + }; + Dictionary<(string? Scope, string Key), SensitivityLevel> sensitivities = new() + { + [(sensitivityScope, "TestValue")] = sensitivity, + }; + Mock source = CreateStateContext(values, sensitivities); + DeclarativeWorkflowContext context = new(source.Object, this.State); + + // Act + await this.State.RestoreAsync(source.Object, CancellationToken.None); + + // Assert + Assert.Equal(clear ? null : "legacy", this.State.Get("TestValue").ToObject()); + Assert.Equal(sensitivity, this.State.GetSensitivity("TestValue")); + Assert.Equal("other", this.State.Get("OtherValue").ToObject()); + Assert.DoesNotContain(values.Keys, key => string.Equals(key.Scope, VariableScopeNames.Topic, StringComparison.Ordinal)); + Assert.Contains((VariableScopeNames.Local, "TestValue"), values.Keys); + Assert.Equal(sensitivity, sensitivities[(sensitivityScope, "TestValue")]); + if (!clear) + { + Assert.Equal("legacy", await context.ReadStateAsync("TestValue", VariableScopeNames.Topic)); + Assert.Equal("legacy", await context.ReadOrInitStateAsync("TestValue", () => "replacement", VariableScopeNames.Topic)); + } + + await context.QueueClearScopeAsync(VariableScopeNames.Topic); + await this.State.RestoreCheckpointAsync(source.Object, CancellationToken.None); + Assert.IsType(this.State.Get("TestValue")); + Assert.Equal(SensitivityLevel.None, this.State.GetSensitivity("TestValue")); + } + + [Fact] + public async Task RestoreAsync_RejectsConflictingAliasEntriesOnEveryAttemptAsync() + { + // Arrange + Dictionary<(string? Scope, string Key), PortableValue> values = new() + { + [(VariableScopeNames.Topic, "OtherValue")] = new("other"), + [(VariableScopeNames.Local, "TestValue")] = new("first"), + [(VariableScopeNames.Topic, "TestValue")] = new("second"), + }; + Dictionary<(string? Scope, string Key), SensitivityLevel> sensitivities = new() + { + [(WorkflowFormulaState.GetSensitivityScopeName(VariableScopeNames.Local), "TestValue")] = SensitivityLevel.None, + }; + Mock source = CreateStateContext(values, sensitivities); + + // Act + DeclarativeActionException first = await Assert.ThrowsAsync( + async () => await this.State.RestoreAsync(source.Object, CancellationToken.None)); + DeclarativeActionException retry = await Assert.ThrowsAsync( + async () => await this.State.RestoreAsync(source.Object, CancellationToken.None)); + + // Assert + Assert.Contains("conflicting", first.Message); + Assert.Equal(first.Message, retry.Message); + Assert.Equal(3, values.Count); + Assert.Equal(SensitivityLevel.None, Assert.Single(sensitivities).Value); + source.Verify(c => c.QueueStateUpdateAsync(It.IsAny(), It.IsAny(), It.IsAny(), It.IsAny()), Times.Never); + source.Verify(c => c.QueueClearScopeAsync(It.IsAny(), It.IsAny()), Times.Never); + } + + [Theory] + [InlineData(42)] + [InlineData(42L)] + [InlineData(1.5f)] + [InlineData(1.5d)] + public async Task RestoreAsync_LegacyMigrationPreservesNativeTypesAsync(object value) + { + // Arrange + Dictionary<(string? Scope, string Key), PortableValue> values = new() + { + [(VariableScopeNames.Topic, "TestValue")] = new(value), + }; + Mock source = CreateStateContext(values, new() + { + [(WorkflowFormulaState.GetSensitivityScopeName(VariableScopeNames.Local), "TestValue")] = SensitivityLevel.None, + }); + + // Act + await this.State.RestoreAsync(source.Object, CancellationToken.None); + + // Assert + PortableValue migrated = values[(VariableScopeNames.Local, "TestValue")]; + Assert.True(migrated.TypeId.IsMatch(value.GetType())); + Assert.Equal(value, migrated.AsType(value.GetType())); + } + + [Theory] + [InlineData(false, false)] + [InlineData(false, true)] + [InlineData(true, false)] + [InlineData(true, true)] + public async Task ResumeAsync_PublishesLegacyMigrationBeforeNextExecutorAsync(bool offThread, bool clear) + { + // Arrange + StateExecutor source = new("source", SeedLegacyStateAsync, this.State); + StateExecutor writer = new("writer", async (context, cancellationToken) => + { + DeclarativeWorkflowContext declarative = new(context, this.State); + if (clear) + { + await declarative.QueueClearScopeAsync(VariableScopeNames.Topic, cancellationToken); + } + else + { + await declarative.QueueStateUpdateAsync("TestValue", "updated", VariableScopeNames.Topic, cancellationToken); + } + }); + Workflow workflow = new WorkflowBuilder(source).AddEdge(source, writer).Build(); + CheckpointManager checkpoints = CheckpointManager.CreateInMemory(); + var environment = (offThread ? InProcessExecution.OffThread : InProcessExecution.Lockstep).WithCheckpointing(checkpoints); + CheckpointInfo? legacyCheckpoint = null; + await using (StreamingRun run = await environment.RunStreamingAsync(workflow, 1)) + { + await foreach (WorkflowEvent evt in run.WatchStreamAsync()) + { + if (evt is SuperStepCompletedEvent { CompletionInfo.Checkpoint: { } saved }) + { + legacyCheckpoint ??= saved; + } + } + } + Assert.NotNull(legacyCheckpoint); + + // Act + CheckpointInfo? updatedCheckpoint = null; + await using StreamingRun resumed = await environment.ResumeStreamingAsync(workflow, legacyCheckpoint); + await foreach (WorkflowEvent evt in resumed.WatchStreamAsync()) + { + Assert.False(evt is WorkflowErrorEvent, (evt as WorkflowErrorEvent)?.Exception?.ToString()); + if (evt is SuperStepCompletedEvent { CompletionInfo.Checkpoint: { } saved }) + { + updatedCheckpoint = saved; + } + } + Assert.NotNull(updatedCheckpoint); + await resumed.RestoreCheckpointAsync(updatedCheckpoint); + + // Assert + Assert.Equal(clear ? null : "updated", this.State.Get("TestValue").ToObject()); + Assert.Equal(SensitivityLevel.None, this.State.GetSensitivity("TestValue")); + } + + [Theory] + [InlineData(false)] + [InlineData(true)] + public async Task ResumeAsync_PublishesLegacyMigrationBeforePostLoadHooksAsync(bool offThread) + { + // Arrange + StateExecutor source = new("source", SeedLegacyStateAsync, this.State); + bool hookInvoked = false; + StateExecutor reader = new("reader", (_, _) => default, restored: async (context, cancellationToken) => + { + Assert.Equal("legacy", await context.ReadStateAsync("TestValue", VariableScopeNames.Local, cancellationToken)); + Assert.Equal(SensitivityLevel.Sensitive, await context.ReadStateAsync( + "TestValue", WorkflowFormulaState.GetSensitivityScopeName(VariableScopeNames.Local), cancellationToken)); + Assert.Empty(await context.ReadStateKeysAsync(VariableScopeNames.Topic, cancellationToken)); + hookInvoked = true; + }); + Workflow workflow = new WorkflowBuilder(source).AddEdge(source, reader).Build(); + CheckpointManager checkpoints = CheckpointManager.CreateInMemory(); + var environment = (offThread ? InProcessExecution.OffThread : InProcessExecution.Lockstep).WithCheckpointing(checkpoints); + CheckpointInfo? legacyCheckpoint = null; + await using (StreamingRun run = await environment.RunStreamingAsync(workflow, 1)) + { + await foreach (WorkflowEvent evt in run.WatchStreamAsync()) + { + if (evt is SuperStepCompletedEvent { CompletionInfo.Checkpoint: { } saved }) + { + legacyCheckpoint ??= saved; + } + } + } + Assert.NotNull(legacyCheckpoint); + + // Act + await using StreamingRun resumed = await environment.ResumeStreamingAsync(workflow, legacyCheckpoint); + await resumed.RestoreCheckpointAsync(legacyCheckpoint); + await resumed.RestoreCheckpointAsync(legacyCheckpoint); + + // Assert + Assert.True(hookInvoked); + } + + [Theory] + [InlineData(false)] + [InlineData(true)] + public async Task ResumeAsync_PreservesUnmarkedTopicEntriesAsync(bool offThread) + { + // Arrange + StateExecutor source = new("source", async (context, cancellationToken) => + { + await SeedLegacyStateAsync(context, cancellationToken); + await context.QueueStateUpdateAsync("NativeValue", 42, VariableScopeNames.Topic, cancellationToken); + await context.QueueStateUpdateAsync("NativeValue", 7, VariableScopeNames.Local, cancellationToken); + }, this.State); + bool hookInvoked = false; + StateExecutor reader = new("reader", (_, _) => default, restored: async (context, cancellationToken) => + { + Assert.Equal(42, await context.ReadStateAsync("NativeValue", VariableScopeNames.Topic, cancellationToken)); + Assert.Equal(7, await context.ReadStateAsync("NativeValue", VariableScopeNames.Local, cancellationToken)); + Assert.Equal("legacy", await context.ReadStateAsync("TestValue", VariableScopeNames.Local, cancellationToken)); + Assert.Equal("NativeValue", Assert.Single(await context.ReadStateKeysAsync(VariableScopeNames.Topic, cancellationToken))); + hookInvoked = true; + }); + Workflow workflow = new WorkflowBuilder(source).AddEdge(source, reader).Build(); + CheckpointManager checkpoints = CheckpointManager.CreateInMemory(); + var environment = (offThread ? InProcessExecution.OffThread : InProcessExecution.Lockstep).WithCheckpointing(checkpoints); + CheckpointInfo? legacyCheckpoint = null; + await using (StreamingRun run = await environment.RunStreamingAsync(workflow, 1)) + { + await foreach (WorkflowEvent evt in run.WatchStreamAsync()) + { + if (evt is SuperStepCompletedEvent { CompletionInfo.Checkpoint: { } saved }) + { + legacyCheckpoint ??= saved; + } + } + } + Assert.NotNull(legacyCheckpoint); + + // Act + await using StreamingRun resumed = await environment.ResumeStreamingAsync(workflow, legacyCheckpoint); + + // Assert + Assert.True(hookInvoked); + } + + [Theory] + [InlineData(false, false)] + [InlineData(false, true)] + [InlineData(true, false)] + [InlineData(true, true)] + public async Task ResumeAsync_PlainSensitivityAliasesRemainConsistentAsync(bool offThread, bool overwrite) + { + // Arrange + bool resuming = false; + StateExecutor source = new("source", (context, cancellationToken) => + context.QueueStateUpdateWithSensitivityAsync("TestValue", + new EvaluationResult("initial", SensitivityLevel.Sensitive), VariableScopeNames.Topic, cancellationToken), this.State); + string expected = overwrite ? "updated" : "initial"; + SensitivityLevel expectedSensitivity = overwrite ? SensitivityLevel.None : SensitivityLevel.Sensitive; + StateExecutor writer = new("writer", async (context, cancellationToken) => + { + if (!resuming) + { + return; + } + if (overwrite) + { + await context.QueueStateUpdateWithSensitivityAsync("TestValue", + new EvaluationResult("updated", SensitivityLevel.None), VariableScopeNames.Topic, cancellationToken); + } + EvaluationResult result = await context.ReadStateWithSensitivityAsync("TestValue", VariableScopeNames.Topic, cancellationToken); + Assert.Equal(expected, result.Value); + Assert.Equal(expectedSensitivity, result.Sensitivity); + }); + Workflow workflow = new WorkflowBuilder(source).AddEdge(source, writer).Build(); + CheckpointManager checkpoints = CheckpointManager.CreateInMemory(); + var environment = (offThread ? InProcessExecution.OffThread : InProcessExecution.Lockstep).WithCheckpointing(checkpoints); + CheckpointInfo? checkpoint = null; + await using (StreamingRun run = await environment.RunStreamingAsync(workflow, 1)) + { + await foreach (WorkflowEvent evt in run.WatchStreamAsync()) + { + Assert.False(evt is WorkflowErrorEvent, (evt as WorkflowErrorEvent)?.Exception?.ToString()); + if (evt is SuperStepCompletedEvent { CompletionInfo.Checkpoint: { } saved }) + { + checkpoint ??= saved; + } + } + } + Assert.NotNull(checkpoint); + + // Act + resuming = true; + CheckpointInfo? updatedCheckpoint = null; + await using StreamingRun resumed = await environment.ResumeStreamingAsync(workflow, checkpoint); + await foreach (WorkflowEvent evt in resumed.WatchStreamAsync()) + { + Assert.False(evt is WorkflowErrorEvent, (evt as WorkflowErrorEvent)?.Exception?.ToString()); + if (evt is SuperStepCompletedEvent { CompletionInfo.Checkpoint: { } saved }) + { + updatedCheckpoint = saved; + } + } + Assert.NotNull(updatedCheckpoint); + await resumed.RestoreCheckpointAsync(updatedCheckpoint); + + // Assert + Assert.Equal(expected, this.State.Get("TestValue").ToObject()); + Assert.Equal(expectedSensitivity, this.State.GetSensitivity("TestValue")); + } + + [Theory] + [InlineData(false, false)] + [InlineData(false, true)] + [InlineData(true, false)] + [InlineData(true, true)] + public async Task EvaluateAsync_PlainSensitivityAliasesRemainConsistentAsync(bool offThread, bool overwrite) + { + // Arrange + StateExecutor source = new("source", async (context, cancellationToken) => + { + await context.QueueStateUpdateWithSensitivityAsync("TestValue", + new EvaluationResult("initial", SensitivityLevel.Sensitive), VariableScopeNames.Topic, cancellationToken); + EvaluationResult initial = await context.EvaluateValueWithSensitivityAsync("Local.TestValue", cancellationToken); + Assert.Equal("initial", initial.Value); + Assert.Equal(SensitivityLevel.Sensitive, initial.Sensitivity); + if (overwrite) + { + await context.QueueStateUpdateWithSensitivityAsync("TestValue", + new EvaluationResult("updated", SensitivityLevel.None), VariableScopeNames.Topic, cancellationToken); + } + + EvaluationResult read = await context.ReadStateWithSensitivityAsync("TestValue", VariableScopeNames.Topic, cancellationToken); + EvaluationResult evaluated = await context.EvaluateValueWithSensitivityAsync("Local.TestValue", cancellationToken); + + // Assert + Assert.Equal(overwrite ? "updated" : "initial", read.Value); + Assert.Equal(overwrite ? SensitivityLevel.None : SensitivityLevel.Sensitive, read.Sensitivity); + Assert.Equal(read.Value, evaluated.Value); + Assert.Equal(read.Sensitivity, evaluated.Sensitivity); + }); + Workflow workflow = new WorkflowBuilder(source).Build(); + var environment = offThread ? InProcessExecution.OffThread : InProcessExecution.Lockstep; + + // Act + await using StreamingRun run = await environment.RunStreamingAsync(workflow, 1); + await foreach (WorkflowEvent evt in run.WatchStreamAsync()) + { + Assert.False(evt is WorkflowErrorEvent, (evt as WorkflowErrorEvent)?.Exception?.ToString()); + } + } + + [Theory] + [InlineData(false, false, SensitivityLevel.None)] + [InlineData(false, false, SensitivityLevel.Sensitive)] + [InlineData(false, true, SensitivityLevel.None)] + [InlineData(false, true, SensitivityLevel.Sensitive)] + [InlineData(true, false, SensitivityLevel.None)] + [InlineData(true, false, SensitivityLevel.Sensitive)] + [InlineData(true, true, SensitivityLevel.None)] + [InlineData(true, true, SensitivityLevel.Sensitive)] + public async Task ResumeAsync_PlainLegacyHelpersWorkWithoutFormulaCallbacksAsync(bool offThread, bool overwrite, SensitivityLevel sensitivity) + { + // Arrange + bool resuming = false; + IWorkflowContext? resumedContext = null; + string expected = overwrite ? "updated" : "legacy"; + SensitivityLevel expectedSensitivity = overwrite + ? sensitivity == SensitivityLevel.Sensitive ? SensitivityLevel.None : SensitivityLevel.Sensitive + : sensitivity; + StateExecutor source = new("source", async (context, cancellationToken) => + { + await context.QueueStateUpdateAsync("TestValue", "legacy", VariableScopeNames.Topic, cancellationToken); + await context.QueueStateUpdateAsync("TestValue", sensitivity, + WorkflowFormulaState.GetSensitivityScopeName(VariableScopeNames.Local), cancellationToken); + await context.QueueStateUpdateAsync("OtherValue", "untouched", VariableScopeNames.Topic, cancellationToken); + await context.QueueStateUpdateAsync("OtherValue", SensitivityLevel.None, + WorkflowFormulaState.GetSensitivityScopeName(VariableScopeNames.Local), cancellationToken); + await context.QueueStateUpdateAsync("NativeValue", 42, VariableScopeNames.Topic, cancellationToken); + }); + StateExecutor writer = new("writer", async (context, cancellationToken) => + { + if (!resuming) + { + return; + } + resumedContext = context; + if (overwrite) + { + await context.QueueStateUpdateWithSensitivityAsync("TestValue", + new EvaluationResult("updated", expectedSensitivity), VariableScopeNames.Topic, cancellationToken); + } + foreach (string? scope in new string?[] { null, VariableScopeNames.Local, VariableScopeNames.Topic }) + { + EvaluationResult result = await context.ReadStateWithSensitivityAsync("TestValue", scope, cancellationToken); + Assert.Equal(expected, result.Value); + Assert.Equal(expectedSensitivity, result.Sensitivity); + } + if (!overwrite) + { + Assert.Equal("legacy", await context.ReadStateAsync("TestValue", VariableScopeNames.Topic, cancellationToken)); + Assert.DoesNotContain("TestValue", await context.ReadStateKeysAsync(VariableScopeNames.Local, cancellationToken)); + } + Assert.Equal("untouched", await context.ReadStateAsync("OtherValue", VariableScopeNames.Topic, cancellationToken)); + Assert.DoesNotContain("OtherValue", await context.ReadStateKeysAsync(VariableScopeNames.Local, cancellationToken)); + EvaluationResult evaluated = await context.EvaluateValueWithSensitivityAsync("Local.TestValue", cancellationToken); + Assert.Equal(expected, evaluated.Value); + Assert.Equal(expectedSensitivity, evaluated.Sensitivity); + }); + Workflow workflow = new WorkflowBuilder(source).AddEdge(source, writer).Build(); + CheckpointManager checkpoints = CheckpointManager.CreateInMemory(); + var environment = (offThread ? InProcessExecution.OffThread : InProcessExecution.Lockstep).WithCheckpointing(checkpoints); + CheckpointInfo? checkpoint = null; + await using (StreamingRun run = await environment.RunStreamingAsync(workflow, 1)) + { + await foreach (WorkflowEvent evt in run.WatchStreamAsync()) + { + Assert.False(evt is WorkflowErrorEvent, (evt as WorkflowErrorEvent)?.Exception?.ToString()); + if (evt is SuperStepCompletedEvent { CompletionInfo.Checkpoint: { } saved }) + { + checkpoint ??= saved; + } + } + } + Assert.NotNull(checkpoint); + + // Act + resuming = true; + CheckpointInfo? updatedCheckpoint = null; + await using StreamingRun resumed = await environment.ResumeStreamingAsync(workflow, checkpoint); + await foreach (WorkflowEvent evt in resumed.WatchStreamAsync()) + { + Assert.False(evt is WorkflowErrorEvent, (evt as WorkflowErrorEvent)?.Exception?.ToString()); + if (evt is SuperStepCompletedEvent { CompletionInfo.Checkpoint: { } saved }) + { + updatedCheckpoint = saved; + } + } + Assert.NotNull(updatedCheckpoint); + await resumed.RestoreCheckpointAsync(updatedCheckpoint); + + // Assert + Assert.NotNull(resumedContext); + EvaluationResult restored = await resumedContext.ReadStateWithSensitivityAsync("TestValue", VariableScopeNames.Topic); + Assert.Equal(expected, restored.Value); + Assert.Equal(expectedSensitivity, restored.Sensitivity); + if (overwrite) + { + Assert.DoesNotContain("TestValue", await resumedContext.ReadStateKeysAsync(VariableScopeNames.Topic)); + } + else + { + Assert.Contains("TestValue", await resumedContext.ReadStateKeysAsync(VariableScopeNames.Topic)); + } + Assert.Equal(42, await resumedContext.ReadStateAsync("NativeValue", VariableScopeNames.Topic)); + } + + [Theory] + [InlineData(false)] + [InlineData(true)] + public async Task PlainLegacyHelpers_RejectOverlappingMarkedEntriesAsync(bool overwrite) + { + // Arrange + Dictionary<(string? Scope, string Key), PortableValue> values = new() + { + [(VariableScopeNames.Local, "TestValue")] = new("local"), + [(VariableScopeNames.Topic, "TestValue")] = new("legacy"), + }; + Dictionary<(string? Scope, string Key), SensitivityLevel> sensitivities = new() + { + [(WorkflowFormulaState.GetSensitivityScopeName(VariableScopeNames.Local), "TestValue")] = SensitivityLevel.None, + }; + IWorkflowContext context = CreateStateContext(values, sensitivities).Object; + + // Act + DeclarativeActionException error = await Assert.ThrowsAsync(async () => + { + if (overwrite) + { + await context.QueueStateUpdateWithSensitivityAsync("TestValue", + new EvaluationResult("updated", SensitivityLevel.Sensitive), VariableScopeNames.Topic); + } + else + { + await context.ReadStateWithSensitivityAsync("TestValue", VariableScopeNames.Topic); + } + }); + + // Assert + Assert.Contains("conflicting Local and Topic", error.Message); + Assert.Equal("local", values[(VariableScopeNames.Local, "TestValue")].As()); + Assert.Equal("legacy", values[(VariableScopeNames.Topic, "TestValue")].As()); + Assert.Equal(SensitivityLevel.None, Assert.Single(sensitivities).Value); + } + + [Theory] + [InlineData(false, SensitivityLevel.None)] + [InlineData(false, SensitivityLevel.Sensitive)] + [InlineData(true, SensitivityLevel.None)] + [InlineData(true, SensitivityLevel.Sensitive)] + public async Task LegacyBoundary_ParallelExpressionReadersDoNotWriteStateAsync(bool offThread, SensitivityLevel sensitivity) + { + // Arrange + bool resuming = false; + int reads = 0; + IWorkflowContext? resumedContext = null; + StateExecutor source = new("source", async (context, cancellationToken) => + { + await context.QueueStateUpdateAsync("TestValue", "legacy", VariableScopeNames.Topic, cancellationToken); + await context.QueueStateUpdateAsync("TestValue", sensitivity, + WorkflowFormulaState.GetSensitivityScopeName(VariableScopeNames.Local), cancellationToken); + }); + StateExecutor first = new("first", ReadExpressionAsync); + StateExecutor second = new("second", ReadExpressionAsync); + Workflow workflow = new WorkflowBuilder(source).AddEdge(source, first).AddEdge(source, second).Build(); + CheckpointManager checkpoints = CheckpointManager.CreateInMemory(); + var environment = (offThread ? InProcessExecution.OffThread : InProcessExecution.Lockstep).WithCheckpointing(checkpoints); + CheckpointInfo? checkpoint = null; + await using (StreamingRun run = await environment.RunStreamingAsync(workflow, 1)) + { + await foreach (WorkflowEvent evt in run.WatchStreamAsync()) + { + Assert.False(evt is WorkflowErrorEvent, (evt as WorkflowErrorEvent)?.Exception?.ToString()); + if (evt is SuperStepCompletedEvent { CompletionInfo.Checkpoint: { } saved }) + { + checkpoint ??= saved; + } + } + } + Assert.NotNull(checkpoint); + + // Act + resuming = true; + await using StreamingRun resumed = await environment.ResumeStreamingAsync(workflow, checkpoint); + await foreach (WorkflowEvent evt in resumed.WatchStreamAsync()) + { + Assert.False(evt is WorkflowErrorEvent, (evt as WorkflowErrorEvent)?.Exception?.ToString()); + } + + // Assert + Assert.Equal(2, reads); + Assert.NotNull(resumedContext); + Assert.Equal("legacy", await resumedContext.ReadStateAsync("TestValue", VariableScopeNames.Topic)); + Assert.DoesNotContain("TestValue", await resumedContext.ReadStateKeysAsync(VariableScopeNames.Local)); + Assert.Equal(sensitivity, await resumedContext.ReadStateAsync("TestValue", + WorkflowFormulaState.GetSensitivityScopeName(VariableScopeNames.Local))); + + async ValueTask ReadExpressionAsync(IWorkflowContext context, CancellationToken cancellationToken) + { + if (!resuming) + { + return; + } + EvaluationResult result = await context.EvaluateValueWithSensitivityAsync("Local.TestValue", cancellationToken); + Assert.Equal("legacy", result.Value); + Assert.Equal(sensitivity, result.Sensitivity); + resumedContext = context; + Interlocked.Increment(ref reads); + } + } + + [Theory] + [InlineData(false, null)] + [InlineData(false, VariableScopeNames.Local)] + [InlineData(false, VariableScopeNames.Topic)] + [InlineData(true, null)] + [InlineData(true, VariableScopeNames.Local)] + [InlineData(true, VariableScopeNames.Topic)] + public async Task LegacyBoundary_CanonicalWritesRejectUnmarkedTopicCollisionAsync(bool wrapped, string? scopeName) + { + // Arrange + Dictionary<(string? Scope, string Key), PortableValue> values = new() + { + [(VariableScopeNames.Local, "TestValue")] = new("existing"), + [(VariableScopeNames.Topic, "TestValue")] = new("native"), + }; + Dictionary<(string? Scope, string Key), SensitivityLevel> sensitivities = []; + IWorkflowContext source = CreateStateContext(values, sensitivities).Object; + WorkflowFormulaState state = new(RecalcEngineFactory.Create()); + state.Set("TestValue", FormulaValue.New("existing")); + state.Bind(); + IWorkflowContext context = wrapped ? new DeclarativeWorkflowContext(source, state) : source; + + // Act + DeclarativeActionException error = await Assert.ThrowsAsync(async () => + await context.QueueStateUpdateWithSensitivityAsync("TestValue", + new EvaluationResult("updated", SensitivityLevel.Sensitive), scopeName)); + + // Assert + Assert.Contains("Topic", error.Message); + Assert.Equal("existing", values[(VariableScopeNames.Local, "TestValue")].As()); + Assert.Equal("native", values[(VariableScopeNames.Topic, "TestValue")].As()); + Assert.Empty(sensitivities); + Assert.Equal("existing", state.Get("TestValue").ToObject()); + Assert.Equal(SensitivityLevel.None, state.GetSensitivity("TestValue")); + } + + [Theory] + [InlineData(false, SensitivityLevel.None)] + [InlineData(false, SensitivityLevel.Sensitive)] + [InlineData(true, SensitivityLevel.None)] + [InlineData(true, SensitivityLevel.Sensitive)] + public async Task LegacyBoundary_PlainClearDoesNotClaimReusedTopicKeyAsync(bool overwrite, SensitivityLevel sensitivity) + { + // Arrange + Dictionary<(string? Scope, string Key), PortableValue> values = new() + { + [(VariableScopeNames.Local, "TestValue")] = new("before"), + }; + Dictionary<(string? Scope, string Key), SensitivityLevel> sensitivities = new() + { + [(WorkflowFormulaState.GetSensitivityScopeName(VariableScopeNames.Local), "TestValue")] = sensitivity, + }; + IWorkflowContext context = CreateStateContext(values, sensitivities).Object; + await context.QueueStateUpdateWithSensitivityAsync("TestValue", + new EvaluationResult(null, sensitivity), VariableScopeNames.Topic); + values[(VariableScopeNames.Topic, "TestValue")] = new("native"); + + // Act + if (overwrite) + { + await Assert.ThrowsAsync(async () => + await context.QueueStateUpdateWithSensitivityAsync("TestValue", + new EvaluationResult("updated", SensitivityLevel.Sensitive), VariableScopeNames.Topic)); + } + else + { + EvaluationResult result = await context.ReadStateWithSensitivityAsync("TestValue"); + Assert.Null(result.Value); + Assert.Equal(SensitivityLevel.None, result.Sensitivity); + } + + // Assert + Assert.Equal("native", values[(VariableScopeNames.Topic, "TestValue")].As()); + Assert.False(values.ContainsKey((VariableScopeNames.Local, "TestValue"))); + Assert.Empty(sensitivities); + } + + [Theory] + [InlineData(false, false, SensitivityLevel.None)] + [InlineData(false, false, SensitivityLevel.Sensitive)] + [InlineData(false, true, SensitivityLevel.None)] + [InlineData(false, true, SensitivityLevel.Sensitive)] + [InlineData(true, false, SensitivityLevel.None)] + [InlineData(true, false, SensitivityLevel.Sensitive)] + [InlineData(true, true, SensitivityLevel.None)] + [InlineData(true, true, SensitivityLevel.Sensitive)] + public async Task CanonicalOwnership_MixedFanOutWritesSurviveFreshRestoreAsync(bool offThread, bool wrapped, SensitivityLevel sensitivity) + { + // Arrange + IWorkflowContext? writerContext = null; + WorkflowFormulaState state = new(RecalcEngineFactory.Create()); + Workflow workflow = CreateWorkflow(state); + CheckpointManager checkpoints = CreateJsonCheckpointManager(); + var environment = (offThread ? InProcessExecution.OffThread : InProcessExecution.Lockstep).WithCheckpointing(checkpoints); + CheckpointInfo? checkpoint = null; + + // Act + await using (StreamingRun run = await environment.RunStreamingAsync(workflow, 1)) + { + await foreach (WorkflowEvent evt in run.WatchStreamAsync()) + { + Assert.False(evt is WorkflowErrorEvent, (evt as WorkflowErrorEvent)?.Exception?.ToString()); + if (evt is SuperStepCompletedEvent { CompletionInfo.Checkpoint: { } saved }) + { + checkpoint = saved; + } + } + } + + // Assert + Assert.NotNull(checkpoint); + Assert.NotNull(writerContext); + await AssertValuesAsync(writerContext); + WorkflowFormulaState restoredState = new(RecalcEngineFactory.Create()); + await using StreamingRun resumed = await environment.ResumeStreamingAsync(CreateWorkflow(restoredState), checkpoint); + Assert.Equal("canonical", restoredState.Get("TestValue").ToObject()); + Assert.Equal(sensitivity, restoredState.GetSensitivity("TestValue")); + Assert.NotNull(writerContext); + await AssertValuesAsync(writerContext); + + Workflow CreateWorkflow(WorkflowFormulaState formulaState) + { + StateExecutor source = new("source", (_, _) => default); + StateExecutor raw = new("raw", async (context, cancellationToken) => + await context.QueueStateUpdateAsync("TestValue", "native", VariableScopeNames.Topic, cancellationToken)); + StateExecutor canonical = new("canonical", async (context, cancellationToken) => + { + writerContext = context; + IWorkflowContext target = wrapped ? new DeclarativeWorkflowContext(context, formulaState) : context; + await target.QueueStateUpdateWithSensitivityAsync("TestValue", + new EvaluationResult("canonical", sensitivity), VariableScopeNames.Topic, cancellationToken); + }, formulaState, (context, _) => + { + writerContext = context; + return default; + }); + return new WorkflowBuilder(source).AddEdge(source, raw).AddEdge(source, canonical).Build(); + } + + async Task AssertValuesAsync(IWorkflowContext context) + { + Assert.Equal("native", await context.ReadStateAsync("TestValue", VariableScopeNames.Topic)); + foreach (string? scope in new string?[] { null, VariableScopeNames.Local, VariableScopeNames.Topic }) + { + EvaluationResult result = await context.ReadStateWithSensitivityAsync("TestValue", scope); + Assert.Equal("canonical", result.Value); + Assert.Equal(sensitivity, result.Sensitivity); + } + EvaluationResult evaluated = await context.EvaluateValueWithSensitivityAsync("Topic.TestValue"); + Assert.Equal("canonical", evaluated.Value); + Assert.Equal(sensitivity, evaluated.Sensitivity); + } + } + + [Theory] + [InlineData(false, false)] + [InlineData(false, true)] + [InlineData(true, false)] + [InlineData(true, true)] + public async Task CanonicalOwnership_MixedFanOutInitializationAllowsLaterAssignmentsAsync(bool offThread, bool formula) + { + // Arrange + WorkflowFormulaState state = new(RecalcEngineFactory.Create()); + IWorkflowContext? storedContext = null; + bool updated = false; + StateExecutor source = new("source", (_, _) => default); + StateExecutor raw = new("raw", async (context, cancellationToken) => + await context.QueueStateUpdateAsync("TestValue", "native", VariableScopeNames.Topic, cancellationToken)); + StateExecutor initializer = new("initializer", async (context, cancellationToken) => + { + DeclarativeWorkflowContext target = new(context, state); + if (formula) + { + FormulaValue value = await target.ReadOrInitStateAsync( + "TestValue", () => FormulaValue.New(42), VariableScopeNames.Topic, cancellationToken); + Assert.Equal(42m, value.ToObject()); + } + else + { + Assert.Equal(42, await target.ReadOrInitStateAsync( + "TestValue", () => 42, VariableScopeNames.Topic, cancellationToken)); + } + }, state); + StateExecutor writer = new("writer", async (context, cancellationToken) => + { + storedContext = context; + DeclarativeWorkflowContext target = new(context, state); + if (formula) + { + FormulaValue existing = await target.ReadOrInitStateAsync( + "TestValue", () => throw new InvalidOperationException("Existing values must not be initialized."), + VariableScopeNames.Local, cancellationToken); + Assert.Equal(42m, existing.ToObject()); + } + else + { + Assert.Equal(42, await context.ReadStateAsync("TestValue", VariableScopeNames.Local, cancellationToken)); + Assert.Equal(42, await target.ReadOrInitStateAsync( + "TestValue", () => throw new InvalidOperationException("Existing values must not be initialized."), + VariableScopeNames.Local, cancellationToken)); + } + await target.QueueStateUpdateWithSensitivityAsync("TestValue", + new EvaluationResult("updated", SensitivityLevel.Sensitive), VariableScopeNames.Topic, cancellationToken); + updated = true; + }); + Workflow workflow = new WorkflowBuilder(source) + .AddEdge(source, raw).AddEdge(source, initializer).AddEdge(initializer, writer).Build(); + var environment = offThread ? InProcessExecution.OffThread : InProcessExecution.Lockstep; + + // Act + await using StreamingRun run = await environment.RunStreamingAsync(workflow, 1); + await foreach (WorkflowEvent evt in run.WatchStreamAsync()) + { + Assert.False(evt is WorkflowErrorEvent, (evt as WorkflowErrorEvent)?.Exception?.ToString()); + } + + // Assert + Assert.True(updated); + Assert.NotNull(storedContext); + Assert.Equal("native", await storedContext.ReadStateAsync("TestValue", VariableScopeNames.Topic)); + WorkflowFormulaState restored = new(RecalcEngineFactory.Create()); + await restored.RestoreSnapshotAsync(storedContext, CancellationToken.None); + Assert.Equal("updated", restored.Get("TestValue").ToObject()); + Assert.Equal(SensitivityLevel.Sensitive, restored.GetSensitivity("TestValue")); + } + + [Theory] + [InlineData(false, false)] + [InlineData(false, true)] + [InlineData(true, false)] + [InlineData(true, true)] + public async Task CanonicalOwnership_LegacyInitializationPreservesNumericPayloadsAsync(bool offThread, bool formula) + { + // Arrange + object[] values = [42, 42L, 1.5f, 1.5d, 1.5m]; + bool read = false; + bool restored = false; + CheckpointManager checkpoints = CreateJsonCheckpointManager(); + var environment = (offThread ? InProcessExecution.OffThread : InProcessExecution.Lockstep).WithCheckpointing(checkpoints); + CheckpointInfo? checkpoint = null; + + // Act + await using (StreamingRun run = await environment.RunStreamingAsync(CreateWorkflow(), 1)) + { + await foreach (WorkflowEvent evt in run.WatchStreamAsync()) + { + Assert.False(evt is WorkflowErrorEvent, (evt as WorkflowErrorEvent)?.Exception?.ToString()); + if (evt is SuperStepCompletedEvent { CompletionInfo.Checkpoint: { } saved }) + { + checkpoint = saved; + } + } + } + + // Assert + Assert.True(read); + Assert.NotNull(checkpoint); + await using StreamingRun resumed = await environment.ResumeStreamingAsync(CreateWorkflow(), checkpoint); + Assert.True(restored); + + Workflow CreateWorkflow() + { + WorkflowFormulaState state = new(RecalcEngineFactory.Create()); + StateExecutor source = new("source", async (context, cancellationToken) => + { + for (int i = 0; i < values.Length; i++) + { + await context.QueueStateUpdateAsync($"Value{i}", values[i], VariableScopeNames.Topic, cancellationToken); + await context.QueueStateUpdateAsync($"Value{i}", SensitivityLevel.Sensitive, + WorkflowFormulaState.GetSensitivityScopeName(VariableScopeNames.Local), cancellationToken); + } + }); + StateExecutor initializer = new("initializer", async (context, cancellationToken) => + { + DeclarativeWorkflowContext target = new(context, state); + for (int i = 0; i < values.Length; i++) + { + if (formula) + { + await target.ReadOrInitStateAsync( + $"Value{i}", () => throw new InvalidOperationException("Existing values must not be initialized."), + VariableScopeNames.Topic, cancellationToken); + } + else + { + PortableValue result = await target.ReadOrInitStateAsync( + $"Value{i}", () => throw new InvalidOperationException("Existing values must not be initialized."), + VariableScopeNames.Topic, cancellationToken); + Assert.Equal(values[i], result.AsType(values[i].GetType())); + } + } + }); + StateExecutor reader = new("reader", async (context, cancellationToken) => + { + await AssertValuesAsync(context, cancellationToken); + read = true; + }, restored: async (context, cancellationToken) => + { + await AssertValuesAsync(context, cancellationToken); + restored = true; + }); + return new WorkflowBuilder(source).AddEdge(source, initializer).AddEdge(initializer, reader).Build(); + } + + async Task AssertValuesAsync(IWorkflowContext context, CancellationToken cancellationToken) + { + Assert.Equal(42, await context.ReadStateAsync("Value0", VariableScopeNames.Local, cancellationToken)); + Assert.Equal(42L, await context.ReadStateAsync("Value1", VariableScopeNames.Local, cancellationToken)); + Assert.Equal(1.5f, await context.ReadStateAsync("Value2", VariableScopeNames.Local, cancellationToken)); + Assert.Equal(1.5d, await context.ReadStateAsync("Value3", VariableScopeNames.Local, cancellationToken)); + Assert.Equal(1.5m, await context.ReadStateAsync("Value4", VariableScopeNames.Local, cancellationToken)); + for (int i = 0; i < values.Length; i++) + { + EvaluationResult result = await context.ReadStateWithSensitivityAsync( + $"Value{i}", VariableScopeNames.Local, cancellationToken); + Assert.NotNull(result.Value); + Assert.True(result.Value.TypeId.IsMatch(values[i].GetType())); + Assert.Equal(values[i], result.Value.AsType(values[i].GetType())); + Assert.Equal(SensitivityLevel.Sensitive, result.Sensitivity); + } + Assert.Empty(await context.ReadStateKeysAsync(VariableScopeNames.Topic, cancellationToken)); + } + } + + [Theory] + [InlineData(false)] + [InlineData(true)] + public async Task CanonicalOwnership_InitializationPreservesMarkedLegacyValueAsync(bool formula) + { + // Arrange + Dictionary<(string? Scope, string Key), PortableValue> values = new() + { + [(VariableScopeNames.Topic, "TestValue")] = new("legacy"), + }; + Dictionary<(string? Scope, string Key), SensitivityLevel> sensitivities = new() + { + [(WorkflowFormulaState.GetSensitivityScopeName(VariableScopeNames.Local), "TestValue")] = SensitivityLevel.Sensitive, + }; + IWorkflowContext source = CreateStateContext(values, sensitivities).Object; + WorkflowFormulaState state = new(RecalcEngineFactory.Create()); + DeclarativeWorkflowContext context = new(source, state); + + // Act + if (formula) + { + FormulaValue result = await context.ReadOrInitStateAsync( + "TestValue", () => FormulaValue.New("replacement"), VariableScopeNames.Topic); + Assert.Equal("legacy", result.ToObject()); + } + else + { + Assert.Equal("legacy", await context.ReadOrInitStateAsync("TestValue", () => "replacement", VariableScopeNames.Topic)); + } + + // Assert + EvaluationResult restored = await source.ReadStateWithSensitivityAsync("TestValue"); + Assert.Equal("legacy", restored.Value); + Assert.Equal(SensitivityLevel.Sensitive, restored.Sensitivity); + Assert.Equal(SensitivityLevel.Sensitive, state.GetSensitivity("TestValue")); + Assert.False(values.ContainsKey((VariableScopeNames.Topic, "TestValue"))); + } + + [Theory] + [InlineData(SensitivityLevel.None)] + [InlineData(SensitivityLevel.Sensitive)] + public async Task CanonicalOwnership_MigrationDoesNotClaimLaterRawTopicReuseAsync(SensitivityLevel sensitivity) + { + // Arrange + Dictionary<(string? Scope, string Key), PortableValue> values = new() + { + [(VariableScopeNames.Topic, "TestValue")] = new("legacy"), + }; + Dictionary<(string? Scope, string Key), SensitivityLevel> sensitivities = new() + { + [(WorkflowFormulaState.GetSensitivityScopeName(VariableScopeNames.Local), "TestValue")] = sensitivity, + }; + IWorkflowContext context = CreateStateContext(values, sensitivities).Object; + WorkflowFormulaState state = new(RecalcEngineFactory.Create()); + await state.RestoreCheckpointAsync(context, CancellationToken.None); + values[(VariableScopeNames.Topic, "TestValue")] = new("native"); + + // Act + await state.RestoreCheckpointAsync(context, CancellationToken.None); + EvaluationResult result = await context.ReadStateWithSensitivityAsync("TestValue"); + + // Assert + Assert.Equal("legacy", result.Value); + Assert.Equal(sensitivity, result.Sensitivity); + Assert.Equal("legacy", state.Get("TestValue").ToObject()); + Assert.Equal(sensitivity, state.GetSensitivity("TestValue")); + Assert.Equal("native", values[(VariableScopeNames.Topic, "TestValue")].As()); + } + + [Theory] + [InlineData(false, SensitivityLevel.None)] + [InlineData(false, SensitivityLevel.Sensitive)] + [InlineData(true, SensitivityLevel.None)] + [InlineData(true, SensitivityLevel.Sensitive)] + public async Task CanonicalOwnership_UpdatesAndClearsPreserveRawTopicStateAsync(bool wrapped, SensitivityLevel sensitivity) + { + // Arrange + Dictionary<(string? Scope, string Key), PortableValue> values = []; + Dictionary<(string? Scope, string Key), SensitivityLevel> sensitivities = []; + IWorkflowContext source = CreateStateContext(values, sensitivities).Object; + WorkflowFormulaState state = new(RecalcEngineFactory.Create()); + IWorkflowContext context = wrapped ? new DeclarativeWorkflowContext(source, state) : source; + await context.QueueStateUpdateWithSensitivityAsync("TestValue", + new EvaluationResult("initial", sensitivity), VariableScopeNames.Local); + values[(VariableScopeNames.Topic, "TestValue")] = new("native"); + + // Act + await context.QueueStateUpdateWithSensitivityAsync("TestValue", + new EvaluationResult("updated", sensitivity), VariableScopeNames.Topic); + EvaluationResult updated = await source.ReadStateWithSensitivityAsync("TestValue"); + await context.QueueStateUpdateWithSensitivityAsync("TestValue", + new EvaluationResult(null, sensitivity), VariableScopeNames.Local); + await state.RestoreSnapshotAsync(source, CancellationToken.None); + + // Assert + Assert.Equal("updated", updated.Value); + Assert.Equal(sensitivity, updated.Sensitivity); + Assert.Equal("native", values[(VariableScopeNames.Topic, "TestValue")].As()); + Assert.IsType(state.Get("TestValue")); + if (wrapped) + { + Assert.Equal(sensitivity, state.GetSensitivity("TestValue")); + } + else + { + EvaluationResult cleared = await source.ReadStateWithSensitivityAsync("TestValue"); + Assert.Null(cleared.Value); + Assert.Equal(SensitivityLevel.None, cleared.Sensitivity); + Assert.Empty(sensitivities); + await Assert.ThrowsAsync(async () => + await source.QueueStateUpdateWithSensitivityAsync("TestValue", + new EvaluationResult("replacement", sensitivity))); + } + } + + [Theory] + [InlineData(false)] + [InlineData(true)] + public async Task CanonicalOwnership_PointOperationsUseBoundedKeyLookupsAsync(bool wrapped) + { + // Arrange + const int Assignments = 32; + Dictionary<(string? Scope, string Key), PortableValue> values = []; + Dictionary<(string? Scope, string Key), SensitivityLevel> sensitivities = []; + Mock source = CreateStateContext(values, sensitivities); + IWorkflowContext context = wrapped ? new DeclarativeWorkflowContext(source.Object, new(RecalcEngineFactory.Create())) : source.Object; + + // Act + for (int i = 0; i < Assignments; ++i) + { + string key = $"Value{i}"; + await context.QueueStateUpdateWithSensitivityAsync(key, + new EvaluationResult("value", SensitivityLevel.None), VariableScopeNames.Local); + EvaluationResult result = await source.Object.ReadStateWithSensitivityAsync(key, VariableScopeNames.Topic); + Assert.Equal("value", result.Value); + } + + // Assert + source.Verify(c => c.ReadStateKeysAsync(It.IsAny(), It.IsAny()), Times.Never); + source.Verify(c => c.ReadStateAsync(It.IsAny(), It.IsAny(), It.IsAny()), + Times.AtMost(Assignments * 8)); + } + + [Theory] + [InlineData(false, false)] + [InlineData(false, true)] + [InlineData(true, false)] + [InlineData(true, true)] + public async Task CanonicalOwnership_UnsupportedVersionsDoNotDisambiguateStateAsync(bool restore, bool overwrite) + { + // Arrange + Dictionary<(string? Scope, string Key), PortableValue> values = new() + { + [(VariableScopeNames.Local, "TestValue")] = new("canonical"), + [(VariableScopeNames.Topic, "TestValue")] = new("native"), + [(WorkflowFormulaState.LocalOwnershipScopeName, "TestValue")] = new(2), + }; + Dictionary<(string? Scope, string Key), SensitivityLevel> sensitivities = new() + { + [(WorkflowFormulaState.GetSensitivityScopeName(VariableScopeNames.Local), "TestValue")] = SensitivityLevel.None, + }; + IWorkflowContext context = CreateStateContext(values, sensitivities).Object; + WorkflowFormulaState state = new(RecalcEngineFactory.Create()); + + // Act + DeclarativeActionException error = await Assert.ThrowsAsync(async () => + { + if (restore) + { + if (overwrite) + { + await state.RestoreCheckpointAsync(context, CancellationToken.None); + } + else + { + await state.RestoreSnapshotAsync(context, CancellationToken.None); + } + } + else if (overwrite) + { + await context.QueueStateUpdateWithSensitivityAsync("TestValue", + new EvaluationResult("replacement", SensitivityLevel.Sensitive)); + } + else + { + await context.ReadStateWithSensitivityAsync("TestValue"); + } + }); + + // Assert + Assert.Contains("Unsupported declarative state ownership version", error.Message); + Assert.Equal("canonical", values[(VariableScopeNames.Local, "TestValue")].As()); + Assert.Equal("native", values[(VariableScopeNames.Topic, "TestValue")].As()); + Assert.Equal(2, values[(WorkflowFormulaState.LocalOwnershipScopeName, "TestValue")].As()); + Assert.Equal(SensitivityLevel.None, Assert.Single(sensitivities).Value); + } + + private static CheckpointManager CreateJsonCheckpointManager() + { + Dictionary stored = []; + Mock> store = new(); + store.Setup(s => s.CreateCheckpointAsync(It.IsAny(), It.IsAny(), It.IsAny())) + .ReturnsAsync((string sessionId, JsonElement value, CheckpointInfo? _) => + { + CheckpointInfo info = new(sessionId, Guid.NewGuid().ToString("N")); + stored.Add(info, value); + return info; + }); + store.Setup(s => s.RetrieveCheckpointAsync(It.IsAny(), It.IsAny())) + .ReturnsAsync((string _, CheckpointInfo info) => stored[info]); + store.Setup(s => s.RetrieveIndexAsync(It.IsAny(), It.IsAny())) + .ReturnsAsync((string sessionId, CheckpointInfo? _) => + stored.Keys.Where(info => string.Equals(info.SessionId, sessionId, StringComparison.Ordinal)).ToArray().AsEnumerable()); + return CheckpointManager.CreateJson(store.Object, DeclarativeWorkflowJsonOptions.Default); + } + + private static async ValueTask SeedLegacyStateAsync(IWorkflowContext context, CancellationToken cancellationToken) + { + await context.QueueStateUpdateAsync("TestValue", "legacy", VariableScopeNames.Topic, cancellationToken); + await context.QueueStateUpdateAsync("TestValue", SensitivityLevel.Sensitive, + WorkflowFormulaState.GetSensitivityScopeName(VariableScopeNames.Local), cancellationToken); + } + + private sealed class StateExecutor : Executor, IResettableExecutor + { + private readonly Func _handle; + private readonly Func? _restored; + + public StateExecutor( + string id, + Func handle, + WorkflowFormulaState? state = null, + Func? restored = null) : base(id) + { + this._handle = handle; + this._restored = restored; + if (state is not null) + { + this.RegisterCheckpointRestoringCallback(state, state.RestoreCheckpointAsync); + } + } + + public ValueTask ResetAsync() => default; + + [SendsMessage(typeof(int))] + public override async ValueTask HandleAsync(int message, IWorkflowContext context, CancellationToken cancellationToken = default) + { + await this._handle(context, cancellationToken); + await context.SendMessageAsync(message, cancellationToken: cancellationToken); + } + + protected override ValueTask OnCheckpointRestoredAsync(IWorkflowContext context, CancellationToken cancellationToken = default) => + this._restored?.Invoke(context, cancellationToken) ?? default; + } + + private static Mock CreateStateContext( + Dictionary<(string? Scope, string Key), PortableValue> values, + Dictionary<(string? Scope, string Key), SensitivityLevel> sensitivities) + { + Mock source = new(MockBehavior.Strict); + source.Setup(c => c.QueueStateUpdateAsync(It.IsAny(), It.IsAny(), It.IsAny(), It.IsAny())) + .Callback((string key, object? value, string? scope, CancellationToken _) => + { + if (value is null) + { + values.Remove((scope, key)); + sensitivities.Remove((scope, key)); + } + else + { + values[(scope, key)] = new PortableValue(value); + } + }) + .Returns(default(ValueTask)); + source.Setup(c => c.QueueStateUpdateAsync(It.IsAny(), It.IsAny(), It.IsAny(), It.IsAny())) + .Callback((string key, UnassignedValue value, string? scope, CancellationToken _) => values[(scope, key)] = new(value)) + .Returns(default(ValueTask)); + source.Setup(c => c.QueueStateUpdateAsync(It.IsAny(), It.IsAny(), It.IsAny(), It.IsAny())) + .Callback((string key, SensitivityLevel value, string? scope, CancellationToken _) => sensitivities[(scope, key)] = value) + .Returns(default(ValueTask)); + source.Setup(c => c.QueueStateUpdateAsync(It.IsAny(), It.IsAny(), It.IsAny(), It.IsAny())) + .Callback((string key, int value, string? scope, CancellationToken _) => values[(scope, key)] = new(value)) + .Returns(default(ValueTask)); + source.Setup(c => c.QueueClearScopeAsync(It.IsAny(), It.IsAny())) + .Callback((string? scope, CancellationToken _) => + { + foreach (var key in values.Keys.Where(key => string.Equals(key.Scope, scope, StringComparison.Ordinal)).ToArray()) + { + values.Remove(key); + } + }) + .Returns(default(ValueTask)); + source.Setup(c => c.ReadStateKeysAsync(It.IsAny(), It.IsAny())) + .ReturnsAsync((string? scope, CancellationToken _) => values.Keys.Concat(sensitivities.Keys) + .Where(k => string.Equals(k.Scope, scope, StringComparison.Ordinal)).Select(k => k.Key).ToHashSet()); + source.Setup(c => c.ReadStateAsync(It.IsAny(), It.IsAny(), It.IsAny())) + .ReturnsAsync((string key, string? scope, CancellationToken _) => values.TryGetValue((scope, key), out PortableValue? value) + ? value : sensitivities.TryGetValue((scope, key), out SensitivityLevel sensitivity) ? new PortableValue(sensitivity) : null); + source.Setup(c => c.ReadStateAsync(It.IsAny(), It.IsAny(), It.IsAny())) + .ReturnsAsync((string key, string? scope, CancellationToken _) => sensitivities.TryGetValue((scope, key), out SensitivityLevel sensitivity) ? sensitivity : SensitivityLevel.None); + source.Setup(c => c.ReadStateAsync(It.IsAny(), It.IsAny(), It.IsAny())) + .ReturnsAsync((string key, string? scope, CancellationToken _) => values.TryGetValue((scope, key), out PortableValue? value) ? value.As() : null); + source.Setup(c => c.ReadOrInitStateAsync(It.IsAny(), It.IsAny>(), It.IsAny(), It.IsAny())) + .ReturnsAsync((string key, Func factory, string? scope, CancellationToken _) => values.TryGetValue((scope, key), out PortableValue? value) ? Assert.IsType(value.As()) : factory()); + return source; + } } diff --git a/dotnet/tests/Microsoft.Agents.AI.Workflows.UnitTests/CheckpointResumeTests.cs b/dotnet/tests/Microsoft.Agents.AI.Workflows.UnitTests/CheckpointResumeTests.cs index 237093dd67d..1095d55a90a 100644 --- a/dotnet/tests/Microsoft.Agents.AI.Workflows.UnitTests/CheckpointResumeTests.cs +++ b/dotnet/tests/Microsoft.Agents.AI.Workflows.UnitTests/CheckpointResumeTests.cs @@ -136,18 +136,205 @@ internal async Task Checkpoint_Resume_RestoresAllStateBeforeSubclassHooksAsync(E Assert.Equal(1, sourceExecutorRestored); } + [Theory] + [InlineData(ExecutionEnvironment.InProcess_OffThread, false)] + [InlineData(ExecutionEnvironment.InProcess_OffThread, true)] + [InlineData(ExecutionEnvironment.InProcess_Lockstep, false)] + [InlineData(ExecutionEnvironment.InProcess_Lockstep, true)] + internal async Task Checkpoint_Restore_FailedPublicationCanBeRetriedAsync(ExecutionEnvironment environment, bool checkRolledBackState) + { + // Arrange + bool conflicting = true; + IWorkflowContext? restoringContext = null; + RestoreProbeExecutor source = new("source", async (context, cancellationToken) => + { + restoringContext = context; + await context.QueueStateUpdateAsync("other", "changed", "other-scope", cancellationToken); + await context.QueueStateUpdateAsync("value", "source", "shared", cancellationToken); + }, () => true); + RestoreProbeExecutor sink = new("sink", (context, cancellationToken) => + conflicting ? context.QueueStateUpdateAsync("value", "sink", "shared", cancellationToken) : default, () => true); + Workflow workflow = new WorkflowBuilder(source).AddEdge(source, sink).Build(); + CheckpointManager checkpoints = CheckpointManager.CreateInMemory(); + InProcessExecutionEnvironment env = environment.ToWorkflowExecutionEnvironment().WithCheckpointing(checkpoints); + CheckpointInfo? checkpoint = null; + await using StreamingRun run = await env.RunStreamingAsync(workflow, 1); + await foreach (WorkflowEvent evt in run.WatchStreamAsync()) + { + if (evt is SuperStepCompletedEvent { CompletionInfo.Checkpoint: { } saved }) + { + checkpoint = saved; + } + } + Assert.NotNull(checkpoint); + + // Act + InvalidOperationException error = await Assert.ThrowsAsync( + async () => await run.RestoreCheckpointAsync(checkpoint)); + + // Assert + Assert.Contains("Expected exactly one update", error.Message); + Assert.Null(source.RestorationCompleteInHook); + Assert.Null(sink.RestorationCompleteInHook); + Assert.NotNull(restoringContext); + if (checkRolledBackState) + { + Assert.Null(await restoringContext.ReadStateAsync("other", "other-scope")); + Assert.Null(await restoringContext.ReadStateAsync("value", "shared")); + } + + conflicting = false; + await run.RestoreCheckpointAsync(checkpoint); + Assert.True(source.RestorationCompleteInHook); + Assert.True(sink.RestorationCompleteInHook); + Assert.Equal("source", await restoringContext.ReadStateAsync("value", "shared")); + } + + [Theory] + [InlineData(ExecutionEnvironment.InProcess_OffThread, false)] + [InlineData(ExecutionEnvironment.InProcess_OffThread, true)] + [InlineData(ExecutionEnvironment.InProcess_Lockstep, false)] + [InlineData(ExecutionEnvironment.InProcess_Lockstep, true)] + internal async Task Checkpoint_Restore_LaterFailurePreservesPublishedStateAsync(ExecutionEnvironment environment, bool failIndex) + { + // Arrange + InvalidOperationException failure = new("Checkpoint load failed."); + string? liveValue = null; + IWorkflowContext? restoringContext = null; + RestoreProbeExecutor source = new("source", async (context, cancellationToken) => + { + restoringContext = context; + liveValue = "restored"; + await context.QueueStateUpdateAsync("value", liveValue, "shared", cancellationToken); + }, () => failIndex ? true : throw failure); + Workflow workflow = new WorkflowBuilder(source).Build(); + IndexFailureCheckpointManager manager = new(); + CheckpointManager checkpoints = new(manager); + InProcessExecutionEnvironment env = environment.ToWorkflowExecutionEnvironment().WithCheckpointing(checkpoints); + CheckpointInfo? saved = null; + await using StreamingRun run = await env.RunStreamingAsync(workflow, 1); + await foreach (WorkflowEvent evt in run.WatchStreamAsync()) + { + if (evt is SuperStepCompletedEvent { CompletionInfo.Checkpoint: { } checkpoint }) + { + saved = checkpoint; + } + } + Assert.NotNull(saved); + + // Act + manager.IndexFailure = failIndex ? failure : null; + InvalidOperationException error = await Assert.ThrowsAsync( + async () => await run.RestoreCheckpointAsync(saved)); + + // Assert + Assert.Same(failure, error); + Assert.Equal("restored", liveValue); + Assert.NotNull(restoringContext); + Assert.Equal(liveValue, await restoringContext.ReadStateAsync("value", "shared")); + } + + [Theory] + [InlineData(ExecutionEnvironment.InProcess_OffThread, false)] + [InlineData(ExecutionEnvironment.InProcess_OffThread, true)] + [InlineData(ExecutionEnvironment.InProcess_Lockstep, false)] + [InlineData(ExecutionEnvironment.InProcess_Lockstep, true)] + internal async Task Checkpoint_Restore_LaterFailureDiscardsHookUpdatesAsync(ExecutionEnvironment environment, bool failIndex) + { + // Arrange + InvalidOperationException failure = new("Checkpoint load failed."); + bool failing = true; + TaskCompletionSource hookEntered = new(TaskCreationOptions.RunContinuationsAsynchronously); + TaskCompletionSource releaseHook = new(TaskCreationOptions.RunContinuationsAsynchronously); + IWorkflowContext? restoringContext = null; + RestoreProbeExecutor source = new("source", async (context, cancellationToken) => + { + restoringContext = context; + await context.QueueStateUpdateAsync("value", "restored", "shared", cancellationToken); + }, () => true, postLoad: async (context, cancellationToken) => + { + if (failing) + { + hookEntered.TrySetResult(true); + await releaseHook.Task; + await context.QueueStateUpdateAsync("hook-value", "pending", "shared", cancellationToken); + } + }); + RestoreProbeExecutor sink = new("sink", (_, _) => default, () => !failing || failIndex ? true : throw failure); + Workflow workflow = new WorkflowBuilder(source).AddEdge(source, sink).Build(); + IndexFailureCheckpointManager manager = new(); + CheckpointManager checkpoints = new(manager); + InProcessExecutionEnvironment env = environment.ToWorkflowExecutionEnvironment().WithCheckpointing(checkpoints); + CheckpointInfo? saved = null; + await using StreamingRun run = await env.RunStreamingAsync(workflow, 1); + await foreach (WorkflowEvent evt in run.WatchStreamAsync()) + { + if (evt is SuperStepCompletedEvent { CompletionInfo.Checkpoint: { } checkpoint }) + { + saved = checkpoint; + } + } + Assert.NotNull(saved); + + // Act + manager.IndexFailure = failIndex ? failure : null; + Task restoreTask = Assert.ThrowsAsync( + async () => await run.RestoreCheckpointAsync(saved)); + await hookEntered.Task; + try + { + Assert.False(restoreTask.IsCompleted); + } + finally + { + releaseHook.TrySetResult(true); + } + InvalidOperationException error = await restoreTask; + + // Assert + Assert.Same(failure, error); + Assert.NotNull(restoringContext); + Assert.Equal("restored", await restoringContext.ReadStateAsync("value", "shared")); + Assert.Null(await restoringContext.ReadStateAsync("hook-value", "shared")); + + failing = false; + manager.IndexFailure = null; + await run.RestoreCheckpointAsync(saved); + Assert.Equal("restored", await restoringContext.ReadStateAsync("value", "shared")); + Assert.Null(await restoringContext.ReadStateAsync("hook-value", "shared")); + } + + private sealed class IndexFailureCheckpointManager : ICheckpointManager + { + private readonly ICheckpointManager _inner = CheckpointManager.CreateInMemory(); + + public Exception? IndexFailure { get; set; } + + public ValueTask CommitCheckpointAsync(string sessionId, Checkpoint checkpoint) => + this._inner.CommitCheckpointAsync(sessionId, checkpoint); + + public ValueTask LookupCheckpointAsync(string sessionId, CheckpointInfo checkpointInfo) => + this._inner.LookupCheckpointAsync(sessionId, checkpointInfo); + + public ValueTask> RetrieveIndexAsync(string sessionId, CheckpointInfo? withParent = null) => + this.IndexFailure is { } failure ? throw failure : this._inner.RetrieveIndexAsync(sessionId, withParent); + } + private sealed class RestoreProbeExecutor : Executor, IResettableExecutor { private readonly Func _isRestored; + private readonly Func? _postLoad; public RestoreProbeExecutor( string id, Func restore, Func isRestored, object? sharedStateKey = null, - Func? restoreExecutor = null) : base(id) + Func? restoreExecutor = null, + Func? postLoad = null) : base(id) { this._isRestored = isRestored; + this._postLoad = postLoad; if (sharedStateKey is null) { this.RegisterCheckpointRestoringCallback(restore); @@ -170,10 +357,13 @@ public RestoreProbeExecutor( public override ValueTask HandleAsync(int message, IWorkflowContext context, CancellationToken cancellationToken = default) => context.SendMessageAsync(message, cancellationToken: cancellationToken); - protected internal override ValueTask OnCheckpointRestoredAsync(IWorkflowContext context, CancellationToken cancellationToken = default) + protected internal override async ValueTask OnCheckpointRestoredAsync(IWorkflowContext context, CancellationToken cancellationToken = default) { this.RestorationCompleteInHook = this._isRestored(); - return default; + if (this._postLoad is { } postLoad) + { + await postLoad(context, cancellationToken); + } } } diff --git a/skills/agent-framework-builder-dotnet/references/declarative.md b/skills/agent-framework-builder-dotnet/references/declarative.md index b5bddca2c74..b78b6eaeea9 100644 --- a/skills/agent-framework-builder-dotnet/references/declarative.md +++ b/skills/agent-framework-builder-dotnet/references/declarative.md @@ -35,6 +35,31 @@ commitments. Validate definitions at startup and fail clearly on unknown providers, actions, tools, variables, or resources. Apply the same approval, evaluation, and telemetry policy used for code-first agents and workflows. +`Local` and `Topic` address the same declarative workflow state. On restore, legacy `Topic` entries identified by +declarative sensitivity metadata migrate to `Local` before post-load hooks or resumed actions run. Only migrated +entries are removed from `Topic`; unmarked entries, including code-first state, remain untouched. +Older checkpoints containing an identified declarative key in both physical scopes without canonical ownership +metadata are rejected because assignment order is unavailable. Reconcile those entries explicitly or start a fresh +session before resuming. Legacy entries without declarative metadata require explicit reconciliation rather than +automatic migration. + +For code-first executors using the declarative sensitivity helpers without a formula-state restoration callback, +reads fall back to the requested metadata-marked legacy entry without modifying it. Writes use `Local` and remove +only that entry from `Topic`; unrelated legacy entries are not migrated by these point operations. +Ad hoc expression evaluation also reads legacy entries without persisting migration, so parallel expression readers +do not queue competing updates. Persisted migration still runs in registered formula-state restoration callbacks. +New canonical assignments, managed initializations, and migrated entries persist separate, versioned ownership +metadata. This identifies their sensitivity metadata as belonging to `Local`, not to a same-key raw code-first `Topic` entry. Concurrent +code-first `Topic` and canonical `Local` assignments can therefore coexist; reads and restoration preserve the +unrelated raw `Topic` value. Preserve these private metadata scopes together with values in custom checkpoint stores. +Initialization leaves existing values and their sensitivity metadata unchanged. Legacy initialization preserves the +stored native payload even when the caller requests a formula or variant value. +Without recorded canonical ownership, writes still reject a visible same-key unmarked code-first `Topic` entry +before changing state; reconcile such collisions explicitly. + +A plain sensitivity-helper null update deletes the canonical value, sensitivity metadata, and ownership metadata. +A wrapped declarative clear retains a blank value with its sensitivity and ownership metadata. + ## Learn more - [Declarative agents](https://learn.microsoft.com/agent-framework/agents/declarative)