.NET: [Feature]: [.NET] Add ValkeyStreamBuffer for resumable agent streaming via Valkey Streams

Description

Description:

The Microsoft.Agents.AI.Valkey package currently provides ValkeyChatHistoryProvider (persistent conversation history via Valkey Lists) and ValkeyContextProvider (memory/context enrichment via Valkey full-text search). The remaining component from the Valkey integration story is a ValkeyStreamBuffer — a resumable streaming buffer backed by Valkey Streams.

Problem: When a client disconnects during a streamed agent response (RunStreamingAsync), partial output is lost. The framework's AgentResponseUpdate.ContinuationToken and AgentRunOptions.ContinuationToken provide the abstraction for stream resumption, and A2A implements it for its protocol, but there is no general-purpose, storage-backed implementation that any ChatClientAgent can use.

Proposed solution: Add a ValkeyStreamBuffer component to the Microsoft.Agents.AI.Valkey package that:

• Writes each AgentResponseUpdate chunk to a Valkey Stream via XADD, keyed by session/response ID
• Stamps each update with a ContinuationToken containing the Valkey Stream entry ID
• On reconnection, reads missed entries via XREAD starting from the client's last-seen entry ID
• Supports consumer groups (XREADGROUP) for multi-consumer scenarios
• Supports configurable TTL via XTRIM or MAXLEN to bound stream growth
• Works with both standalone Valkey and Valkey Cluster via StackExchange.Redis
• The implementation should follow the same patterns as the existing Valkey providers: ProviderSessionState for session-scoped state, IConnectionMultiplexer constructor overload for DI, IAsyncDisposable, CancellationToken forwarding, and ILogger integration.

Server requirements: Any Valkey (or Redis OSS) server — Valkey Streams use core commands only (XADD, XREAD, XREADGROUP, XACK, XTRIM), no modules required.

Reference: The A2A agent (
A2AAgent.cs
) demonstrates the continuation token pattern. AgentResponseUpdate.ContinuationToken and AgentRunOptions.ContinuationToken are the framework abstractions to integrate with.

Code Sample

using Microsoft.Agents.AI;
using Microsoft.Agents.AI.Valkey;
using Microsoft.Extensions.AI;

// Create the stream buffer backed by Valkey
await using var streamBuffer = new ValkeyStreamBuffer(
    "localhost:6379",
    stateInitializer: session => new ValkeyStreamBuffer.State(
        responseId: $"response-{Guid.NewGuid():N}"),
    keyPrefix: "agent_stream",
    maxLength: 1000);  // XTRIM MAXLEN

// Wire it into an agent via middleware or options
AIAgent agent = chatClient.AsAIAgent(new ChatClientAgentOptions
{
    StreamBuffer = streamBuffer
});

// --- Producer side: stream a response ---
AgentSession session = await agent.CreateSessionAsync();
string? lastToken = null;

await foreach (var update in agent.RunStreamingAsync("Tell me about Valkey", session))
{
    Console.Write(update.Text);
    lastToken = update.ContinuationToken?.ToString();
}

// --- Consumer side: resume after disconnect ---
// Client reconnects with the last continuation token it received
var resumeOptions = new AgentRunOptions
{
    ContinuationToken = ResponseContinuationToken.FromBytes(
        System.Text.Encoding.UTF8.GetBytes(lastToken!))
};

await foreach (var update in agent.RunStreamingAsync(session, resumeOptions))
{
    // Receives only the updates after the last-seen entry
    Console.Write(update.Text);
}

Language/SDK

.NET