// Copyright (c) Microsoft. All rights reserved. using System; using System.Collections.Generic; using System.Linq; using System.Runtime.CompilerServices; using System.Threading; using System.Threading.Tasks; using Microsoft.Extensions.AI; namespace Microsoft.Agents.AI; /// /// A delegating chat client that notifies and /// instances of request and response messages after each individual call to the inner chat client, /// or marks messages for later persistence depending on the configured mode. /// /// /// /// This decorator is intended to operate between the and the leaf /// in a pipeline. /// /// /// In persist mode (the default), it ensures that providers are notified and the session's /// is updated after each service call, so that /// intermediate messages (e.g., tool calls and results) are saved even if the process is interrupted /// mid-loop. /// /// /// In mark-only mode ( is ), it marks messages with metadata /// but does not notify providers or update the . /// Both are deferred to the at the end of the run, providing atomic /// run semantics. /// /// /// This chat client must be used within the context of a running . It retrieves the /// current agent and session from , which is set automatically when an agent's /// or /// /// method is called. The ensures the run context always contains a resolved session, /// even when the caller passes null. An is thrown if no run context is /// available or if the agent is not a . /// /// internal sealed class ChatHistoryPersistingChatClient : DelegatingChatClient { /// /// The key used in and /// to mark messages and their content as already persisted to chat history. /// internal const string PersistedMarkerKey = "_chatHistoryPersisted"; /// /// A sentinel value set on by /// when per-service-call persistence is active and no real conversation ID exists. /// /// /// /// This signals to that the chat history is being managed /// externally (by this decorator), which prevents it from adding duplicate /// messages into the request during approval-response processing. Without this sentinel, /// would reconstruct function-call messages from approval /// responses and append them to the original messages — but the loaded history already contains /// those same function calls, causing duplicate tool-call entries that the model rejects. /// /// /// This decorator strips the sentinel before forwarding requests to the inner client, so the /// underlying model never sees it. /// /// internal const string LocalHistoryConversationId = "_agent_local_history"; /// /// Initializes a new instance of the class. /// /// The underlying chat client that will handle the core operations. /// /// When , messages are marked with metadata but not persisted immediately, /// and the session's is not updated. /// The will persist only the marked messages and update the /// conversation ID at the end of the run. /// When (the default), messages are persisted and the conversation ID /// is updated immediately after each service call. /// public ChatHistoryPersistingChatClient(IChatClient innerClient, bool markOnly = false) : base(innerClient) { this.MarkOnly = markOnly; } /// /// Gets a value indicating whether this decorator is in mark-only mode. /// /// /// When , messages are marked with metadata but not persisted immediately, /// and the session's is not updated. /// Both are deferred to the at the end of the run. /// When , messages are persisted and the conversation ID is updated /// after each service call. /// public bool MarkOnly { get; } /// public override async Task GetResponseAsync( IEnumerable messages, ChatOptions? options = null, CancellationToken cancellationToken = default) { var (agent, session) = GetRequiredAgentAndSession(); options = StripLocalHistoryConversationId(options); ChatResponse response; try { response = await base.GetResponseAsync(messages, options, cancellationToken).ConfigureAwait(false); } catch (Exception ex) { var newRequestMessagesOnFailure = GetNewRequestMessages(messages); await agent.NotifyProvidersOfFailureAsync(session, ex, newRequestMessagesOnFailure, options, cancellationToken).ConfigureAwait(false); throw; } var newRequestMessages = GetNewRequestMessages(messages); if (this.ShouldDeferPersistence(options)) { // In mark-only mode or when resuming from a continuation token, just mark messages // for later persistence by ChatClientAgent. Conversation ID and provider notification // are deferred to end-of-run. For continuation tokens, the end-of-run handler needs // to send the combined data from both the previous and current runs. MarkAsPersisted(newRequestMessages); MarkAsPersisted(response.Messages); } else { // In persist mode, persist immediately and update conversation ID. agent.UpdateSessionConversationId(session, response.ConversationId, cancellationToken); await agent.NotifyProvidersOfNewMessagesAsync(session, newRequestMessages, response.Messages, options, cancellationToken).ConfigureAwait(false); MarkAsPersisted(newRequestMessages); MarkAsPersisted(response.Messages); } return response; } /// public override async IAsyncEnumerable GetStreamingResponseAsync( IEnumerable messages, ChatOptions? options = null, [EnumeratorCancellation] CancellationToken cancellationToken = default) { var (agent, session) = GetRequiredAgentAndSession(); options = StripLocalHistoryConversationId(options); List responseUpdates = []; IAsyncEnumerator enumerator; try { enumerator = base.GetStreamingResponseAsync(messages, options, cancellationToken).GetAsyncEnumerator(cancellationToken); } catch (Exception ex) { var newRequestMessagesOnFailure = GetNewRequestMessages(messages); await agent.NotifyProvidersOfFailureAsync(session, ex, newRequestMessagesOnFailure, options, cancellationToken).ConfigureAwait(false); throw; } bool hasUpdates; try { hasUpdates = await enumerator.MoveNextAsync().ConfigureAwait(false); } catch (Exception ex) { var newRequestMessagesOnFailure = GetNewRequestMessages(messages); await agent.NotifyProvidersOfFailureAsync(session, ex, newRequestMessagesOnFailure, options, cancellationToken).ConfigureAwait(false); throw; } while (hasUpdates) { var update = enumerator.Current; responseUpdates.Add(update); yield return update; try { hasUpdates = await enumerator.MoveNextAsync().ConfigureAwait(false); } catch (Exception ex) { var newRequestMessagesOnFailure = GetNewRequestMessages(messages); await agent.NotifyProvidersOfFailureAsync(session, ex, newRequestMessagesOnFailure, options, cancellationToken).ConfigureAwait(false); throw; } } var chatResponse = responseUpdates.ToChatResponse(); var newRequestMessages = GetNewRequestMessages(messages); if (this.ShouldDeferPersistence(options)) { // In mark-only mode or when resuming from a continuation token, just mark messages // for later persistence by ChatClientAgent. Conversation ID and provider notification // are deferred to end-of-run. For continuation tokens, the end-of-run handler needs // to send the combined data from both the previous and current runs. MarkAsPersisted(newRequestMessages); MarkAsPersisted(chatResponse.Messages); } else { // In persist mode, persist immediately and update conversation ID. agent.UpdateSessionConversationId(session, chatResponse.ConversationId, cancellationToken); await agent.NotifyProvidersOfNewMessagesAsync(session, newRequestMessages, chatResponse.Messages, options, cancellationToken).ConfigureAwait(false); MarkAsPersisted(newRequestMessages); MarkAsPersisted(chatResponse.Messages); } } /// /// Gets the current and from the run context. /// private static (ChatClientAgent Agent, ChatClientAgentSession Session) GetRequiredAgentAndSession() { var runContext = AIAgent.CurrentRunContext ?? throw new InvalidOperationException( $"{nameof(ChatHistoryPersistingChatClient)} can only be used within the context of a running AIAgent. " + "Ensure that the chat client is being invoked as part of an AIAgent.RunAsync or AIAgent.RunStreamingAsync call."); var chatClientAgent = runContext.Agent.GetService() ?? throw new InvalidOperationException( $"{nameof(ChatHistoryPersistingChatClient)} can only be used with a {nameof(ChatClientAgent)}. " + $"The current agent is of type '{runContext.Agent.GetType().Name}'."); if (runContext.Session is not ChatClientAgentSession chatClientAgentSession) { throw new InvalidOperationException( $"{nameof(ChatHistoryPersistingChatClient)} requires a {nameof(ChatClientAgentSession)}. " + $"The current session is of type '{runContext.Session?.GetType().Name ?? "null"}'."); } return (chatClientAgent, chatClientAgentSession); } /// /// Determines whether persistence should be deferred to end-of-run instead of happening immediately. /// /// /// when in mode, when the call is resuming from /// a continuation token (since the end-of-run handler needs to combine data from the previous /// and current runs), or when background responses are allowed (since the caller may stop /// consuming the stream mid-run, preventing the post-stream persistence code from executing). /// private bool ShouldDeferPersistence(ChatOptions? options) { return this.MarkOnly || options?.ContinuationToken is not null || options?.AllowBackgroundResponses is true; } /// /// Returns only the request messages that have not yet been persisted to chat history. /// /// /// A message is considered already persisted if any of the following is true: /// /// It has the in its . /// It has an of /// (indicating it was loaded from chat history and does not need to be re-persisted). /// It has and all of its items have the /// in their . This handles the /// streaming case where reconstructs objects /// independently via ToChatResponse(), producing different object references that share the same /// underlying instances. /// /// /// A list of request messages that have not yet been persisted. /// The full set of request messages to filter. private static List GetNewRequestMessages(IEnumerable messages) { return messages.Where(m => !IsAlreadyPersisted(m)).ToList(); } /// /// Determines whether a message has already been persisted to chat history by this decorator. /// private static bool IsAlreadyPersisted(ChatMessage message) { if (message.AdditionalProperties?.TryGetValue(PersistedMarkerKey, out var value) == true && value is true) { return true; } if (message.GetAgentRequestMessageSourceType() == AgentRequestMessageSourceType.ChatHistory) { return true; } // In streaming mode, FunctionInvokingChatClient reconstructs ChatMessage objects via ToChatResponse() // independently, producing different ChatMessage instances. However, the underlying AIContent objects // (e.g., FunctionCallContent, FunctionResultContent) are shared references. Checking for markers on // AIContent handles dedup in this case. if (message.Contents.Count > 0 && message.Contents.All(c => c.AdditionalProperties?.TryGetValue(PersistedMarkerKey, out var value) == true && value is true)) { return true; } return false; } /// /// Marks the given messages as persisted by setting a marker on both the /// and each of its items. /// /// /// Both levels are marked because may reconstruct /// objects in streaming mode (losing the message-level marker), /// but the references are shared and retain their markers. /// /// The messages to mark as persisted. private static void MarkAsPersisted(IEnumerable messages) { foreach (var message in messages) { message.AdditionalProperties ??= new(); message.AdditionalProperties[PersistedMarkerKey] = true; foreach (var content in message.Contents) { content.AdditionalProperties ??= new(); content.AdditionalProperties[PersistedMarkerKey] = true; } } } /// /// If the carry the sentinel, /// returns a clone with the conversation ID cleared so the inner client never sees it. /// Otherwise returns the original unchanged. /// private static ChatOptions? StripLocalHistoryConversationId(ChatOptions? options) { if (options?.ConversationId == LocalHistoryConversationId) { options = options.Clone(); options.ConversationId = null; } return options; } }