diff --git a/dotnet/src/Microsoft.Agents.AI.Foundry.Hosting/AgentFrameworkResponseHandler.cs b/dotnet/src/Microsoft.Agents.AI.Foundry.Hosting/AgentFrameworkResponseHandler.cs index 538d424ad4f..0472fcb38ff 100644 --- a/dotnet/src/Microsoft.Agents.AI.Foundry.Hosting/AgentFrameworkResponseHandler.cs +++ b/dotnet/src/Microsoft.Agents.AI.Foundry.Hosting/AgentFrameworkResponseHandler.cs @@ -3,6 +3,7 @@ using System; using System.Collections.Generic; using System.Diagnostics.CodeAnalysis; +using System.Linq; using System.Runtime.CompilerServices; using System.Security.Cryptography; using System.Threading; @@ -33,6 +34,16 @@ public class AgentFrameworkResponseHandler : ResponseHandler /// private static readonly HostedSessionIsolationKeyProvider s_defaultIsolationKeyProvider = new PlatformHostedSessionIsolationKeyProvider(); + /// Identifies the handler as the source of chat history messages it passes as input. + private const string HistorySourceId = "Microsoft.Agents.AI.Foundry.Hosting.AgentFrameworkResponseHandler"; + + /// + /// The session type a hosted workflow runs with. It is internal to Microsoft.Agents.AI.Workflows, + /// so it is recognised by name: taking a reference to it would mean opening that package's internals, + /// which cannot be done here because both packages compile the same shared source files. + /// + private const string WorkflowSessionTypeName = "WorkflowSession"; + /// /// Initializes a new instance of the class /// that resolves agents from keyed DI services. @@ -112,16 +123,21 @@ public override async IAsyncEnumerable CreateAsync( // (resolvedUserId is null) there is no user to partition on, so the session is unscoped/shared // by design — per-user isolation applies only when a user identity was resolved (hosted). var conversationId = request.GetConversationId(); - var sessionConversationId = HostedConversationKey.Resolve( + var agentSessionId = HostedConversationKey.Resolve( conversationId, request.PreviousResponseId, context.ResponseId); - var chatClientAgent = agent.GetService(); + var agentOptions = agent.GetService(); + + // Load an existing session when there is a conversation key. The store returns null when + // nothing is persisted for it, which is the authoritative "this is a resume" signal: a + // non-null result means a prior turn saved this session. Whether loaded or created, the + // handler owns creating a fresh session when none exists, so the resume signal does not + // depend on inspecting the session for state the handler itself also writes to. + AgentSession? sessionLoadedFromStore = !string.IsNullOrWhiteSpace(agentSessionId) + ? await sessionStore.GetSessionAsync(agent, agentSessionId, resolvedUserId, cancellationToken).ConfigureAwait(false) + : null; - AgentSession? session = !string.IsNullOrWhiteSpace(sessionConversationId) - ? await sessionStore.GetSessionAsync(agent, sessionConversationId, resolvedUserId, cancellationToken).ConfigureAwait(false) - : chatClientAgent is not null - ? await chatClientAgent.CreateSessionAsync(cancellationToken).ConfigureAwait(false) - : await agent.CreateSessionAsync(cancellationToken).ConfigureAwait(false); + AgentSession? session = sessionLoadedFromStore ?? await agent.CreateSessionAsync(cancellationToken).ConfigureAwait(false); // Capture the platform per-request call id (x-agent-foundry-call-id, protocol 2.0.0 only). // It is re-applied to the ambient HostedCallContext immediately before each outbound egress @@ -153,6 +169,19 @@ public override async IAsyncEnumerable CreateAsync( } } + // A hosted agent's conversation is recorded by the AgentServer SDK's own storage provider. A + // conversation id on the session means the service behind the agent's chat client is recording + // a second one, which nothing here reads and which no one reconciles with the first. Refuse + // before any work is done, as a plain bad request rather than a failure part way through. + if (session is ChatClientAgentSession { ConversationId: not null }) + { + throw new ResponsesApiException( + new Error( + "service_managed_chat_history_not_supported", + "Chat history is managed by the hosted agent service, therefore using a ChatClientAgent with its own service storage is not supported. Configure the agent's chat client so the underlying service does not store responses."), + 400); + } + // 3. Create the SDK event stream builder var stream = new ResponseEventStream(context, request); @@ -163,18 +192,17 @@ public override async IAsyncEnumerable CreateAsync( // 4. Convert input: history + current input → ChatMessage[] var messages = new List(); - // Load conversation history only for fresh sessions. When a session already exists - // (e.g. resuming a workflow paused at an external-input port), the workflow's - // checkpointed state already contains the prior turns' messages — replaying history - // would re-drive completed actions and break HITL resume semantics. - var isResume = (!string.IsNullOrWhiteSpace(conversationId) || !string.IsNullOrWhiteSpace(request.PreviousResponseId)) - && session?.StateBag?.Count > 0; - if (!isResume) + // Add the chat history to the request. Workflow sessions accumulate previous turns and must not + // get the full history again; their types are internal, hence the check on the type name. + if (sessionLoadedFromStore is null + || !string.Equals(sessionLoadedFromStore.GetType().Name, WorkflowSessionTypeName, StringComparison.Ordinal)) { var history = await context.GetHistoryAsync(cancellationToken).ConfigureAwait(false); if (history.Count > 0) { - messages.AddRange(InputConverter.ConvertOutputItemsToMessages(history, session?.StateBag)); + messages.AddRange(InputConverter + .ConvertOutputItemsToMessages(history, session?.StateBag) + .Select(m => m.WithAgentRequestMessageSource(AgentRequestMessageSourceType.ChatHistory, HistorySourceId))); } } @@ -191,9 +219,16 @@ public override async IAsyncEnumerable CreateAsync( } // 5. Build chat options - var chatOptions = InputConverter.ConvertToChatOptions(request); + var chatOptions = InputConverter.ConvertToChatOptions(request, agentOptions?.ChatOptions?.RawRepresentationFactory); chatOptions.Instructions = request.Instructions; + // Everything the agent needs for this turn is already in the input, so the provider it would + // otherwise run is replaced for the duration by one that keeps its messages in memory and is + // dropped when the run ends. Serving from a longer-lived one would deliver the conversation + // twice, and storing into it would leave a copy the hosting service never sees. + chatOptions.AdditionalProperties ??= []; + chatOptions.AdditionalProperties.Add(new VolatileChatHistoryProvider()); + // Inject Foundry Toolbox tools when the toolbox service is available. // // Two sources are considered: @@ -445,9 +480,9 @@ await this._toolboxService // Persist session after streaming completes (successful or not). The user id partitions the // persisted session per end user, mirroring the load above so multi-turn continuity is preserved. - if (session is not null && !string.IsNullOrWhiteSpace(sessionConversationId)) + if (session is not null && !string.IsNullOrWhiteSpace(agentSessionId)) { - await sessionStore.SaveSessionAsync(agent, sessionConversationId, session, resolvedUserId, cancellationToken).ConfigureAwait(false); + await sessionStore.SaveSessionAsync(agent, agentSessionId, session, resolvedUserId, cancellationToken).ConfigureAwait(false); } } } diff --git a/dotnet/src/Microsoft.Agents.AI.Foundry.Hosting/AgentSessionStore.cs b/dotnet/src/Microsoft.Agents.AI.Foundry.Hosting/AgentSessionStore.cs index d1c93dc274c..d507db4d966 100644 --- a/dotnet/src/Microsoft.Agents.AI.Foundry.Hosting/AgentSessionStore.cs +++ b/dotnet/src/Microsoft.Agents.AI.Foundry.Hosting/AgentSessionStore.cs @@ -1,5 +1,6 @@ // Copyright (c) Microsoft. All rights reserved. +using System; using System.Diagnostics.CodeAnalysis; using System.Threading; using System.Threading.Tasks; @@ -42,7 +43,8 @@ public abstract ValueTask SaveSessionAsync( CancellationToken cancellationToken = default); /// - /// Retrieves a serialized agent session from persistent storage. + /// Retrieves a serialized agent session from persistent storage, or when + /// no session is stored for the given identifiers. /// /// The agent that owns this session. /// The unique identifier for the conversation/session to retrieve. @@ -55,12 +57,41 @@ public abstract ValueTask SaveSessionAsync( /// /// The to monitor for cancellation requests. /// - /// A task that represents the asynchronous retrieval operation. - /// The task result contains the session, or a new session if not found. + /// A task that represents the asynchronous retrieval operation. The task result contains the restored + /// session, or when nothing is stored for the given identifiers. This is a plain + /// lookup: it never creates a session. Use to get a ready-to-use + /// session (loading an existing one or creating a new one), and use this method when the caller needs to + /// distinguish a resumed session from a fresh one (a non-null result means a prior turn established it). /// - public abstract ValueTask GetSessionAsync( + public abstract ValueTask GetSessionAsync( AIAgent agent, string conversationId, string? userId, CancellationToken cancellationToken = default); + + /// + /// Retrieves the stored session for the given identifiers, or creates a new one via + /// when none is stored. + /// + /// The agent that owns this session. + /// The unique identifier for the conversation/session to retrieve. + /// The per-user partition key; see for its meaning. + /// The to monitor for cancellation requests. + /// A task whose result is always a usable session, never . + /// + /// This is the convenience path for callers that only need a session to work with and do not care whether + /// it was loaded or freshly created. It is implemented in terms of , so a + /// store overriding that method gets this behavior for free. + /// + public virtual async ValueTask GetOrCreateSessionAsync( + AIAgent agent, + string conversationId, + string? userId, + CancellationToken cancellationToken = default) + { + ArgumentNullException.ThrowIfNull(agent); + + return await this.GetSessionAsync(agent, conversationId, userId, cancellationToken).ConfigureAwait(false) + ?? await agent.CreateSessionAsync(cancellationToken).ConfigureAwait(false); + } } diff --git a/dotnet/src/Microsoft.Agents.AI.Foundry.Hosting/FileSystemAgentSessionStore.cs b/dotnet/src/Microsoft.Agents.AI.Foundry.Hosting/FileSystemAgentSessionStore.cs index 17c9a184a35..c7c3b7292d4 100644 --- a/dotnet/src/Microsoft.Agents.AI.Foundry.Hosting/FileSystemAgentSessionStore.cs +++ b/dotnet/src/Microsoft.Agents.AI.Foundry.Hosting/FileSystemAgentSessionStore.cs @@ -208,7 +208,7 @@ private string BuildNotWritableMessage(string sessionFilePath) => $"(for example {nameof(InMemoryAgentSessionStore)}) via AddFoundryResponses(agent, agentSessionStore)."; /// - public override async ValueTask GetSessionAsync(AIAgent agent, string conversationId, string? userId, CancellationToken cancellationToken = default) + public override async ValueTask GetSessionAsync(AIAgent agent, string conversationId, string? userId, CancellationToken cancellationToken = default) { ArgumentNullException.ThrowIfNull(agent); ArgumentException.ThrowIfNullOrWhiteSpace(conversationId); @@ -216,13 +216,13 @@ public override async ValueTask GetSessionAsync(AIAgent agent, str string path = this.GetSessionPath(agent, conversationId, userId); if (!File.Exists(path)) { - return await agent.CreateSessionAsync(cancellationToken).ConfigureAwait(false); + return null; } byte[] bytes = await File.ReadAllBytesAsync(path, cancellationToken).ConfigureAwait(false); if (bytes.Length == 0) { - return await agent.CreateSessionAsync(cancellationToken).ConfigureAwait(false); + return null; } // Parse and clone so the document buffer can be released. diff --git a/dotnet/src/Microsoft.Agents.AI.Foundry.Hosting/InMemoryAgentSessionStore.cs b/dotnet/src/Microsoft.Agents.AI.Foundry.Hosting/InMemoryAgentSessionStore.cs index 3b3b52fda57..579240432b6 100644 --- a/dotnet/src/Microsoft.Agents.AI.Foundry.Hosting/InMemoryAgentSessionStore.cs +++ b/dotnet/src/Microsoft.Agents.AI.Foundry.Hosting/InMemoryAgentSessionStore.cs @@ -40,16 +40,15 @@ public override async ValueTask SaveSessionAsync(AIAgent agent, string conversat } /// - public override async ValueTask GetSessionAsync(AIAgent agent, string conversationId, string? userId, CancellationToken cancellationToken = default) + public override async ValueTask GetSessionAsync(AIAgent agent, string conversationId, string? userId, CancellationToken cancellationToken = default) { var key = GetKey(agent, conversationId, userId); - JsonElement? sessionContent = this._sessions.TryGetValue(key, out var existingSession) ? existingSession : null; - - return sessionContent switch + if (!this._sessions.TryGetValue(key, out var existingSession)) { - null => await agent.CreateSessionAsync(cancellationToken).ConfigureAwait(false), - _ => await agent.DeserializeSessionAsync(sessionContent.Value, cancellationToken: cancellationToken).ConfigureAwait(false), - }; + return null; + } + + return await agent.DeserializeSessionAsync(existingSession, cancellationToken: cancellationToken).ConfigureAwait(false); } // Keyed with the same a-/u-/c- prefix scheme as FileSystemAgentSessionStore so the in-memory store diff --git a/dotnet/src/Microsoft.Agents.AI.Foundry.Hosting/InputConverter.cs b/dotnet/src/Microsoft.Agents.AI.Foundry.Hosting/InputConverter.cs index e104487df76..3a4e95b041c 100644 --- a/dotnet/src/Microsoft.Agents.AI.Foundry.Hosting/InputConverter.cs +++ b/dotnet/src/Microsoft.Agents.AI.Foundry.Hosting/InputConverter.cs @@ -7,6 +7,8 @@ using System.Text.Json; using Azure.AI.AgentServer.Responses.Models; using Microsoft.Extensions.AI; +using ChatCompletionOptions = OpenAI.Chat.ChatCompletionOptions; +using CreateResponseOptions = OpenAI.Responses.CreateResponseOptions; using MeaiTextContent = Microsoft.Extensions.AI.TextContent; using SdkTextContent = Azure.AI.AgentServer.Responses.Models.TextContent; @@ -87,10 +89,14 @@ public static List ConvertOutputItemsToMessages(IReadOnlyList from the SDK request properties. /// /// The create response request. + /// + /// The factory the agent carries on its own , if any, so a request that has + /// to set one of its own can run it rather than replace it. + /// /// A configured instance. - public static ChatOptions ConvertToChatOptions(CreateResponse request) + public static ChatOptions ConvertToChatOptions(CreateResponse request, Func? agentRawRepresentationFactory = null) { - return new ChatOptions + var options = new ChatOptions { Temperature = (float?)request.Temperature, TopP = (float?)request.TopP, @@ -100,6 +106,43 @@ public static ChatOptions ConvertToChatOptions(CreateResponse request) // the client-provided model would override it (causing failures when // clients send placeholder values like "hosted-agent"). }; + + // The service behind the agent's chat client is never asked to store a response. Recording a + // hosted turn is the AgentServer SDK's job, done by its storage provider around this handler, + // and a second recording downstream is a conversation nothing here reads and no one reconciles. + // The caller's own store flag is not carried across: it says what the hosting service should + // record, which is a separate question and one this handler has no say in. + // + // Both OpenAI request shapes carry the setting, so a chat client speaking either protocol is + // covered. Anything else is a request type with no notion of storing a response, and is handed + // back untouched; such a client keeping a conversation of its own is caught later by the + // conversation id check in the handler. + // + // The agent's own factory is invoked here and its result is what gets the setting, because + // ChatClientAgent chains the two by taking the agent's only when the request's returns null + // (ChatClientAgent.PrepareChatOptions). A request factory that always answers would otherwise + // drop whatever the container configured. + options.RawRepresentationFactory = chatClient => + { + switch (agentRawRepresentationFactory?.Invoke(chatClient)) + { + case CreateResponseOptions responseOptions: + responseOptions.StoredOutputEnabled = false; + return responseOptions; + + case ChatCompletionOptions completionOptions: + completionOptions.StoredOutputEnabled = false; + return completionOptions; + + case { } configuredByTheAgent: + return configuredByTheAgent; + + default: + return new CreateResponseOptions { StoredOutputEnabled = false }; + } + }; + + return options; } /// diff --git a/dotnet/src/Microsoft.Agents.AI.Foundry.Hosting/VolatileChatHistoryProvider.cs b/dotnet/src/Microsoft.Agents.AI.Foundry.Hosting/VolatileChatHistoryProvider.cs new file mode 100644 index 00000000000..3fc912665df --- /dev/null +++ b/dotnet/src/Microsoft.Agents.AI.Foundry.Hosting/VolatileChatHistoryProvider.cs @@ -0,0 +1,55 @@ +// Copyright (c) Microsoft. All rights reserved. + +using System.Collections.Generic; +using System.Threading; +using System.Threading.Tasks; +using Microsoft.Extensions.AI; + +namespace Microsoft.Agents.AI.Foundry.Hosting; + +/// +/// A that holds the turn's messages in a field, for the lifetime of +/// one request and no longer. +/// +/// +/// +/// A hosted agent's conversation is recorded by the AgentServer SDK's own storage provider, which +/// writes every turn the caller asked it to store and serves it back through +/// . That happens around +/// the handler, not through it. The handler reads the conversation from there and passes it in as the +/// run's input, so nothing has to be carried between requests, and a provider storing anything of its +/// own would only add a copy the storage provider never sees. +/// +/// +/// Within a single run the provider still does its ordinary work: an agent calling tools goes back to +/// the chat client several times, and each of those calls needs the messages the earlier ones produced. +/// Those live here until the run ends and the instance is dropped. +/// +/// +/// Supplied as a run-scoped override through , so it takes +/// the place of the agent's own provider for the turn without changing the agent. An agent that does not +/// read its history through a provider ignores it. +/// +/// +internal sealed class VolatileChatHistoryProvider : ChatHistoryProvider +{ + private readonly List _messages = []; + + /// + protected override ValueTask> ProvideChatHistoryAsync(InvokingContext context, CancellationToken cancellationToken = default) + => new(this._messages); + + /// + protected override ValueTask StoreChatHistoryAsync(InvokedContext context, CancellationToken cancellationToken = default) + { + // Only what this run produced arrives here: the base class filters out everything already marked + // as chat history, which covers the turns the handler took from the storage provider. + this._messages.AddRange(context.RequestMessages); + if (context.ResponseMessages is not null) + { + this._messages.AddRange(context.ResponseMessages); + } + + return default; + } +} diff --git a/dotnet/tests/Foundry.Hosting.IntegrationTests.TestContainer/DownstreamConversationReportingAgent.cs b/dotnet/tests/Foundry.Hosting.IntegrationTests.TestContainer/DownstreamConversationReportingAgent.cs new file mode 100644 index 00000000000..899c4f0d360 --- /dev/null +++ b/dotnet/tests/Foundry.Hosting.IntegrationTests.TestContainer/DownstreamConversationReportingAgent.cs @@ -0,0 +1,55 @@ +// Copyright (c) Microsoft. All rights reserved. + +using System.Runtime.CompilerServices; +using Microsoft.Agents.AI; +using Microsoft.Extensions.AI; + +namespace Foundry.Hosting.IntegrationTests.TestContainer; + +/// +/// Wraps the container's agent and tells the caller which conversation the agent's own run left behind +/// on the service, by appending DOWNSTREAM_ID=<id> to the reply. +/// +/// +/// +/// The platform already records every hosted turn around the handler, and that is the conversation the +/// caller reads. The agent's run inside the container talks to its own service, and if that service is +/// asked to keep the turn it writes a second record, on a trail of its own that the caller never sees. +/// +/// +/// After the run, the id of that trail is on the session, so reporting it is enough for a test to go +/// look for it on the service. No id means the container asked for nothing to be kept. +/// +/// +internal sealed class DownstreamConversationReportingAgent(AIAgent innerAgent) : DelegatingAIAgent(innerAgent) +{ + /// + /// Marker that carries the id. Tests read the value that follows it. + /// + public const string IdPrefix = "DOWNSTREAM_ID="; + + /// + /// Value reported when the agent's run left nothing behind on the service. + /// + public const string NoId = "none"; + + /// + protected override async IAsyncEnumerable RunCoreStreamingAsync( + IEnumerable messages, + AgentSession? session = null, + AgentRunOptions? options = null, + [EnumeratorCancellation] CancellationToken cancellationToken = default) + { + await foreach (var update in this.InnerAgent + .RunStreamingAsync(messages, session, options, cancellationToken) + .ConfigureAwait(false)) + { + yield return update; + } + + var downstreamId = (session as ChatClientAgentSession)?.ConversationId; + yield return new AgentResponseUpdate( + ChatRole.Assistant, + $" {IdPrefix}{(string.IsNullOrWhiteSpace(downstreamId) ? NoId : downstreamId)}"); + } +} diff --git a/dotnet/tests/Foundry.Hosting.IntegrationTests.TestContainer/Program.cs b/dotnet/tests/Foundry.Hosting.IntegrationTests.TestContainer/Program.cs index 6bea914f978..6f491c7290f 100644 --- a/dotnet/tests/Foundry.Hosting.IntegrationTests.TestContainer/Program.cs +++ b/dotnet/tests/Foundry.Hosting.IntegrationTests.TestContainer/Program.cs @@ -6,6 +6,7 @@ using Azure.Identity; using Azure.Search.Documents; using Azure.Search.Documents.Models; +using Foundry.Hosting.IntegrationTests.TestContainer; using Microsoft.Agents.AI; using Microsoft.Agents.AI.Foundry; using Microsoft.Agents.AI.Foundry.Hosting; @@ -34,6 +35,7 @@ "happy-path" => CreateHappyPathAgent(projectClient, deployment), "unsupported-protocol" => CreateHappyPathAgent(projectClient, deployment), "store-config" => CreateStoreConfigAgent(projectClient, deployment), + "downstream-store" => CreateDownstreamStoreAgent(projectClient, deployment), "tool-calling" => CreateToolCallingAgent(projectClient, deployment), "tool-calling-approval" => CreateToolCallingApprovalAgent(projectClient, deployment), "mcp-toolbox" => CreateMcpToolboxAgent(projectClient, deployment), @@ -89,6 +91,19 @@ static AIAgent CreateStoreConfigAgent(AIProjectClient client, string deployment) name: "store-config-agent", description: "Store and session semantics test agent."); +// downstream-store scenario: an ordinary Foundry ChatClientAgent, like the first hosted agent sample, +// wrapped so the caller is told which conversation the agent's own run left behind on the service. The +// platform already records the hosted turn in the caller's conversation; anything the agent's run also +// leaves behind is a second copy of the same turn, on a trail nobody reads. +static AIAgent CreateDownstreamStoreAgent(AIProjectClient client, string deployment) => + new DownstreamConversationReportingAgent( + client.AsAIAgent( + model: deployment, + instructions: "You are a helpful assistant. Answer the user's question concisely and accurately, " + + "and use any facts the user told you earlier in the conversation.", + name: "downstream-store-agent", + description: "Downstream store test agent.")); + static AIAgent CreateToolCallingAgent(AIProjectClient client, string deployment) => client.AsAIAgent( model: deployment, diff --git a/dotnet/tests/Foundry.Hosting.IntegrationTests/Fixtures/DownstreamStoreHostedAgentFixture.cs b/dotnet/tests/Foundry.Hosting.IntegrationTests/Fixtures/DownstreamStoreHostedAgentFixture.cs new file mode 100644 index 00000000000..d5033a1e913 --- /dev/null +++ b/dotnet/tests/Foundry.Hosting.IntegrationTests/Fixtures/DownstreamStoreHostedAgentFixture.cs @@ -0,0 +1,14 @@ +// Copyright (c) Microsoft. All rights reserved. + +namespace Foundry.Hosting.IntegrationTests.Fixtures; + +/// +/// Provisions a hosted agent that runs the test container in IT_SCENARIO=downstream-store mode. +/// Used by HostedDownstreamStoreTests. The container runs an ordinary Foundry +/// ChatClientAgent and reports back which conversation its own run left behind on the service, +/// so the test can check whether a second copy of the turn was kept. +/// +public sealed class DownstreamStoreHostedAgentFixture : HostedAgentFixture +{ + protected override string ScenarioName => "downstream-store"; +} diff --git a/dotnet/tests/Foundry.Hosting.IntegrationTests/Fixtures/HostedAgentFixture.cs b/dotnet/tests/Foundry.Hosting.IntegrationTests/Fixtures/HostedAgentFixture.cs index 1725aa4831f..5b895d69643 100644 --- a/dotnet/tests/Foundry.Hosting.IntegrationTests/Fixtures/HostedAgentFixture.cs +++ b/dotnet/tests/Foundry.Hosting.IntegrationTests/Fixtures/HostedAgentFixture.cs @@ -3,6 +3,7 @@ using System; using System.ClientModel.Primitives; using System.Collections.Generic; +using System.Linq; using System.Threading; using System.Threading.Tasks; using AgentConformance.IntegrationTests.Support; @@ -11,6 +12,7 @@ using Azure.AI.Projects.Agents; using Microsoft.Agents.AI; using Microsoft.Extensions.AI; +using OpenAI.Responses; using Shared.IntegrationTests; namespace Foundry.Hosting.IntegrationTests.Fixtures; @@ -146,6 +148,83 @@ public async Task CountConversationItemsAsync(string conversationId) return count; } + /// + /// Reads every message stored in a conversation, oldest first, as a role and text pair. Used by + /// tests that need to see how many times a given turn was recorded, not just how many items there + /// are. + /// + public async Task> ReadConversationMessagesAsync(string conversationId) + { + List<(string Role, string Text)> messages = []; + await foreach (AgentResponseItem item in this.AgentOpenAIClient.GetProjectConversationsClient().GetProjectConversationItemsAsync(conversationId, order: "asc").ConfigureAwait(false)) + { + if (item.AsResponseResultItem() is MessageResponseItem message) + { + var text = string.Concat(message.Content + .Where(c => c.Kind is ResponseContentPartKind.OutputText or ResponseContentPartKind.InputText) + .Select(c => c.Text)); + + messages.Add((message.Role.ToString(), text)); + } + } + + return messages; + } + + /// + /// Reads the input a stored response was run with, oldest first, as a role and text pair. Along a + /// previous_response_id chain this is what the turn actually received, so tests can see + /// whether an earlier turn was handed to it more than once. + /// + public async Task> ReadResponseInputMessagesAsync(string responseId) + { + List<(string Role, string Text)> messages = []; + await foreach (ResponseItem item in this.AgentOpenAIClient.GetProjectResponsesClient().GetResponseInputItemsAsync(responseId).ConfigureAwait(false)) + { + if (item is MessageResponseItem message) + { + var text = string.Concat(message.Content + .Where(c => c.Kind is ResponseContentPartKind.OutputText or ResponseContentPartKind.InputText) + .Select(c => c.Text)); + + messages.Add((message.Role.ToString(), text)); + } + } + + return messages; + } + + /// + /// Tries to read a response back off the service by id, returning when + /// nothing is stored under it. Both the project-wide client and this scenario's per-agent client + /// are tried, because a response created inside the container is not necessarily reachable through + /// the same endpoint as one created for the caller. + /// + public async Task TryReadResponseAsync(string responseId) + { + foreach (var responses in new[] + { + this.ProjectClient.GetProjectOpenAIClient().GetProjectResponsesClient(), + this.AgentOpenAIClient.GetProjectResponsesClient(), + }) + { + try + { + var response = await responses.GetResponseAsync(responseId).ConfigureAwait(false); + if (response?.Value is not null) + { + return response.Value; + } + } + catch + { + // Not readable through this endpoint; try the next one. + } + } + + return null; + } + public async ValueTask InitializeAsync() { var endpoint = new Uri(TestConfiguration.GetRequiredValue(TestSettings.AzureAIProjectEndpoint)); diff --git a/dotnet/tests/Foundry.Hosting.IntegrationTests/HostedDownstreamStoreTests.cs b/dotnet/tests/Foundry.Hosting.IntegrationTests/HostedDownstreamStoreTests.cs new file mode 100644 index 00000000000..ae820b314c5 --- /dev/null +++ b/dotnet/tests/Foundry.Hosting.IntegrationTests/HostedDownstreamStoreTests.cs @@ -0,0 +1,123 @@ +// Copyright (c) Microsoft. All rights reserved. + +using System; +using System.Threading.Tasks; +using Foundry.Hosting.IntegrationTests.Fixtures; +using Microsoft.Agents.AI; + +namespace Foundry.Hosting.IntegrationTests; + +/// +/// Proves a hosted turn is kept once. +/// +/// +/// +/// The AgentServer SDK's storage provider records every hosted turn around the container's handler, +/// and that record is the conversation the caller reads. The agent's own run inside the container +/// talks to its own service, and when that service is asked to keep the turn it writes a second copy +/// of the same exchange, on a trail of its own that nobody reads and nobody reconciles. The caller's +/// conversation looks clean, so the second copy goes unnoticed. +/// +/// +/// The container agent here is an ordinary Foundry ChatClientAgent, like the first hosted agent +/// sample. It is wrapped so that after the run it appends DOWNSTREAM_ID=<id> to the reply, +/// carrying whatever its own run left behind on the service. The tests then go looking for that id: +/// finding it means a second copy exists. +/// +/// +[Trait("Category", "FoundryHostedAgents")] +public sealed class HostedDownstreamStoreTests(DownstreamStoreHostedAgentFixture fixture) : IClassFixture +{ + private const string IdPrefix = "DOWNSTREAM_ID="; + private const string NoId = "none"; + + private readonly DownstreamStoreHostedAgentFixture _fixture = fixture; + + [Fact] + public async Task StoredTurn_LeavesNothingBehindOnTheAgentsOwnServiceAsync() + { + // Arrange: a session bound to a conversation, which is how a caller keeps a hosted agent on one + // thread. The session carries the conversation, so no per-run options are needed. + var agent = this._fixture.Agent; + var chatClientAgent = agent.GetService(); + Assert.NotNull(chatClientAgent); + + var conversationId = await this._fixture.CreateConversationAsync(); + try + { + var session = await chatClientAgent.CreateSessionAsync(conversationId); + + // Act: one stored turn, the way any caller would send it. + var response = await agent.RunAsync("Reply with the word 'ack'.", session); + + // Assert: the caller's conversation holds the turn, so it was recorded once already. + var recorded = await this._fixture.ReadConversationMessagesAsync(conversationId); + Assert.NotEmpty(recorded); + + // And the agent's own run left nothing behind that can be read back off the service. + await this.AssertNothingWasLeftBehindAsync(response.Text); + } + finally + { + await this._fixture.DeleteConversationAsync(conversationId); + } + } + + [Fact] + public async Task MultiTurn_LeavesNothingBehindOnTheAgentsOwnServiceAsync() + { + // Arrange: the agent's own default session, with nothing set up ahead of time. Whatever the + // hosted agent keeps for the caller lands on the session once the first turn comes back. + var agent = this._fixture.Agent; + var session = await agent.CreateSessionAsync(); + + // Act + var first = await agent.RunAsync("Remember the number 73. Acknowledge briefly.", session); + var second = await agent.RunAsync("What number did I just tell you?", session); + + // Assert: the conversation works, so history is reaching the model. + Assert.Contains("73", second.Text); + + // The hosted agent handed the caller something to continue from, and it is on the session. + var keptForTheCaller = (session as ChatClientAgentSession)?.ConversationId; + Assert.False( + string.IsNullOrWhiteSpace(keptForTheCaller), + "The hosted agent did not hand the caller anything to continue the conversation from."); + + // Every turn's own run, though, left nothing behind on the service. + await this.AssertNothingWasLeftBehindAsync(first.Text); + await this.AssertNothingWasLeftBehindAsync(second.Text); + } + + /// + /// Fails when the id the container reported still resolves on the service, which means the agent's + /// own run kept a second copy of a turn the platform had already recorded. + /// + private async Task AssertNothingWasLeftBehindAsync(string? replyText) + { + var downstreamId = ParseDownstreamId(replyText); + if (downstreamId is null) + { + return; + } + + var found = await this._fixture.TryReadResponseAsync(downstreamId); + Assert.True( + found is null, + $"The agent's own run left a second copy of the turn on the service, readable as '{downstreamId}'."); + } + + /// + /// Reads the id the container reported, or when the run left nothing behind. + /// + private static string? ParseDownstreamId(string? text) + { + Assert.False(string.IsNullOrWhiteSpace(text)); + + var marker = text!.IndexOf(IdPrefix, StringComparison.Ordinal); + Assert.True(marker >= 0, $"Expected the container to report '{IdPrefix}...' but got: {text}"); + + var value = text[(marker + IdPrefix.Length)..].Trim(); + return value.Length == 0 || value.Equals(NoId, StringComparison.Ordinal) ? null : value; + } +} diff --git a/dotnet/tests/Foundry.Hosting.IntegrationTests/README.md b/dotnet/tests/Foundry.Hosting.IntegrationTests/README.md index 589bd6cd060..234dee03a55 100644 --- a/dotnet/tests/Foundry.Hosting.IntegrationTests/README.md +++ b/dotnet/tests/Foundry.Hosting.IntegrationTests/README.md @@ -208,6 +208,7 @@ human-only operation; CI only adds and deletes versions under existing agents. | --- | --- | --- | --- | | `HappyPathHostedAgentFixture` | `happy-path` | `it-happy-path` | Round trip, streaming, and container-instruction behaviour. | | `HostedResponsesStoreConfigFixture` | `store-config` | `it-store-config` | Store/session semantics: `store=true` vs `store=false`, `previous_response_id` and `conversation_id` forks (read history without appending), multi-turn recall. | +| `DownstreamStoreHostedAgentFixture` | `downstream-store` | `it-downstream-store` | An ordinary Foundry `ChatClientAgent` that reports back which conversation its own run left behind on the service, so the test can assert the container does not keep a second copy of a turn the platform already recorded. | | `ToolCallingHostedAgentFixture` | `tool-calling` | `it-tool-calling` | Server side AIFunction invocation; arguments; multi turn referencing prior tool result. | | `ToolCallingApprovalHostedAgentFixture` | `tool-calling-approval` | `it-tool-calling-approval` | Approval requests raised, approved, denied. | | `McpToolboxHostedAgentFixture` | `mcp-toolbox` | `it-mcp-toolbox` | MCP backed tool invocation against `https://learn.microsoft.com/api/mcp` (placeholder). | diff --git a/dotnet/tests/Foundry.Hosting.IntegrationTests/scripts/it-bootstrap-agents.ps1 b/dotnet/tests/Foundry.Hosting.IntegrationTests/scripts/it-bootstrap-agents.ps1 index d22f8661965..1b200e42966 100644 --- a/dotnet/tests/Foundry.Hosting.IntegrationTests/scripts/it-bootstrap-agents.ps1 +++ b/dotnet/tests/Foundry.Hosting.IntegrationTests/scripts/it-bootstrap-agents.ps1 @@ -42,6 +42,7 @@ $ErrorActionPreference = 'Stop' $Scenarios = @( 'happy-path', 'store-config', + 'downstream-store', 'tool-calling', 'tool-calling-approval', 'mcp-toolbox', diff --git a/dotnet/tests/Microsoft.Agents.AI.Foundry.Hosting.UnitTests/AgentFrameworkResponseHandlerTests.cs b/dotnet/tests/Microsoft.Agents.AI.Foundry.Hosting.UnitTests/AgentFrameworkResponseHandlerTests.cs index 65aa9ce9ade..ae711b6902b 100644 --- a/dotnet/tests/Microsoft.Agents.AI.Foundry.Hosting.UnitTests/AgentFrameworkResponseHandlerTests.cs +++ b/dotnet/tests/Microsoft.Agents.AI.Foundry.Hosting.UnitTests/AgentFrameworkResponseHandlerTests.cs @@ -15,6 +15,8 @@ using Microsoft.Extensions.Logging; using Microsoft.Extensions.Logging.Abstractions; using Moq; +using ChatCompletionOptions = OpenAI.Chat.ChatCompletionOptions; +using CreateResponseOptions = OpenAI.Responses.CreateResponseOptions; using MeaiTextContent = Microsoft.Extensions.AI.TextContent; namespace Microsoft.Agents.AI.Foundry.Hosting.UnitTests; @@ -711,6 +713,574 @@ public async Task CreateAsync_DefaultAgent_IsAutoWrappedWithOpenTelemetryAsync() Assert.IsType(events[1]); } + #region Resume detection + + [Fact] + public async Task CreateAsync_FirstTurnOfAKnownConversation_StillReceivesTheServiceHistoryAsync() + { + // Arrange: the first turn this container serves for a conversation the service already holds + // history for. Nothing has been persisted for it yet, so this is not a resume: the history has + // to be handed to the agent, otherwise it answers knowing nothing of the conversation. + var agent = new CapturingAgent(); + var handler = BuildHandlerWith(agent, new FakeHostedSessionIsolationKeyProvider(), new InMemoryAgentSessionStore()); + var request = new CreateResponse { Model = "test" }; + request.Conversation = BinaryData.FromString("\"conv-known\""); + request.Input = BinaryData.FromObjectAsJson(new[] + { + new { type = "message", id = "msg_1", status = "completed", role = "user", + content = new[] { new { type = "input_text", text = "new question" } } } + }); + var ctx = new Mock("resp_" + new string('4', 46)) { CallBase = true }; + ctx.Setup(x => x.PlatformContext).Returns(new PlatformContext("alice", null)); + ctx.Setup(x => x.GetHistoryAsync(It.IsAny())) + .ReturnsAsync([NewHistoryMessageItem("msg_hist_1", "earlier turn")]); + ctx.Setup(x => x.GetInputItemsAsync(It.IsAny(), It.IsAny())).ReturnsAsync(Array.Empty()); + + // Act + await DrainEventsAsync(handler.CreateAsync(request, ctx.Object, CancellationToken.None)); + + // Assert: whether this is a resume is answered by the session store, not by looking for state on + // the session. The handler writes the caller's identity onto a session before this point, so a + // freshly created session already carries state and reading that as "it has run before" made the + // first turn of every conversation look like a resume, dropping its history. It only showed up + // when hosted, because there is no identity to write locally. + Assert.NotNull(agent.CapturedMessages); + Assert.Contains(agent.CapturedMessages!, m => m.Text.Contains("earlier turn", StringComparison.Ordinal)); + } + + [Fact] + public async Task CreateAsync_SecondTurnOfAWorkflow_DoesNotReplayTheServiceHistoryAsync() + { + // Arrange: a hosted workflow, whose session carries the conversation in its own state, and a + // first turn that persists it. + const string ConversationId = "conv-resumed"; + var agent = new WorkflowLikeAgent(); + var store = new InMemoryAgentSessionStore(); + var handler = BuildHandlerWith(agent, new FakeHostedSessionIsolationKeyProvider(), store); + await DrainEventsAsync(handler.CreateAsync( + NewConversationTurn(ConversationId, "first question"), + NewServingContext("resp_" + new string('5', 46), []), + CancellationToken.None)); + + // Act: a second turn of the same conversation, for which the service now reports history. + await DrainEventsAsync(handler.CreateAsync( + NewConversationTurn(ConversationId, "second question"), + NewServingContext("resp_" + new string('6', 46), [NewHistoryMessageItem("msg_hist_1", "first question")]), + CancellationToken.None)); + + // Assert: a workflow takes everything handed to it as newly arrived input, and its session + // already holds these turns, so handing them over again would re-drive work it has already done. + Assert.NotNull(agent.CapturedMessages); + Assert.DoesNotContain(agent.CapturedMessages!, m => m.Text.Contains("first question", StringComparison.Ordinal)); + } + + [Fact] + public async Task CreateAsync_SecondTurnOfAnAgentThatKeepsNothing_StillReceivesTheServiceHistoryAsync() + { + // Arrange: an agent written outside this repo that runs no chat history provider and keeps + // nothing in its session, with a first turn that persists one anyway. + const string ConversationId = "conv-keeps-nothing"; + var agent = new CapturingAgent(); + var store = new InMemoryAgentSessionStore(); + var handler = BuildHandlerWith(agent, new FakeHostedSessionIsolationKeyProvider(), store); + await DrainEventsAsync(handler.CreateAsync( + NewConversationTurn(ConversationId, "first question"), + NewServingContext("resp_" + new string('7', 46), []), + CancellationToken.None)); + + // Act: a second turn of the same conversation, for which the service now reports history. + await DrainEventsAsync(handler.CreateAsync( + NewConversationTurn(ConversationId, "second question"), + NewServingContext("resp_" + new string('8', 46), [NewHistoryMessageItem("msg_hist_1", "first question")]), + CancellationToken.None)); + + // Assert: a persisted session says a prior turn ran here, not that the conversation is inside it. + // Only a workflow keeps its messages that way; anything else starts each turn with nothing, so + // withholding the history would leave it answering blind. + Assert.NotNull(agent.CapturedMessages); + Assert.Contains(agent.CapturedMessages!, m => m.Text.Contains("first question", StringComparison.Ordinal)); + } + + [Fact] + public async Task CreateAsync_WhenTheModelReportsAConversationId_TurnsStillCompleteAsync() + { + // Arrange: a container whose model call reports a conversation id, which is what happens when + // the container's chat client lets the model keep the conversation. The agent records that id on + // the session, and from then on its own conflict policy would reject the provider the host + // registers, failing the turn. + var agent = new ChatClientAgent( + CreateCapturingChatClient([], conversationId: "conv-from-the-model"), + new ChatClientAgentOptions { Name = "hosted" }); + var handler = BuildHandlerWith(agent, new FakeHostedSessionIsolationKeyProvider(), new InMemoryAgentSessionStore()); + + // Act + var first = await CollectEventNamesAsync(handler, "conv-model", "resp_" + new string('9', 46), "first question"); + var second = await CollectEventNamesAsync(handler, "conv-model", "resp_" + new string('a', 46), "second question"); + + // Assert: both turns run to completion. The host owns history for this agent, so the agent's + // policy of refusing a second history manager must not be left to fire on the host's own + // registration. + Assert.Contains("ResponseCompletedEvent", first); + Assert.DoesNotContain("ResponseFailedEvent", first); + Assert.Contains("ResponseCompletedEvent", second); + Assert.DoesNotContain("ResponseFailedEvent", second); + } + + private static async Task> CollectEventNamesAsync( + AgentFrameworkResponseHandler handler, string conversationId, string responseId, string text) + { + var names = new List(); + await foreach (var evt in handler.CreateAsync( + NewConversationTurn(conversationId, text), NewServingContext(responseId, []), CancellationToken.None)) + { + names.Add(evt.GetType().Name); + } + + return names; + } + + private static CreateResponse NewConversationTurn(string conversationId, string text) + { + var request = new CreateResponse { Model = "test" }; + request.Conversation = BinaryData.FromString($"\"{conversationId}\""); + request.Input = BinaryData.FromObjectAsJson(new[] + { + new { type = "message", id = "msg_" + Guid.NewGuid().ToString("N")[..8], status = "completed", role = "user", + content = new[] { new { type = "input_text", text } } } + }); + return request; + } + + private static ResponseContext NewServingContext(string responseId, IReadOnlyList history) + { + var ctx = new Mock(responseId) { CallBase = true }; + ctx.Setup(x => x.PlatformContext).Returns(new PlatformContext("alice", null)); + ctx.Setup(x => x.GetHistoryAsync(It.IsAny())).ReturnsAsync(history); + ctx.Setup(x => x.GetInputItemsAsync(It.IsAny(), It.IsAny())).ReturnsAsync(Array.Empty()); + return ctx.Object; + } + + #endregion + + #region Chat history source routing + + // These tests pin down who supplies the conversation history to a hosted agent. Three of them are + // regression tests for the behaviour this region replaced: the handler used to fetch the platform + // history and prepend it to the input of every turn, while a ChatClientAgent independently ran its + // own ChatHistoryProvider. Against that older handler these three fail: + // - DoesNotCopyPlatformHistoryIntoTheSession (the service's turns ended up in the session) + // - DoesNotAskItToStorePlatformHistory (and in a custom provider's own database) + // - UsesThatProviderInsteadOfThePlatform (both sources reached the model at once) + + [Fact] + public async Task CreateAsync_AgentWithoutProviderPipeline_ReceivesPlatformHistoryInInputAsync() + { + // Arrange: a plain AIAgent (a hosted workflow, for example) has no ChatHistoryProvider + // pipeline, so the handler is the only thing that can hand it the platform history. + var agent = new CapturingAgent(); + var handler = BuildHandlerWith(agent, new FakeHostedSessionIsolationKeyProvider(), new InMemoryAgentSessionStore()); + var (request, ctx) = BuildChainRequest("resp_" + new string('1', 46), callId: null); + ctx.Setup(x => x.GetHistoryAsync(It.IsAny())) + .ReturnsAsync([NewHistoryMessageItem("msg_hist_1", "earlier turn")]); + + // Act + await DrainEventsAsync(handler.CreateAsync(request, ctx.Object, CancellationToken.None)); + + // Assert + Assert.NotNull(agent.CapturedMessages); + Assert.Contains(agent.CapturedMessages!, m => m.Text.Contains("earlier turn", StringComparison.Ordinal)); + } + + [Fact] + public async Task CreateAsync_ChatClientAgentWithoutHistoryProvider_SendsPlatformHistoryExactlyOnceAsync() + { + // Arrange: no chat history provider was supplied, so the platform stays the source and the + // handler registers FoundryChatHistoryProvider for the turn. + var captured = new List(); + var agent = new ChatClientAgent(CreateCapturingChatClient(captured)); + var handler = BuildHandlerWith(agent, new FakeHostedSessionIsolationKeyProvider(), new InMemoryAgentSessionStore()); + var (request, ctx) = BuildChainRequest("resp_" + new string('2', 46), callId: null); + ctx.Setup(x => x.GetHistoryAsync(It.IsAny())) + .ReturnsAsync([NewHistoryMessageItem("msg_hist_1", "earlier turn")]); + + // Act + await DrainEventsAsync(handler.CreateAsync(request, ctx.Object, CancellationToken.None)); + + // Assert: the earlier turn still reaches the model, and only one copy of it does. + Assert.Single(captured, m => m.Text.Contains("earlier turn", StringComparison.Ordinal)); + } + + [Fact] + public async Task CreateAsync_ChatClientAgentWithHistoryProvider_DoesNotAskItToStorePlatformHistoryAsync() + { + // Arrange: an agent whose own provider records everything it is asked to store, and a platform + // that already holds an earlier turn of this conversation. + var recordingProvider = new RecordingChatHistoryProvider(); + var agent = new ChatClientAgent( + CreateCapturingChatClient([]), + new ChatClientAgentOptions { ChatHistoryProvider = recordingProvider }); + var handler = BuildHandlerWith(agent, new FakeHostedSessionIsolationKeyProvider(), new InMemoryAgentSessionStore()); + var (request, ctx) = BuildChainRequest("resp_" + new string('5', 46), callId: null); + ctx.Setup(x => x.GetHistoryAsync(It.IsAny())) + .ReturnsAsync([NewHistoryMessageItem("msg_hist_1", "already kept by the service")]); + + // Act + await DrainEventsAsync(handler.CreateAsync(request, ctx.Object, CancellationToken.None)); + + // Assert: the agent's own store must not be told to write a turn the service already holds. The + // older handler passed that turn in as ordinary input, and since platform items carry no + // chat-history source marker the provider took it for newly written content and stored it, + // duplicating into the agent's own database a conversation the service was already keeping. + Assert.DoesNotContain(recordingProvider.Stored, m => m.Text.Contains("already kept by the service", StringComparison.Ordinal)); + } + + [Fact] + public async Task CreateAsync_ChatClientAgentWithoutHistoryProvider_DoesNotCopyPlatformHistoryIntoTheSessionAsync() + { + // Arrange: the model inside the container keeps the conversation, so it reports a conversation + // id of its own, and the platform reports one earlier turn for the same conversation. + const string ResponseId = "resp_" + "4444444444444444444444444444444444444444444444"; + var store = new InMemoryAgentSessionStore(); + var agent = new ChatClientAgent(CreateCapturingChatClient([], conversationId: "conv-model")); + var handler = BuildHandlerWith(agent, new FakeHostedSessionIsolationKeyProvider(), store); + var (request, ctx) = BuildChainRequest(ResponseId, callId: null); + ctx.Setup(x => x.GetHistoryAsync(It.IsAny())) + .ReturnsAsync([NewHistoryMessageItem("msg_hist_1", "already kept by the service")]); + + // Act + await DrainEventsAsync(handler.CreateAsync(request, ctx.Object, CancellationToken.None)); + + // Assert: the model and the service are both keeping this conversation, so the container keeps + // none of it. The older handler fed the service's history to the agent as ordinary input, and + // because platform items carry no chat-history source marker the agent's default in-memory + // provider stored it as if this turn had produced it, leaving a third copy on disk that then + // drifts from the other two. + Assert.DoesNotContain("already kept by the service", await SerializedSessionOfAsync(agent, store, ResponseId), StringComparison.Ordinal); + } + + [Fact] + public async Task CreateAsync_ChatClientAgentWithHistoryProvider_UsesThatProviderInsteadOfThePlatformAsync() + { + // Arrange: the agent was created with its own chat history provider. + var captured = new List(); + var agent = new ChatClientAgent( + CreateCapturingChatClient(captured), + new ChatClientAgentOptions { ChatHistoryProvider = new FixedChatHistoryProvider("from my own store") }); + var handler = BuildHandlerWith(agent, new FakeHostedSessionIsolationKeyProvider(), new InMemoryAgentSessionStore()); + var (request, ctx) = BuildChainRequest("resp_" + new string('3', 46), callId: null); + ctx.Setup(x => x.GetHistoryAsync(It.IsAny())) + .ReturnsAsync([NewHistoryMessageItem("msg_hist_1", "from the platform")]); + + // Act + await DrainEventsAsync(handler.CreateAsync(request, ctx.Object, CancellationToken.None)); + + // Assert: one source only, and hosted it is the one the AgentServer SDK's storage provider + // records and serves back. A provider storing a second copy inside the container would add a + // conversation that storage provider never sees, so the agent's provider is stood down for the + // turn rather than mixed in. + Assert.Contains(captured, m => m.Text.Contains("from the platform", StringComparison.Ordinal)); + Assert.DoesNotContain(captured, m => m.Text.Contains("from my own store", StringComparison.Ordinal)); + } + + [Fact] + public async Task CreateAsync_SessionIsGone_RecoversTheHistoryFromTheServiceAsync() + { + // Arrange: a turn lands on a container that has no session for the conversation, which is what a + // restart or a second replica looks like. + var captured = new List(); + var agent = new ChatClientAgent(CreateCapturingChatClient(captured)); + var handler = BuildHandlerWith(agent, new FakeHostedSessionIsolationKeyProvider(), new InMemoryAgentSessionStore()); + + // Act + await DrainEventsAsync(handler.CreateAsync( + NewConversationRequest("conv-cold", "second question", store: true), + NewContextServing("resp_" + new string('9', 46), [NewHistoryMessageItem("msg_hist_1", "first question")]), + CancellationToken.None)); + + // Assert: nothing inside the container remembers this conversation, and nothing needs to. The + // AgentServer SDK's storage provider holds it and hands it back, so the turn runs as if the + // container had served every one before it. + Assert.Single(captured, m => m.Text.Contains("first question", StringComparison.Ordinal)); + } + + [Fact] + public async Task CreateAsync_UnstoredRequestAndAgentWithARawRepresentationFactory_KeepsBothAsync() + { + // Arrange: an agent whose own ChatOptions carry a raw representation factory, the way a container + // adds settings the chat client only understands in its own request type. The caller asks for a + // turn the service must not store. + ChatOptions? sentToTheClient = null; + var client = new Mock(); + client.Setup(c => c.GetStreamingResponseAsync( + It.IsAny>(), It.IsAny(), It.IsAny())) + .Returns((IEnumerable _, ChatOptions? options, CancellationToken _) => + { + sentToTheClient = options; + return ToAsyncEnumerableUpdatesAsync(new ChatResponseUpdate(ChatRole.Assistant, "ok") { MessageId = "resp_msg_1" }); + }); + + var agent = new ChatClientAgent(client.Object, new ChatClientAgentOptions + { + ChatOptions = new ChatOptions + { + RawRepresentationFactory = _ => new CreateResponseOptions { EndUserId = "set by the container" }, + }, + }); + var handler = BuildHandlerWith(agent, new FakeHostedSessionIsolationKeyProvider(), new InMemoryAgentSessionStore()); + + // Act + await DrainEventsAsync(handler.CreateAsync( + NewConversationRequest("conv-raw", "a question", store: false), + NewContextServing("resp_" + new string('7', 46), []), + CancellationToken.None)); + + // Assert: the agent chains a request factory with its own by taking the agent's only when the + // request's returns null, so a request factory that always answers would silently drop whatever + // the container configured. Both settings have to survive on the way to the client. + Assert.NotNull(sentToTheClient?.RawRepresentationFactory); + var raw = Assert.IsType(sentToTheClient!.RawRepresentationFactory!(client.Object)); + Assert.False(raw.StoredOutputEnabled); + Assert.Equal("set by the container", raw.EndUserId); + } + + [Fact] + public async Task CreateAsync_AgentWhoseChatClientReportsAConversationId_IsRejectedAsync() + { + // Arrange: a chat client whose underlying service keeps the conversation and says so on every + // answer, whatever the host asks of it. + var client = new Mock(); + client.Setup(c => c.GetStreamingResponseAsync( + It.IsAny>(), It.IsAny(), It.IsAny())) + .Returns(() => ToAsyncEnumerableUpdatesAsync( + new ChatResponseUpdate(ChatRole.Assistant, "ok") { MessageId = "resp_msg_1", ConversationId = "conv-downstream" })); + + var agent = new ChatClientAgent(client.Object); + var handler = BuildHandlerWith(agent, new FakeHostedSessionIsolationKeyProvider(), new InMemoryAgentSessionStore()); + + await DrainEventsAsync(handler.CreateAsync( + NewConversationRequest("conv-rejected", "first question", store: true), + NewContextServing("resp_" + new string('3', 45) + "0", []), + CancellationToken.None)); + + // Act + Assert: a hosted agent's conversation is recorded by the AgentServer SDK's storage + // provider, so a second one held by the service behind the chat client has no owner and no way + // to stay in step. The next turn is refused as a plain bad request rather than run against a + // conversation nobody can reconcile. + var failure = await Assert.ThrowsAsync(() => DrainEventsAsync(handler.CreateAsync( + NewConversationRequest("conv-rejected", "second question", store: true), + NewContextServing("resp_" + new string('3', 45) + "1", []), + CancellationToken.None))); + + Assert.Equal("service_managed_chat_history_not_supported", failure.Error.Code); + Assert.Equal(400, failure.StatusCode); + } + + [Fact] + public async Task CreateAsync_ChatClientAgent_TakesTheWholeConversationFromTheHostingServiceAsync() + { + // Arrange: the AgentServer SDK's storage provider holds the conversation, which is the only + // place it lives. + var captured = new List(); + var agent = new ChatClientAgent(CreateCapturingChatClient(captured)); + var handler = BuildHandlerWith(agent, new FakeHostedSessionIsolationKeyProvider(), new InMemoryAgentSessionStore()); + + await DrainEventsAsync(handler.CreateAsync( + NewConversationRequest("conv-single-source", "first question", store: true), + NewContextServing("resp_" + new string('4', 45) + "0", []), + CancellationToken.None)); + captured.Clear(); + + // Act: a second turn, with that storage provider serving the first one back. + await DrainEventsAsync(handler.CreateAsync( + NewConversationRequest("conv-single-source", "second question", store: true), + NewContextServing("resp_" + new string('4', 45) + "1", [NewHistoryMessageItem("msg_hist_1", "first question")]), + CancellationToken.None)); + + // Assert: what it holds plus this turn's input, each exactly once. + Assert.Single(captured, m => m.Text.Contains("first question", StringComparison.Ordinal)); + Assert.Single(captured, m => m.Text.Contains("second question", StringComparison.Ordinal)); + } + + [Fact] + public async Task CreateAsync_StoredRequest_StillAsksTheChatClientNotToStoreAsync() + { + // Arrange: the caller asks for the turn to be stored. + ChatOptions? sentToTheClient = null; + var client = new Mock(); + client.Setup(c => c.GetStreamingResponseAsync( + It.IsAny>(), It.IsAny(), It.IsAny())) + .Returns((IEnumerable _, ChatOptions? options, CancellationToken _) => + { + sentToTheClient = options; + return ToAsyncEnumerableUpdatesAsync(new ChatResponseUpdate(ChatRole.Assistant, "ok") { MessageId = "resp_msg_1" }); + }); + + var agent = new ChatClientAgent(client.Object); + var handler = BuildHandlerWith(agent, new FakeHostedSessionIsolationKeyProvider(), new InMemoryAgentSessionStore()); + + // Act + await DrainEventsAsync(handler.CreateAsync( + NewConversationRequest("conv-never-downstream", "a question", store: true), + NewContextServing("resp_" + new string('5', 45) + "0", []), + CancellationToken.None)); + + // Assert: storing is the AgentServer SDK's job, done by its storage provider around this + // handler. Letting the service behind the chat client store as well writes the same + // conversation twice, in two places that then drift apart. + Assert.NotNull(sentToTheClient?.RawRepresentationFactory); + var raw = Assert.IsType(sentToTheClient!.RawRepresentationFactory!(client.Object)); + Assert.False(raw.StoredOutputEnabled); + } + + [Fact] + public async Task CreateAsync_AgentSpeakingChatCompletions_AlsoAsksItNotToStoreAsync() + { + // Arrange: a container whose chat client speaks Chat Completions rather than Responses, so the + // request it understands is a ChatCompletionOptions. + ChatOptions? sentToTheClient = null; + var client = new Mock(); + client.Setup(c => c.GetStreamingResponseAsync( + It.IsAny>(), It.IsAny(), It.IsAny())) + .Returns((IEnumerable _, ChatOptions? options, CancellationToken _) => + { + sentToTheClient = options; + return ToAsyncEnumerableUpdatesAsync(new ChatResponseUpdate(ChatRole.Assistant, "ok") { MessageId = "resp_msg_1" }); + }); + + var agent = new ChatClientAgent(client.Object, new ChatClientAgentOptions + { + ChatOptions = new ChatOptions + { + RawRepresentationFactory = _ => new ChatCompletionOptions { EndUserId = "set by the container" }, + }, + }); + var handler = BuildHandlerWith(agent, new FakeHostedSessionIsolationKeyProvider(), new InMemoryAgentSessionStore()); + + // Act + await DrainEventsAsync(handler.CreateAsync( + NewConversationRequest("conv-completions", "a question", store: true), + NewContextServing("resp_" + new string('6', 45) + "0", []), + CancellationToken.None)); + + // Assert: the setting has the same name on both OpenAI request shapes, so a chat client speaking + // either protocol is covered, and what the container configured survives alongside it. + Assert.NotNull(sentToTheClient?.RawRepresentationFactory); + var raw = Assert.IsType(sentToTheClient!.RawRepresentationFactory!(client.Object)); + Assert.False(raw.StoredOutputEnabled); + Assert.Equal("set by the container", raw.EndUserId); + } + + private static CreateResponse NewConversationRequest(string conversationId, string text, bool store) + { + var request = new CreateResponse { Model = "test", Store = store }; + request.Conversation = BinaryData.FromString($"\"{conversationId}\""); + request.Input = BinaryData.FromObjectAsJson(new[] + { + new { type = "message", id = "msg_" + Guid.NewGuid().ToString("N")[..8], status = "completed", role = "user", + content = new[] { new { type = "input_text", text } } } + }); + return request; + } + + private static ResponseContext NewContextServing(string responseId, IReadOnlyList history) + { + var ctx = new Mock(responseId) { CallBase = true }; + ctx.Setup(x => x.PlatformContext).Returns(new PlatformContext("alice", null)); + ctx.Setup(x => x.GetHistoryAsync(It.IsAny())).ReturnsAsync(history); + ctx.Setup(x => x.GetInputItemsAsync(It.IsAny(), It.IsAny())).ReturnsAsync(Array.Empty()); + return ctx.Object; + } + + /// Reads back the session the handler persisted for a response and returns it as JSON text. + private static async Task SerializedSessionOfAsync(AIAgent agent, InMemoryAgentSessionStore store, string responseId) + { + var sessionKey = HostedConversationKey.Resolve(conversationId: null, previousResponseId: null, responseId); + var session = await store.GetSessionAsync(agent, sessionKey!, FakeHostedSessionIsolationKeyProvider.DefaultUserId, CancellationToken.None); + + // The handler persists the session at the end of every turn, so a missing one means the turn did + // not get that far and the assertions below would otherwise pass without proving anything. + Assert.NotNull(session); + + var serialized = await agent.SerializeSessionAsync(session, cancellationToken: CancellationToken.None); + return serialized.GetRawText(); + } + + private static OutputItemMessage NewHistoryMessageItem(string id, string text) => + new( + id: id, + role: MessageRole.Assistant, + content: [new MessageContentOutputTextContent(text, Array.Empty(), Array.Empty())], + status: MessageStatus.Completed); + + private static IChatClient CreateCapturingChatClient(List captured, string? conversationId = null) + { + var mock = new Mock(); + mock.Setup(c => c.GetStreamingResponseAsync( + It.IsAny>(), + It.IsAny(), + It.IsAny())) + .Returns((IEnumerable messages, ChatOptions? options, CancellationToken _) => + { + captured.AddRange(messages); + + // Mirror the MEAI OpenAI adapter, which reports no conversation id for a response the + // service was not asked to store: OpenAIResponsesChatClient sets ChatResponse.ConversationId + // to null whenever CreateResponseOptions.StoredOutputEnabled is false. Without that rule + // here a fake would keep handing back a stored thread the caller opted out of. + var storedOutputDisabled = + options?.RawRepresentationFactory?.Invoke(mock.Object) is CreateResponseOptions { StoredOutputEnabled: false }; + + return ToAsyncEnumerableUpdatesAsync( + new ChatResponseUpdate(ChatRole.Assistant, "ok") + { + MessageId = "resp_msg_1", + ConversationId = storedOutputDisabled ? null : conversationId, + }); + }); + return mock.Object; + } + + private static async IAsyncEnumerable ToAsyncEnumerableUpdatesAsync(params ChatResponseUpdate[] updates) + { + foreach (var update in updates) + { + yield return update; + } + + await Task.CompletedTask; + } + + /// A chat history provider that always returns the same message, standing in for one backed by a store. + private sealed class FixedChatHistoryProvider(string text) : ChatHistoryProvider + { + protected override ValueTask> ProvideChatHistoryAsync(InvokingContext context, CancellationToken cancellationToken = default) + => new([new ChatMessage(ChatRole.User, text)]); + + protected override ValueTask StoreChatHistoryAsync(InvokedContext context, CancellationToken cancellationToken = default) => default; + } + + /// A chat history provider that records everything it is asked to write, standing in for one backed by a database. + private sealed class RecordingChatHistoryProvider : ChatHistoryProvider + { + public List Stored { get; } = []; + + protected override ValueTask> ProvideChatHistoryAsync(InvokingContext context, CancellationToken cancellationToken = default) + => new([]); + + protected override ValueTask StoreChatHistoryAsync(InvokedContext context, CancellationToken cancellationToken = default) + { + this.Stored.AddRange(context.RequestMessages); + if (context.ResponseMessages is not null) + { + this.Stored.AddRange(context.ResponseMessages); + } + + return default; + } + } + + #endregion + private static TestAgent CreateTestAgent(string responseText) { return new TestAgent(responseText); @@ -840,6 +1410,57 @@ protected override ValueTask DeserializeSessionCoreAsync( new(new SimpleAgentSession()); } + /// + /// Stands in for a hosted workflow: an whose session type is named the way the + /// real one is, which is how the handler recognises a session that already carries the conversation. + /// + private sealed class WorkflowLikeAgent : AIAgent + { + public IEnumerable? CapturedMessages { get; private set; } + + protected override IAsyncEnumerable RunCoreStreamingAsync( + IEnumerable messages, + AgentSession? session, + AgentRunOptions? options, + CancellationToken cancellationToken = default) + { + this.CapturedMessages = messages.ToList(); + return ToAsyncEnumerableAsync(new AgentResponseUpdate + { + MessageId = "resp_msg_1", + Contents = [new MeaiTextContent("captured")] + }); + } + + protected override Task RunCoreAsync( + IEnumerable messages, + AgentSession? session, + AgentRunOptions? options, + CancellationToken cancellationToken = default) => + throw new NotImplementedException(); + + protected override ValueTask CreateSessionCoreAsync( + CancellationToken cancellationToken = default) => + new(new WorkflowSession()); + + protected override ValueTask SerializeSessionCoreAsync( + AgentSession session, + JsonSerializerOptions? jsonSerializerOptions, + CancellationToken cancellationToken = default) => + new(JsonDocument.Parse("{}").RootElement); + + protected override ValueTask DeserializeSessionCoreAsync( + JsonElement serializedState, + JsonSerializerOptions? jsonSerializerOptions, + CancellationToken cancellationToken = default) => + new(new WorkflowSession()); + } + + /// Carries the name the handler looks for; the real one is internal to its own package. + private sealed class WorkflowSession : AgentSession + { + } + private sealed class CancellationCheckingAgent : AIAgent { protected override async IAsyncEnumerable RunCoreStreamingAsync( diff --git a/dotnet/tests/Microsoft.Agents.AI.Foundry.Hosting.UnitTests/FileSystemAgentSessionStoreTests.cs b/dotnet/tests/Microsoft.Agents.AI.Foundry.Hosting.UnitTests/FileSystemAgentSessionStoreTests.cs index de43efbc5a3..01552edcd15 100644 --- a/dotnet/tests/Microsoft.Agents.AI.Foundry.Hosting.UnitTests/FileSystemAgentSessionStoreTests.cs +++ b/dotnet/tests/Microsoft.Agents.AI.Foundry.Hosting.UnitTests/FileSystemAgentSessionStoreTests.cs @@ -50,20 +50,33 @@ public void Constructor_NullOrWhitespaceRoot_Throws() } [Fact] - public async Task GetSessionAsync_NoFileOnDisk_ReturnsFreshSessionFromAgentAsync() + public async Task GetSessionAsync_NoFileOnDisk_ReturnsNullAsync() { var store = new FileSystemAgentSessionStore(this._root); var agent = new TestAgent(); var session = await store.GetSessionAsync(agent, "conv-1", userId: null); + Assert.Null(session); + Assert.Equal(0, agent.CreateCalls); + Assert.Equal(0, agent.DeserializeCalls); + } + + [Fact] + public async Task GetOrCreateSessionAsync_NoFileOnDisk_ReturnsFreshSessionFromAgentAsync() + { + var store = new FileSystemAgentSessionStore(this._root); + var agent = new TestAgent(); + + var session = await store.GetOrCreateSessionAsync(agent, "conv-1", userId: null); + Assert.NotNull(session); Assert.Equal(1, agent.CreateCalls); Assert.Equal(0, agent.DeserializeCalls); } [Fact] - public async Task GetSessionAsync_EmptyFileOnDisk_ReturnsFreshSessionAsync() + public async Task GetSessionAsync_EmptyFileOnDisk_ReturnsNullAsync() { var store = new FileSystemAgentSessionStore(this._root); Directory.CreateDirectory(store.RootDirectory); @@ -72,8 +85,8 @@ public async Task GetSessionAsync_EmptyFileOnDisk_ReturnsFreshSessionAsync() var agent = new TestAgent(); var session = await store.GetSessionAsync(agent, "conv-empty", userId: null); - Assert.NotNull(session); - Assert.Equal(1, agent.CreateCalls); + Assert.Null(session); + Assert.Equal(0, agent.CreateCalls); Assert.Equal(0, agent.DeserializeCalls); } @@ -245,7 +258,7 @@ public async Task GetSessionAsync_NoExistingFile_DoesNotCreateAgentDirectoryAsyn var session = await store.GetSessionAsync(agent, "missing-id", userId: null); - Assert.NotNull(session); + Assert.Null(session); Assert.False(Directory.Exists(this._root), "Read miss must not create the root directory."); } @@ -385,11 +398,11 @@ public async Task GetSessionAsync_DifferentUser_DoesNotReadAnotherUsersSessionAs await store.SaveSessionAsync(agent, "shared-conv", NewSession(), userId: "alice"); // Bob requests the same conversationId. The per-user partition means Bob's path is distinct, - // so the store returns a fresh session (no leak), not Alice's persisted state. + // so the store returns null (no leak), not Alice's persisted state. var bobSession = await store.GetSessionAsync(agent, "shared-conv", userId: "bob"); - Assert.NotNull(bobSession); - Assert.Equal(1, agent.CreateCalls); // fresh session created for Bob + Assert.Null(bobSession); // no session for Bob under his partition + Assert.Equal(0, agent.CreateCalls); // a plain lookup never creates Assert.Equal(0, agent.DeserializeCalls); // Alice's file never deserialized for Bob }