// Copyright (c) Microsoft. All rights reserved. using System.Diagnostics; using System.Diagnostics.CodeAnalysis; using System.Text.Json; using Microsoft.Agents.AI.Workflows; namespace Microsoft.Agents.AI.DurableTask.Workflows; /// /// A workflow context for durable workflow execution. /// /// /// State is passed in from the orchestration and updates are collected for return. /// Events emitted during execution are collected and returned to the orchestration /// as part of the activity output for streaming to callers. /// [DebuggerDisplay("Executor = {_executor.Id}, StateEntries = {_initialState.Count}")] internal sealed class DurableWorkflowContext : IWorkflowContext { /// /// The default scope name used when no explicit scope is specified. /// Scopes partition shared state into logical namespaces so that different /// parts of a workflow can manage their state keys independently. /// private const string DefaultScopeName = "__default__"; private readonly Dictionary _initialState; private readonly Executor _executor; /// /// Initializes a new instance of the class. /// /// The shared state passed from the orchestration. /// The executor running in this context. internal DurableWorkflowContext(Dictionary? initialState, Executor executor) { this._executor = executor; this._initialState = initialState ?? []; } /// /// Gets the messages sent during activity execution via . /// internal List SentMessages { get; } = []; /// /// Gets the outbound events that were added during activity execution. /// internal List OutboundEvents { get; } = []; /// /// Gets the state updates made during activity execution. /// internal Dictionary StateUpdates { get; } = []; /// /// Gets the scopes that were cleared during activity execution. /// internal HashSet ClearedScopes { get; } = []; /// /// Gets a value indicating whether the executor requested a workflow halt. /// internal bool HaltRequested { get; private set; } /// public ValueTask AddEventAsync( WorkflowEvent workflowEvent, CancellationToken cancellationToken = default) { if (workflowEvent is not null) { this.OutboundEvents.Add(workflowEvent); } return default; } /// [UnconditionalSuppressMessage("AOT", "IL3050", Justification = "Serializing workflow message types registered at startup.")] [UnconditionalSuppressMessage("Trimming", "IL2026", Justification = "Serializing workflow message types registered at startup.")] public ValueTask SendMessageAsync( object message, string? targetId = null, CancellationToken cancellationToken = default) { if (message is not null) { Type messageType = message.GetType(); this.SentMessages.Add(new TypedPayload { Data = JsonSerializer.Serialize(message, messageType, DurableSerialization.Options), TypeName = messageType.AssemblyQualifiedName }); } return default; } /// public ValueTask YieldOutputAsync( object output, CancellationToken cancellationToken = default) { if (output is not null) { Type outputType = output.GetType(); if (!this._executor.CanOutput(outputType)) { throw new InvalidOperationException( $"Cannot output object of type {outputType.Name}. " + $"Expecting one of [{string.Join(", ", this._executor.OutputTypes)}]."); } this.OutboundEvents.Add(new WorkflowOutputEvent(output, this._executor.Id)); } return default; } /// public ValueTask RequestHaltAsync() { this.HaltRequested = true; this.OutboundEvents.Add(new DurableHaltRequestedEvent(this._executor.Id)); return default; } /// public ValueTask ReadStateAsync( string key, string? scopeName = null, CancellationToken cancellationToken = default) { ArgumentException.ThrowIfNullOrEmpty(key); string scopeKey = GetScopeKey(scopeName, key); string normalizedScope = scopeName ?? DefaultScopeName; bool scopeCleared = this.ClearedScopes.Contains(normalizedScope); // Local updates take priority over initial state. if (this.StateUpdates.TryGetValue(scopeKey, out string? updated)) { return DeserializeStateAsync(updated); } // If scope was cleared, ignore initial state if (scopeCleared) { return ValueTask.FromResult(default); } // Fall back to initial state passed from orchestration if (this._initialState.TryGetValue(scopeKey, out string? initial)) { return DeserializeStateAsync(initial); } return ValueTask.FromResult(default); } /// public async ValueTask ReadOrInitStateAsync( string key, Func initialStateFactory, string? scopeName = null, CancellationToken cancellationToken = default) { ArgumentException.ThrowIfNullOrEmpty(key); ArgumentNullException.ThrowIfNull(initialStateFactory); // Cannot rely on `value is not null` because T? on an unconstrained generic // parameter does not become Nullable for value types — the null check is // always true for types like int. Instead, check key existence directly. if (this.HasStateKey(key, scopeName)) { T? value = await this.ReadStateAsync(key, scopeName, cancellationToken).ConfigureAwait(false); if (value is not null) { return value; } } T initialValue = initialStateFactory(); await this.QueueStateUpdateAsync(key, initialValue, scopeName, cancellationToken).ConfigureAwait(false); return initialValue; } /// public ValueTask> ReadStateKeysAsync( string? scopeName = null, CancellationToken cancellationToken = default) { string scopePrefix = GetScopePrefix(scopeName); int scopePrefixLength = scopePrefix.Length; HashSet keys = new(StringComparer.Ordinal); bool scopeCleared = scopeName is null ? this.ClearedScopes.Contains(DefaultScopeName) : this.ClearedScopes.Contains(scopeName); // Start with keys from initial state (skip if scope was cleared) if (!scopeCleared) { foreach (string stateKey in this._initialState.Keys) { if (stateKey.StartsWith(scopePrefix, StringComparison.Ordinal)) { keys.Add(stateKey[scopePrefixLength..]); } } } // Merge local updates: add if non-null, remove if null (deleted) foreach (KeyValuePair update in this.StateUpdates) { if (!update.Key.StartsWith(scopePrefix, StringComparison.Ordinal)) { continue; } string key = update.Key[scopePrefixLength..]; if (update.Value is not null) { keys.Add(key); } else { keys.Remove(key); } } return ValueTask.FromResult(keys); } /// public ValueTask QueueStateUpdateAsync( string key, T? value, string? scopeName = null, CancellationToken cancellationToken = default) { ArgumentException.ThrowIfNullOrEmpty(key); string scopeKey = GetScopeKey(scopeName, key); this.StateUpdates[scopeKey] = value is null ? null : SerializeState(value); return default; } /// public ValueTask QueueClearScopeAsync( string? scopeName = null, CancellationToken cancellationToken = default) { this.ClearedScopes.Add(scopeName ?? DefaultScopeName); // Remove any pending updates in this scope (snapshot keys to allow removal during iteration) string scopePrefix = GetScopePrefix(scopeName); foreach (string key in this.StateUpdates.Keys.ToList()) { if (key.StartsWith(scopePrefix, StringComparison.Ordinal)) { this.StateUpdates.Remove(key); } } return default; } /// public IReadOnlyDictionary? TraceContext => null; /// public bool ConcurrentRunsEnabled => false; private static string GetScopeKey(string? scopeName, string key) => $"{GetScopePrefix(scopeName)}{key}"; /// /// Checks whether the given key exists in local updates or initial state, /// respecting cleared scopes. /// private bool HasStateKey(string key, string? scopeName) { string scopeKey = GetScopeKey(scopeName, key); if (this.StateUpdates.TryGetValue(scopeKey, out string? updated)) { return updated is not null; } string normalizedScope = scopeName ?? DefaultScopeName; if (this.ClearedScopes.Contains(normalizedScope)) { return false; } return this._initialState.ContainsKey(scopeKey); } /// /// Returns the key prefix for the given scope. Scopes partition shared state /// into logical namespaces, allowing different workflow executors to manage /// their state keys independently. When no scope is specified, the /// is used. /// private static string GetScopePrefix(string? scopeName) => scopeName is null ? $"{DefaultScopeName}:" : $"{scopeName}:"; [UnconditionalSuppressMessage("AOT", "IL3050", Justification = "Serializing workflow state types.")] [UnconditionalSuppressMessage("Trimming", "IL2026", Justification = "Serializing workflow state types.")] private static string SerializeState(T value) => JsonSerializer.Serialize(value, DurableSerialization.Options); [UnconditionalSuppressMessage("AOT", "IL3050", Justification = "Deserializing workflow state types.")] [UnconditionalSuppressMessage("Trimming", "IL2026", Justification = "Deserializing workflow state types.")] private static ValueTask DeserializeStateAsync(string? json) { if (json is null) { return ValueTask.FromResult(default); } return ValueTask.FromResult(JsonSerializer.Deserialize(json, DurableSerialization.Options)); } }