diff --git a/clients/dotnet/AgentHostProtocol.slnx b/clients/dotnet/AgentHostProtocol.slnx
index c9ea641c3..ab398c077 100644
--- a/clients/dotnet/AgentHostProtocol.slnx
+++ b/clients/dotnet/AgentHostProtocol.slnx
@@ -2,6 +2,8 @@
+
+
diff --git a/clients/dotnet/README.md b/clients/dotnet/README.md
index b5d3839b8..9b8db87c1 100644
--- a/clients/dotnet/README.md
+++ b/clients/dotnet/README.md
@@ -51,7 +51,8 @@ Reducers.ApplyToSession(state, action); // mutates `state` in place
```
See [`examples/`](examples/) for runnable `ConnectWs` and `ReducersDemo`
-console apps.
+console apps, and the windowing examples described under
+[Windowed delivery](#windowed-delivery-opt-in).
## Dependency injection
@@ -87,6 +88,153 @@ state, each returned resource is delivered as a `SubscriptionEventSnapshot`
before the host reports `Connected`. Consumers should replace that resource's
local reducer state with the snapshot before processing later actions.
+## Windowed delivery (opt-in)
+
+Windowing is off by default. It is offered per subscription, without changing
+the protocol versions offered by initialization. A host that omits flow control
+declines the offer; the returned handle reports `FlowControl.Mode == "ordinary"`
+and `FallbackReason == "declined"`. Its snapshot is still the first event
+(when an ordinary empty-state result omits its snapshot, there is no snapshot
+event). Ordinary fallback retains the existing drop-oldest buffering behavior.
+TCP subscriptions require acceptance and cannot fall back.
+
+```csharp
+var config = new ClientConfig
+{
+ Windowing = new WindowingConfig
+ {
+ WindowBytes = 256 * 1024,
+ MaximumFrameBytes = 16 * 1024,
+ MaxBufferedSendBytes = 1024 * 1024, // per subscription
+ },
+};
+await using var client = AhpClient.Connect(transport, config);
+await client.InitializeAsync("my-client");
+var (result, terminal) = await client.SubscribeWindowedAsync("ahp-terminal:/t1");
+await foreach (var evt in terminal.ReadEventsAsync(cancellationToken))
+{
+ await RenderOrApplyAsync(evt, cancellationToken);
+} // advancing (or disposing) the iterator releases the preceding event
+```
+
+The additive `IWindowedAhpClient` capability and `IAhpClient` extension methods
+also work with clients returned by the DI factory. Existing `IAhpClient`,
+`ITransport`, and serializer implementations do not need new interface members.
+Per-host defaults use `HostConfig.ClientConfig.Windowing`; the multi-host
+runtime offers the same `SubscribeWindowedAsync(host, uri, options)` API.
+
+For manual release, choose `ReceiveMode.Manual` and process delivery leases:
+
+```csharp
+var (_, tcp) = await client.SubscribeWindowedAsync(tcpUri,
+ new WindowOptions { Receive = ReceiveMode.Manual, Coalescing = false });
+await foreach (var delivery in tcp.Deliveries.ReadAllAsync(cancellationToken))
+{
+ using (delivery)
+ {
+ await WriteToBoundedSinkAsync(delivery.Event, cancellationToken);
+ } // Dispose() or Release() returns credit exactly once
+}
+```
+
+There is one credit-owning handle per URI. Re-subscribing updates that same
+handle. `AttachSubscription`, global event streams, and host observer streams
+remain lossy observers and **never** release credit. The owning handle retains
+every windowed event; it does not silently discard events when a consumer is slow.
+New event variants cover snapshots, TCP data/EOF/reset, and unknown notifications.
+
+Send-side backpressure is exposed independently:
+
+```csharp
+await tcp.WaitWritableAsync(cancellationToken); // advisory, not a reservation
+await tcp.NotifyAsync("tcp/data",
+ new TcpDataParams { Channel = tcpUri, Data = Convert.ToBase64String(bytes) },
+ cancellationToken); // completes after transport transmission
+var status = tcp.FlowControl;
+tcp.FlowControlChanged += (_, current) => UpdateStatus(current);
+await foreach (var current in tcp.FlowControlChanges().ReadAllAsync(cancellationToken))
+{
+ UpdateStatus(current);
+}
+```
+
+Existing `NotifyAsync` and `DispatchAsync` use the same fair scheduler for an
+enabled URI. An oversized single send is accepted when its subscription's queue
+is empty; fragmentation and the negotiated window pace it. `MaxBufferedSendBytes`
+bounds additional queued data, not the size of that first logical message.
+Cancellation removes unsent messages, but an already-started fragmented message
+finishes to avoid corrupting the channel.
+
+Receive coalescing is enabled by default, only for adjacent unconsumed terminal
+output or chat deltas targeting the same part with matching metadata. It does
+not introduce latency or delivery preferences. TCP sends can combine decoded
+bytes; numbered dispatches, client-origin echoes, snapshots and EOF/reset never
+combine. Register custom rules per connection:
+
+```csharp
+config.Windowing.Coalescers.Add(new ProgressCoalescer()); // before Connect()
+
+sealed class ProgressCoalescer : IChannelMessageCoalescer
+{
+ public JsonRpcNotification? Coalesce(JsonRpcNotification earlier,
+ JsonRpcNotification later, CoalescingContext context)
+ {
+ // Example only when this application's progress events are replaceable.
+ return context.Direction == CoalescingDirection.Incoming &&
+ earlier.Method == "root/progress" &&
+ earlier.Params?.GetProperty("progressToken").GetString() ==
+ later.Params?.GetProperty("progressToken").GetString()
+ ? later : null;
+ }
+}
+```
+
+A rule must preserve channel/method and channel-required event semantics.
+Custom coalescers must be fast and synchronous; the library validates their
+output and faults invalid results. Merged receive deliveries preserve original
+wire charges, not the serialization size of the merged event.
+
+All protocol sizes are **UTF-16 code units × 2 of the exact JSON text**, not
+UTF-8 bytes. `WebSocketTransportOptions.MaxMessageBytes` defaults to 32 MiB of
+complete inbound wire-message bytes; its growing 64 KiB receive buffer limits
+only a read chunk. The transport-derived AHP frame default is
+`2 * (MaxMessageBytes / 3)` = **22,369,620 UTF-16 bytes**. Unknown or unbounded
+transport limits use **1,048,576**. Custom transports may implement the optional
+`ITransportMessageLimits` capability; raw text/binary payloads are already
+retained by `TransportMessage`.
+
+Managed windowed subscriptions retain their owner and unread deliveries across
+reconnect, reset byte counters, and resume with snapshot/replay/live recovery.
+Releasing old-connection deliveries never credits the new connection. Missing
+subscriptions fail visibly. TCP channels close on connection loss and are never
+automatically recreated. Low-level users open a new client and call
+`ResumeAsync(ReconnectParams)` with the generated `Windows.Items` wrapper;
+`GetWindowedSubscription(uri)` exposes its new owners.
+
+The receive window does not cap complete reassembled message size or total
+memory. Incomplete fragments are credited as they are stored so messages larger
+than a window can finish; consumer release holds the final fragment's credit.
+Applications must still avoid retaining deliveries indefinitely.
+
+### Examples
+
+Two runnable examples use an in-memory demo host that follows the window rules,
+so they need no server:
+
+- [`examples/WindowingBasic`](examples/WindowingBasic/Program.cs) opts a
+ terminal subscription into windowed delivery and shows the host pausing when
+ the window is full and resuming as the client returns credit.
+- [`examples/WindowingBackpressure`](examples/WindowingBackpressure/Program.cs)
+ adds a slow renderer, a custom coalescer that keeps only the latest terminal
+ resize, and a paste that waits for `WaitWritableAsync()` instead of queueing.
+
+```sh
+dotnet run --project examples/WindowingBasic
+dotnet run --project examples/WindowingBackpressure
+```
+
+The demo host and in-memory transport are in [`examples/Shared`](examples/Shared).
+
## Observability
The client emits OpenTelemetry-native traces and metrics via `System.Diagnostics`
diff --git a/clients/dotnet/TELEMETRY.md b/clients/dotnet/TELEMETRY.md
index ad392ec2a..300b0500c 100644
--- a/clients/dotnet/TELEMETRY.md
+++ b/clients/dotnet/TELEMETRY.md
@@ -51,7 +51,37 @@ semantic conventions.
| `ahp.client.reconnects` | Counter | `{reconnect}` | `ahp.outcome` |
| `ahp.client.events.dropped` | Counter | `{event}` | `ahp.stream` (`subscription` \| `event` \| `state` \| `host-*`) |
| `ahp.client.frames.malformed` | Counter | `{frame}` | — |
+| `ahp.client.window.data.sent` | Counter | `By` | — |
+| `ahp.client.window.data.received` | Counter | `By` | — |
+| `ahp.client.window.credit.sent` | Counter | `By` | — |
+| `ahp.client.window.credit.received` | Counter | `By` | — |
+| `ahp.client.window.blocked.duration` | Histogram | `ms` | — |
+| `ahp.client.window.buffered` | ObservableGauge | `By` | `direction` (`send` \| `receive`) |
Metric names are lowercase-dotted per OpenTelemetry convention (the C# instrument
fields are PascalCase). A consumer's own `ILogger` written inside one of the
client's spans auto-correlates to that trace, so no logs are originated here.
+
+Window data counters measure the actual JSON message/fragment envelopes in
+UTF-16 bytes. Credit counters measure the credit notifications' UTF-16 wire
+size, not their cumulative `consumedBytes` value. The buffered gauge sums unsent
+data and retained receive deliveries across live owning handles, including
+completed unread deliveries retained across managed reconnect. Merged receive
+items retain the sum of their original charges. The duration histogram records
+blocked-to-writable (or closed) intervals, rather than polling samples.
+Closed owners are explicitly removed from the process-wide gauge registry;
+cleanup does not depend on a metrics listener collecting the gauge.
+
+Each flow-control transition adds an `ahp.flowcontrol.{kind}` Activity event
+with `ahp.flowcontrol.bytes`, without payload content. When no ambient Activity
+exists, an `ahp.flowcontrol` span hosts the event; no span is created when
+tracing is disabled.
+
+`WindowedSubscription.FlowControl`, the `FlowControlChanged` event, and
+`FlowControlChanges()` expose current limits, fallback reason, buffered and
+outstanding amounts, and writability. `client.FlowControlEvents()` is a lossy,
+observe-only connection stream carrying channel, kind and status. Kinds are
+`negotiated`, `declined`, `blocked`, `unblocked`, `creditSent`, `creditReceived`,
+`fragmentedMessageReceived`, `fragmentedMessageSent`, `coalesced`, `resumed`,
+and `dropped`. `MultiHostClient.FlowControlEvents()` adds host attribution.
+Reading these streams does not consume deliveries or return credit.
diff --git a/clients/dotnet/examples/Shared/DemoSupport.cs b/clients/dotnet/examples/Shared/DemoSupport.cs
new file mode 100644
index 000000000..3b94330c6
--- /dev/null
+++ b/clients/dotnet/examples/Shared/DemoSupport.cs
@@ -0,0 +1,37 @@
+// Helpers shared by the windowing examples.
+#nullable enable
+
+using System.Collections.Generic;
+using System.Runtime.CompilerServices;
+using System.Text.Encodings.Web;
+using System.Threading;
+using System.Threading.Tasks;
+using Microsoft.AgentHostProtocol;
+
+namespace WindowingExamples;
+
+internal static class DemoSupport
+{
+ private static readonly JsonSerializerOptions QuoteOptions =
+ new() { Encoder = JavaScriptEncoder.UnsafeRelaxedJsonEscaping };
+
+ /// Quotes text the way JavaScript's JSON.stringify does.
+ public static string Quote(string text) => JsonSerializer.Serialize(text, QuoteOptions);
+
+ ///
+ /// Takes deliveries in order and releases each one when the next one is
+ /// taken, or when the loop ends. This is the automatic release that
+ /// performs, but it also
+ /// exposes each delivery's charged size (),
+ /// which the examples print. Subscribe with .
+ ///
+ public static async IAsyncEnumerable TakeDeliveriesAsync(
+ this WindowedSubscription subscription,
+ [EnumeratorCancellation] CancellationToken cancellationToken = default)
+ {
+ await foreach (var delivery in subscription.Deliveries.ReadAllAsync(cancellationToken).ConfigureAwait(false))
+ {
+ using (delivery) yield return delivery;
+ }
+ }
+}
diff --git a/clients/dotnet/examples/Shared/InMemoryTransport.cs b/clients/dotnet/examples/Shared/InMemoryTransport.cs
new file mode 100644
index 000000000..6f9f74330
--- /dev/null
+++ b/clients/dotnet/examples/Shared/InMemoryTransport.cs
@@ -0,0 +1,62 @@
+// An in-memory ITransport pair for the windowing examples.
+//
+// Every message is a text frame built with TransportMessage.FromText, so the
+// client sees the exact JSON text the host wrote and can charge each windowed
+// message its UTF-16 size. Closing either half closes both.
+#nullable enable
+
+using System.Threading;
+using System.Threading.Channels;
+using System.Threading.Tasks;
+using Microsoft.AgentHostProtocol;
+
+namespace WindowingExamples;
+
+internal sealed class InMemoryTransport : ITransport
+{
+ private readonly Channel _inbox;
+ private readonly Channel _outbox;
+
+ private InMemoryTransport(Channel inbox, Channel outbox)
+ {
+ _inbox = inbox;
+ _outbox = outbox;
+ }
+
+ /// Creates a linked pair: what one half sends, the other half receives.
+ public static (InMemoryTransport Client, InMemoryTransport Host) CreatePair()
+ {
+ var toHost = Channel.CreateUnbounded(new UnboundedChannelOptions { SingleReader = true });
+ var toClient = Channel.CreateUnbounded(new UnboundedChannelOptions { SingleReader = true });
+ return (new InMemoryTransport(toClient, toHost), new InMemoryTransport(toHost, toClient));
+ }
+
+ public ValueTask SendAsync(TransportMessage message, CancellationToken cancellationToken = default)
+ {
+ cancellationToken.ThrowIfCancellationRequested();
+ return _outbox.Writer.TryWrite(message)
+ ? default
+ : throw new AhpTransportException("closed");
+ }
+
+ public async ValueTask ReceiveAsync(CancellationToken cancellationToken = default)
+ {
+ try
+ {
+ return await _inbox.Reader.ReadAsync(cancellationToken).ConfigureAwait(false);
+ }
+ catch (ChannelClosedException error)
+ {
+ throw new AhpTransportException("closed", inner: error);
+ }
+ }
+
+ public ValueTask CloseAsync(CancellationToken cancellationToken = default)
+ {
+ _outbox.Writer.TryComplete();
+ _inbox.Writer.TryComplete();
+ return default;
+ }
+
+ public ValueTask DisposeAsync() => CloseAsync();
+}
diff --git a/clients/dotnet/examples/Shared/WindowingDemoHost.cs b/clients/dotnet/examples/Shared/WindowingDemoHost.cs
new file mode 100644
index 000000000..3edb5225f
--- /dev/null
+++ b/clients/dotnet/examples/Shared/WindowingDemoHost.cs
@@ -0,0 +1,324 @@
+// A small in-memory AHP host for the windowing examples.
+//
+// It implements only what the examples need: `initialize`, a windowed
+// `subscribe` to one terminal, and the host side of the window rules:
+//
+// - It counts every data message it sends, in UTF-16 bytes, and stops while
+// `sent - consumed` is at or above the client's window.
+// - It counts every data message the client sends, and returns
+// `channel/credit` after it has processed each one.
+//
+// A real host applies the same rules to every windowed subscription.
+#nullable enable
+
+using System.Text;
+using System.Text.Encodings.Web;
+using System.Threading;
+using System.Threading.Channels;
+using System.Threading.Tasks;
+using Microsoft.AgentHostProtocol;
+
+namespace WindowingExamples;
+
+internal sealed class DemoHostOptions
+{
+ /// The largest window the host accepts for data that it sends.
+ public required long ClientWindowBytes { get; init; }
+
+ /// The window the host offers for data that the client sends.
+ public required long HostWindowBytes { get; init; }
+
+ /// How long the host takes to process one message from the client.
+ public TimeSpan InputDelay { get; init; }
+}
+
+internal sealed class WindowingDemoHost : IAsyncDisposable
+{
+ private const long MaximumFrameBytes = 16_384;
+
+ // Escape JSON the way JavaScript's JSON.stringify does, so the sizes match
+ // the TypeScript version of this example.
+ private static readonly SystemTextJsonAhpSerializer Json = new(
+ new JsonSerializerOptions { Encoder = JavaScriptEncoder.UnsafeRelaxedJsonEscaping });
+
+ private readonly DemoHostOptions _options;
+ private readonly InMemoryTransport _transport;
+ private readonly TaskCompletionSource _subscribed = new(TaskCreationOptions.RunContinuationsAsynchronously);
+ private readonly SemaphoreSlim _dataLock = new(1, 1);
+ private readonly object _gate = new();
+ private readonly Channel<(long Bytes, StateAction Action)> _input =
+ Channel.CreateUnbounded<(long Bytes, StateAction Action)>(new UnboundedChannelOptions { SingleReader = true });
+ private readonly StringBuilder _receivedInput = new();
+ private readonly Task _run;
+ private readonly Task _processInput;
+ private ChannelFlowControl? _limits;
+ private string _channel = "";
+ private long _serverSeq;
+
+ // Data that the host sends to the client.
+ private long _sentBytes;
+ private long _consumedBytes;
+ private TaskCompletionSource _credit = new(TaskCreationOptions.RunContinuationsAsynchronously);
+
+ // Data that the client sends to the host.
+ private long _receivedBytes;
+ private long _processedBytes;
+
+ public WindowingDemoHost(DemoHostOptions options)
+ {
+ _options = options;
+ (var client, _transport) = InMemoryTransport.CreatePair();
+ ClientTransport = client;
+ _run = RunAsync();
+ _processInput = ProcessInputAsync();
+ }
+
+ /// The half of the transport to give to .
+ public ITransport ClientTransport { get; }
+
+ /// Terminal input that the host has processed, in order.
+ public string ReceivedInput
+ {
+ get { lock (_gate) return _receivedInput.ToString(); }
+ }
+
+ /// Completes after the host has accepted the subscription and sent the snapshot.
+ public Task WhenSubscribedAsync() => _subscribed.Task;
+
+ /// Sends one terminal action, first waiting for credit if the client's window is full.
+ public async Task SendActionAsync(StateAction action, CancellationToken cancellationToken = default)
+ {
+ await _dataLock.WaitAsync(cancellationToken).ConfigureAwait(false);
+ try
+ {
+ long window;
+ lock (_gate)
+ window = (_limits ?? throw new InvalidOperationException("no windowed subscription")).ClientReceive.WindowBytes;
+ if (Outstanding() >= window)
+ {
+ Log($"window full ({Outstanding()} of {window} bytes outstanding); pausing output");
+ while (true)
+ {
+ Task credit;
+ lock (_gate)
+ {
+ if (_sentBytes - _consumedBytes < window) break;
+ credit = _credit.Task;
+ }
+ await credit.WaitAsync(cancellationToken).ConfigureAwait(false);
+ }
+ long consumed;
+ lock (_gate) consumed = _consumedBytes;
+ Log($"credit received (consumedBytes={consumed}); resuming output");
+ }
+ await SendDataAsync(new JsonRpcNotification
+ {
+ Method = "action",
+ Params = Json.SerializeToElement(new ActionEnvelope
+ {
+ Channel = _channel,
+ Action = action,
+ ServerSeq = ++_serverSeq,
+ }),
+ }).ConfigureAwait(false);
+ }
+ finally
+ {
+ _dataLock.Release();
+ }
+ }
+
+ public async ValueTask DisposeAsync()
+ {
+ await _transport.CloseAsync().ConfigureAwait(false);
+ await Task.WhenAll(_run, _processInput).ConfigureAwait(false);
+ _dataLock.Dispose();
+ }
+
+ private static void Log(string message) => Console.WriteLine($" host | {message}");
+
+ private static long Utf16Bytes(string text) => (long)text.Length * 2;
+
+ private long Outstanding()
+ {
+ lock (_gate) return _sentBytes - _consumedBytes;
+ }
+
+ // Callers hold _dataLock, so data messages keep their serverSeq order.
+ private async Task SendDataAsync(JsonRpcNotification notification)
+ {
+ var text = Json.EncodeMessage(new JsonRpcMessage { Notification = notification }).Text!;
+ lock (_gate) _sentBytes += Utf16Bytes(text);
+ await _transport.SendAsync(TransportMessage.FromText(text)).ConfigureAwait(false);
+ }
+
+ private ValueTask SendControlAsync(JsonRpcMessage message) =>
+ _transport.SendAsync(Json.EncodeMessage(message));
+
+ private static JsonRpcMessage Response(ulong id, T result) => new()
+ {
+ SuccessResponse = new JsonRpcSuccessResponse { Id = id, Result = Json.SerializeToElement(result) },
+ };
+
+ private async Task RunAsync()
+ {
+ try
+ {
+ while (true)
+ {
+ TransportMessage frame;
+ try
+ {
+ frame = await _transport.ReceiveAsync().ConfigureAwait(false);
+ }
+ catch (AhpTransportException)
+ {
+ return;
+ }
+ if (frame.Frame != TransportFrame.Text) continue;
+ var message = Json.DecodeMessage(frame);
+ switch (message.Request?.Method ?? message.Notification?.Method)
+ {
+ case "initialize":
+ await SendControlAsync(Response(message.Request!.Id, new InitializeResult
+ {
+ ProtocolVersion = ProtocolVersion.Supported[0],
+ ServerSeq = 0,
+ Snapshots = new List(),
+ })).ConfigureAwait(false);
+ break;
+ case "subscribe":
+ await AcceptSubscriptionAsync(message.Request!).ConfigureAwait(false);
+ break;
+ case "channel/credit":
+ ReceiveCredit(Json.Deserialize(message.Notification!.Params!.Value));
+ break;
+ case "dispatchAction":
+ ReceiveInput(frame.Text!, Json.Deserialize(message.Notification!.Params!.Value).Action);
+ break;
+ }
+ }
+ }
+ finally
+ {
+ _input.Writer.TryComplete();
+ }
+ }
+
+ private void ReceiveCredit(ChannelCreditParams credit)
+ {
+ lock (_gate)
+ {
+ // Credit is a running total, so ignore values that are not larger.
+ if (credit.ConsumedBytes <= _consumedBytes) return;
+ _consumedBytes = credit.ConsumedBytes;
+ _credit.TrySetResult();
+ _credit = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
+ }
+ }
+
+ private async Task AcceptSubscriptionAsync(JsonRpcRequest request)
+ {
+ var parameters = Json.Deserialize(request.Params!.Value);
+ var offered = parameters.FlowControl?.Receive
+ ?? throw new InvalidOperationException("the examples expect a windowed subscribe");
+ await _dataLock.WaitAsync().ConfigureAwait(false);
+ try
+ {
+ _channel = parameters.Channel;
+ // A host can lower the limits that the client offered, but never raise them.
+ var limits = new ChannelFlowControl
+ {
+ ClientReceive = new ChannelReceiveLimits
+ {
+ WindowBytes = Math.Min(offered.WindowBytes, _options.ClientWindowBytes),
+ MaximumFrameBytes = Math.Min(offered.MaximumFrameBytes, MaximumFrameBytes),
+ },
+ HostReceive = new ChannelReceiveLimits
+ {
+ WindowBytes = _options.HostWindowBytes,
+ MaximumFrameBytes = MaximumFrameBytes,
+ },
+ };
+ lock (_gate) _limits = limits;
+ Log($"accepted windowed delivery: client window {limits.ClientReceive.WindowBytes} bytes "
+ + $"(offered {offered.WindowBytes}), host window {limits.HostReceive.WindowBytes} bytes");
+ await SendControlAsync(Response(request.Id, new SubscribeResult { FlowControl = limits })).ConfigureAwait(false);
+
+ // The first data message on a windowed subscription is always the snapshot.
+ await SendDataAsync(new JsonRpcNotification
+ {
+ Method = "channel/snapshot",
+ Params = Json.SerializeToElement(new ChannelSnapshotParams
+ {
+ Channel = _channel,
+ Snapshot = new Snapshot
+ {
+ Resource = _channel,
+ State = new SnapshotState
+ {
+ Terminal = new TerminalState
+ {
+ Title = "demo shell",
+ Content = new List(),
+ Lifecycle = new TerminalLifecycleState(new TerminalRunningLifecycleState()),
+ Claim = new TerminalClaim(new TerminalClientClaim { ClientId = "windowing-example" }),
+ },
+ },
+ FromSeq = _serverSeq,
+ },
+ }),
+ }).ConfigureAwait(false);
+ }
+ finally
+ {
+ _dataLock.Release();
+ }
+ _subscribed.TrySetResult();
+ }
+
+ private void ReceiveInput(string text, StateAction action)
+ {
+ var bytes = Utf16Bytes(text);
+ lock (_gate)
+ {
+ _receivedBytes += bytes;
+ var limits = _limits!.HostReceive;
+ // The client can go past the window only with the message that fills it.
+ if (_receivedBytes - _processedBytes > limits.WindowBytes + limits.MaximumFrameBytes)
+ Log("client went past the host window");
+ }
+ _input.Writer.TryWrite((bytes, action));
+ }
+
+ // Process input one message at a time, then return credit for it.
+ private async Task ProcessInputAsync()
+ {
+ await foreach (var (bytes, action) in _input.Reader.ReadAllAsync().ConfigureAwait(false))
+ {
+ await Task.Delay(_options.InputDelay).ConfigureAwait(false);
+ long processed;
+ lock (_gate)
+ {
+ if (action.Value is TerminalInputAction input) _receivedInput.Append(input.Data);
+ _processedBytes += bytes;
+ processed = _processedBytes;
+ }
+ try
+ {
+ await SendControlAsync(new JsonRpcMessage
+ {
+ Notification = new JsonRpcNotification
+ {
+ Method = "channel/credit",
+ Params = Json.SerializeToElement(new ChannelCreditParams { Channel = _channel, ConsumedBytes = processed }),
+ },
+ }).ConfigureAwait(false);
+ }
+ catch (AhpTransportException)
+ {
+ return;
+ }
+ }
+ }
+}
diff --git a/clients/dotnet/examples/WindowingBackpressure/Program.cs b/clients/dotnet/examples/WindowingBackpressure/Program.cs
new file mode 100644
index 000000000..ff69fae51
--- /dev/null
+++ b/clients/dotnet/examples/WindowingBackpressure/Program.cs
@@ -0,0 +1,170 @@
+// Windowed delivery with backpressure and a custom coalescer.
+//
+// Part 1, receiving: the client renders slowly. While it is busy, the host
+// fills the window and the client queues what arrives. Adjacent queued
+// messages are combined before the renderer sees them:
+//
+// - The built-in coalescer joins adjacent `terminal/data` output.
+// - A custom coalescer keeps only the last of adjacent `terminal/resized`
+// actions, because only the final size matters.
+//
+// Part 2, sending: the client pastes a large block of text into the terminal.
+// The host processes input slowly, so the client's send window fills. The
+// client waits until the subscription is writable before it sends each chunk,
+// instead of queueing the whole paste in memory.
+//
+// Run: dotnet run --project examples/WindowingBackpressure
+#nullable enable
+
+using System;
+using System.Collections.Generic;
+using System.Linq;
+using System.Threading.Tasks;
+using Microsoft.AgentHostProtocol;
+using WindowingExamples;
+
+const string Terminal = "ahp-terminal:/demo";
+
+var host = new WindowingDemoHost(new DemoHostOptions
+{
+ ClientWindowBytes = 2048,
+ HostWindowBytes = 1024,
+ InputDelay = TimeSpan.FromMilliseconds(15),
+});
+var config = new ClientConfig();
+config.Windowing.Coalescers.Add(new LatestResizeCoalescer());
+// Allow at most 2048 bytes of unsent data to wait behind the message being sent.
+config.Windowing.MaxBufferedSendBytes = 2048;
+var client = AhpClient.Connect(host.ClientTransport, config);
+await client.InitializeAsync("windowing-example", ProtocolVersion.Supported);
+
+// Count what the client did. This stream is for monitoring only and can drop
+// events; it never affects credit.
+var counts = new SortedDictionary(StringComparer.Ordinal);
+var counting = CountFlowControlEventsAsync();
+
+// Manual receive mode exposes each delivery's charged size; TakeDeliveriesAsync
+// still releases deliveries automatically. Coalescing is on by default.
+var (_, subscription) = await client.SubscribeWindowedAsync(Terminal,
+ new WindowOptions { WindowBytes = 4096, Receive = ReceiveMode.Manual });
+Log($"subscribed: accepted window={subscription.FlowControl.Limits?.ClientReceive.WindowBytes} bytes");
+
+// ─── Part 1: a slow renderer ────────────────────────────────────────────────
+
+Console.WriteLine("\nPart 1: slow renderer");
+StateAction[] hostActions =
+[
+ Data("Building 120 modules\r\n"),
+ // The user drags the corner of the terminal window.
+ Resize(100, 30),
+ Resize(110, 32),
+ Resize(120, 36),
+ Resize(132, 40),
+ .. Enumerable.Range(1, 10)
+ .Select(step => step * 10)
+ .Select(percent => Data($"[{percent,3}%] compiled {percent * 120 / 100} modules\r\n")),
+ Data("Build finished\r\n"),
+];
+var writer = WriteHostActionsAsync();
+
+var renders = 0;
+// Leaving this loop releases the last delivery but keeps the subscription open.
+await foreach (var delivery in subscription.TakeDeliveriesAsync())
+{
+ renders++;
+ if (delivery.Event is SubscriptionEventSnapshot)
+ {
+ Log($"render snapshot ({delivery.WireBytes} bytes)");
+ }
+ else if (delivery.Event is SubscriptionEventAction { Envelope.Action.Value: TerminalResizedAction resize })
+ {
+ Log($"render resize to {resize.Cols}x{resize.Rows} ({delivery.WireBytes} bytes)");
+ }
+ else if (delivery.Event is SubscriptionEventAction { Envelope.Action.Value: TerminalDataAction data })
+ {
+ var lines = data.Data.Split("\r\n", StringSplitOptions.RemoveEmptyEntries);
+ Log($"render {lines.Length} line(s) of output ({delivery.WireBytes} bytes): {DemoSupport.Quote(lines[^1])}");
+ if (data.Data.Contains("Build finished", StringComparison.Ordinal)) break;
+ }
+ await Task.Delay(60); // The renderer is slow; the host fills the window meanwhile.
+}
+await writer;
+Log($"rendered {renders} deliveries for {hostActions.Length + 1} host messages");
+
+// ─── Part 2: pasting into a slow terminal ───────────────────────────────────
+
+Console.WriteLine("\nPart 2: pasting into a slow terminal");
+var paste = Enumerable.Range(1, 8).Select(i => $"echo \"line {i} of the pasted script\"\n").ToArray();
+var waits = 0;
+for (var i = 0; i < paste.Length; i++)
+{
+ // Wait for room in the host window and the local send queue, rather than
+ // queueing the whole paste. With one producer, this keeps sends within the limit.
+ var status = subscription.FlowControl;
+ if (!status.Writable)
+ {
+ Log($"not writable ({status.OutstandingSendBytes} bytes awaiting credit, {status.BufferedSendBytes} bytes queued); waiting");
+ waits++;
+ await subscription.WaitWritableAsync();
+ }
+ Log($"send chunk {i + 1} of {paste.Length}");
+ // Completes when the message is written to the transport. If another
+ // producer filled the local send queue first, the .NET client waits for
+ // room in the queue rather than failing, so there is nothing to retry.
+ await client.DispatchAsync(Terminal, new StateAction(new TerminalInputAction { Data = paste[i] }));
+}
+// Wait until the host has processed, and credited, every chunk.
+await foreach (var current in subscription.FlowControlChanges().ReadAllAsync())
+{
+ if (current.OutstandingSendBytes == 0 && current.BufferedSendBytes == 0) break;
+}
+var intact = host.ReceivedInput == string.Concat(paste);
+Log($"sent {paste.Length} chunks, waiting {waits} times; host received the paste intact: {(intact ? "true" : "false")}");
+
+await subscription.DisposeAsync();
+await client.DisposeAsync();
+await host.DisposeAsync();
+
+// Shutting down the client completes the event stream, so every event is counted.
+await counting;
+Console.WriteLine("\nFlow-control events:");
+foreach (var (kind, count) in counts) Console.WriteLine($" {kind}: {count}");
+
+async Task CountFlowControlEventsAsync()
+{
+ await foreach (var flowEvent in client.FlowControlEvents().ReadAllAsync())
+ {
+ counts[flowEvent.Kind] = counts.GetValueOrDefault(flowEvent.Kind) + 1;
+ }
+}
+
+async Task WriteHostActionsAsync()
+{
+ await host.WhenSubscribedAsync();
+ foreach (var action in hostActions) await host.SendActionAsync(action);
+}
+
+static StateAction Data(string data) => new(new TerminalDataAction { Data = data });
+
+static StateAction Resize(long cols, long rows) => new(new TerminalResizedAction { Cols = cols, Rows = rows });
+
+static void Log(string message) => Console.WriteLine($" client | {message}");
+
+// A custom coalescer runs on two adjacent messages that are still queued. It
+// returns the combined message, or null to leave them separate. The client
+// never passes it messages that the protocol requires to stay separate, such
+// as actions with an origin.
+internal sealed class LatestResizeCoalescer : IChannelMessageCoalescer
+{
+ public JsonRpcNotification? Coalesce(JsonRpcNotification earlier, JsonRpcNotification later, CoalescingContext context)
+ {
+ if (context.Direction != CoalescingDirection.Incoming || earlier.Method != "action") return null;
+ return IsResize(earlier) && IsResize(later) ? later : null;
+ }
+
+ private static bool IsResize(JsonRpcNotification notification) =>
+ notification.Params is { } parameters
+ && parameters.TryGetProperty("action", out var action)
+ && action.TryGetProperty("type", out var type)
+ && type.GetString() == ActionType.TerminalResized.Value;
+}
diff --git a/clients/dotnet/examples/WindowingBackpressure/WindowingBackpressure.csproj b/clients/dotnet/examples/WindowingBackpressure/WindowingBackpressure.csproj
new file mode 100644
index 000000000..77766f1ee
--- /dev/null
+++ b/clients/dotnet/examples/WindowingBackpressure/WindowingBackpressure.csproj
@@ -0,0 +1,21 @@
+
+
+
+ net8.0
+ Exe
+ false
+ false
+
+ Major
+
+
+
+
+
+
+
+
+
+
+
+
diff --git a/clients/dotnet/examples/WindowingBasic/Program.cs b/clients/dotnet/examples/WindowingBasic/Program.cs
new file mode 100644
index 000000000..7f71d06db
--- /dev/null
+++ b/clients/dotnet/examples/WindowingBasic/Program.cs
@@ -0,0 +1,86 @@
+// Basic windowed delivery.
+//
+// The client offers windowed delivery on a terminal subscription. The host
+// accepts a smaller window, then sends a snapshot and terminal output. When
+// the window is full, the host waits; each time the client takes the next
+// delivery, it releases the previous one and returns credit, so the host can
+// continue.
+//
+// Run: dotnet run --project examples/WindowingBasic
+#nullable enable
+
+using System;
+using System.Threading.Tasks;
+using Microsoft.AgentHostProtocol;
+using WindowingExamples;
+
+const string Terminal = "ahp-terminal:/demo";
+string[] output =
+[
+ "$ npm test",
+ "> demo@1.0.0 test",
+ "> node --test",
+ "ok 1 - parses the configuration file",
+ "ok 2 - rejects an unknown option",
+ "ok 3 - reads input from stdin",
+ "ok 4 - writes the report",
+ "# pass 4, fail 0",
+];
+
+// The host accepts at most a 1024-byte window, so it fills after a few messages.
+await using var host = new WindowingDemoHost(new DemoHostOptions { ClientWindowBytes = 1024, HostWindowBytes = 4096 });
+await using var client = AhpClient.Connect(host.ClientTransport);
+await client.InitializeAsync("windowing-example", ProtocolVersion.Supported);
+
+// Windowing is off by default. Opt in for this subscription and offer a
+// 4096-byte window. Coalescing is turned off so that every host message is a
+// separate delivery; see WindowingBackpressure for coalescing. Manual receive
+// mode exposes each delivery's charged size; TakeDeliveriesAsync still
+// releases deliveries automatically.
+var (_, subscription) = await client.SubscribeWindowedAsync(Terminal, new WindowOptions
+{
+ WindowBytes = 4096,
+ Coalescing = false,
+ Receive = ReceiveMode.Manual,
+});
+var flowControl = subscription.FlowControl;
+Log($"subscribed: mode={flowControl.Mode}, accepted window={flowControl.Limits?.ClientReceive.WindowBytes} bytes");
+
+// The host writes terminal output as fast as the window allows.
+var writer = WriteOutputAsync();
+
+// With automatic release, taking the next delivery releases the previous one.
+// The client then returns credit for it.
+var lines = 0;
+await foreach (var delivery in subscription.TakeDeliveriesAsync())
+{
+ var buffered = subscription.FlowControl.BufferedReceiveBytes;
+ switch (delivery.Event)
+ {
+ case SubscriptionEventSnapshot:
+ Log($"snapshot ({delivery.WireBytes} bytes; {buffered} bytes not yet released)");
+ break;
+ case SubscriptionEventAction { Envelope.Action.Value: TerminalDataAction data }:
+ Log($"output {DemoSupport.Quote(data.Data.TrimEnd())} ({delivery.WireBytes} bytes; {buffered} bytes not yet released)");
+ lines++;
+ break;
+ }
+ await Task.Delay(20); // Simulate rendering the delivery.
+ if (lines == output.Length) break; // Leaving the loop releases the last delivery.
+}
+await subscription.DisposeAsync(); // Unsubscribe.
+
+Log($"received all {lines} lines");
+await writer;
+// Leaving the program disposes the client, which shuts it down, and then the host.
+
+async Task WriteOutputAsync()
+{
+ await host.WhenSubscribedAsync();
+ foreach (var line in output)
+ {
+ await host.SendActionAsync(new StateAction(new TerminalDataAction { Data = line + "\r\n" }));
+ }
+}
+
+static void Log(string message) => Console.WriteLine($" client | {message}");
diff --git a/clients/dotnet/examples/WindowingBasic/WindowingBasic.csproj b/clients/dotnet/examples/WindowingBasic/WindowingBasic.csproj
new file mode 100644
index 000000000..77766f1ee
--- /dev/null
+++ b/clients/dotnet/examples/WindowingBasic/WindowingBasic.csproj
@@ -0,0 +1,21 @@
+
+
+
+ net8.0
+ Exe
+ false
+ false
+
+ Major
+
+
+
+
+
+
+
+
+
+
+
+
diff --git a/clients/dotnet/src/AgentHostProtocol.Abstractions/Generated/State.generated.cs b/clients/dotnet/src/AgentHostProtocol.Abstractions/Generated/State.generated.cs
index 9d5ab51ce..63b8dc9e0 100644
--- a/clients/dotnet/src/AgentHostProtocol.Abstractions/Generated/State.generated.cs
+++ b/clients/dotnet/src/AgentHostProtocol.Abstractions/Generated/State.generated.cs
@@ -8013,11 +8013,10 @@ public override SnapshotState Read(ref Utf8JsonReader reader, Type typeToConvert
{
result.Chat = root.Deserialize(AhpJsonTypeInfo.Get(options));
}
- else if (root.TryGetProperty("lifecycle", out _))
+ else if (root.TryGetProperty("lifecycle", out _) && root.TryGetProperty("provider", out _))
{
- // SessionState is discriminated on its required `lifecycle` field.
- // (It no longer carries `summary`; that field was removed when the
- // session state was flattened.)
+ // SessionState is discriminated on its required `lifecycle` and
+ // `provider` fields; TerminalState also has `lifecycle`.
result.Session = root.Deserialize(AhpJsonTypeInfo.Get(options));
}
else if (root.TryGetProperty("instanceId", out _) &&
diff --git a/clients/dotnet/src/AgentHostProtocol.Abstractions/Transport/ITransportMessageLimits.cs b/clients/dotnet/src/AgentHostProtocol.Abstractions/Transport/ITransportMessageLimits.cs
new file mode 100644
index 000000000..f3545bbd0
--- /dev/null
+++ b/clients/dotnet/src/AgentHostProtocol.Abstractions/Transport/ITransportMessageLimits.cs
@@ -0,0 +1,10 @@
+#nullable enable
+
+namespace Microsoft.AgentHostProtocol;
+
+/// Optional transport capability describing its complete inbound wire-message limit.
+public interface ITransportMessageLimits
+{
+ /// Maximum wire bytes per message, or null when unknown or unlimited.
+ long? MaximumInboundMessageBytes { get; }
+}
diff --git a/clients/dotnet/src/AgentHostProtocol/AhpClient.cs b/clients/dotnet/src/AgentHostProtocol/AhpClient.cs
index e90eb47d0..b9b74417e 100644
--- a/clients/dotnet/src/AgentHostProtocol/AhpClient.cs
+++ b/clients/dotnet/src/AgentHostProtocol/AhpClient.cs
@@ -7,6 +7,7 @@
using System.Collections.Generic;
using System.Diagnostics;
using System.Diagnostics.CodeAnalysis;
+using System.Linq;
using System.Text.Json;
using System.Threading;
using System.Threading.Channels;
@@ -72,6 +73,8 @@ public static KeepAlivePolicy Enabled(TimeSpan? interval = null, TimeSpan? timeo
/// Tuning knobs for an .
public sealed class ClientConfig
{
+ /// Defaults for subscriptions explicitly opting into windowed delivery.
+ public WindowingConfig Windowing { get; set; } = new();
///
/// How long waits for a
/// response. Zero disables the timeout. Defaults to 30 seconds.
@@ -112,6 +115,7 @@ internal static ClientConfig Snapshot(ClientConfig? config)
? source.SubscriptionBufferCapacity
: 256,
KeepAlive = source.KeepAlive ?? KeepAlivePolicy.Disabled,
+ Windowing = (source.Windowing ?? new WindowingConfig()).Snapshot(),
};
}
}
@@ -156,7 +160,7 @@ public enum ConnectionState
/// All public methods are safe to call from multiple threads.
///
///
-public sealed class AhpClient : IAhpClient
+public sealed partial class AhpClient : IWindowedAhpClient
{
// ── State that lives for the client lifetime ──────────────────────────
@@ -240,6 +244,7 @@ internal int SubscriptionCount
// Lifecycle
private readonly TaskCompletionSource _doneTcs = new(TaskCreationOptions.RunContinuationsAsynchronously);
+ private readonly TaskCompletionSource _teardownTcs = new(TaskCreationOptions.RunContinuationsAsynchronously);
private readonly CancellationTokenSource _lifetimeCts = new();
private int _shutdownStarted; // 0 = running, 1 = shut down
private Exception? _closeErr;
@@ -254,6 +259,8 @@ private sealed class OutboundMessage
public JsonRpcMessage Message { get; }
public TaskCompletionSource? Sent { get; }
public CancellationToken CancellationToken { get; }
+ public bool IsBarrier { get; init; }
+ public WindowResponse? WindowRequest { get; init; }
public OutboundMessage(
JsonRpcMessage message,
@@ -329,11 +336,26 @@ public static AhpClient Connect(
///
public async Task ShutdownAsync(CancellationToken cancellationToken = default)
{
- await ShutdownWithErrorAsync(null).ConfigureAwait(false);
+ try
+ {
+ if (Volatile.Read(ref _shutdownStarted) == 0)
+ await DrainOutboundAsync(cancellationToken).ConfigureAwait(false);
+ }
+ catch (Exception) when (!cancellationToken.IsCancellationRequested) { }
+ finally { await ShutdownWithErrorAsync(null).ConfigureAwait(false); }
+ await _teardownTcs.Task.ConfigureAwait(false);
// Wait for both background tasks to exit.
await Task.WhenAll(_readerTask, _writerTask).WaitAsync(cancellationToken).ConfigureAwait(false);
}
+ // An uninstalled connection attempt has no application sends to drain.
+ internal async Task AbortAsync()
+ {
+ await ShutdownWithErrorAsync(null).ConfigureAwait(false);
+ await _teardownTcs.Task.ConfigureAwait(false);
+ await Task.WhenAll(_readerTask, _writerTask).ConfigureAwait(false);
+ }
+
///
public async ValueTask DisposeAsync()
{
@@ -360,6 +382,7 @@ private async Task ShutdownWithErrorAsync(Exception? cause, bool fromKeepAlive =
Volatile.Write(ref _closeErr, cause);
_lifetimeCts.Cancel();
+ EndWindows(cause);
// Signal the done task so Done-waiters unblock.
_doneTcs.TrySetResult(true);
@@ -400,6 +423,7 @@ private async Task ShutdownWithErrorAsync(Exception? cause, bool fromKeepAlive =
var shutdownEx = cause is null
? new AhpClientClosedException()
: new AhpClientClosedException($"ahp: client shut down: {cause.Message}");
+ while (_outbound.Reader.TryRead(out var queued)) queued.Sent?.TrySetException(shutdownEx);
foreach (var kv in _pending)
{
@@ -439,6 +463,17 @@ private async Task ShutdownWithErrorAsync(Exception? cause, bool fromKeepAlive =
_stateListeners.Clear();
}
foreach (var st in allStateListeners) st.Close();
+ _teardownTcs.TrySetResult(true);
+ }
+
+ private async Task DrainOutboundAsync(CancellationToken cancellationToken)
+ {
+ using var timeout = new CancellationTokenSource(TimeSpan.FromSeconds(2));
+ using var linked = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken, timeout.Token, _lifetimeCts.Token);
+ var sent = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
+ await _outbound.Writer.WriteAsync(new OutboundMessage(new JsonRpcMessage(), sent) { IsBarrier = true }, linked.Token).ConfigureAwait(false);
+ _writerWake.Release();
+ await sent.Task.WaitAsync(linked.Token).ConfigureAwait(false);
}
// ── Connection state ──────────────────────────────────────────────────
@@ -533,49 +568,65 @@ await ShutdownWithErrorAsync(
private async Task RunWriterAsync()
{
+ OutboundMessage? active = null;
try
{
- await foreach (var item in _outbound.Reader.ReadAllAsync(_lifetimeCts.Token).ConfigureAwait(false))
+ var controlBurst = 0;
+ while (!_lifetimeCts.IsCancellationRequested)
{
- if (item.CancellationToken.IsCancellationRequested)
- {
- item.Sent?.TrySetCanceled(item.CancellationToken);
- continue;
- }
-
- var frame = _serializer.EncodeMessage(item.Message);
- if (item.CancellationToken.IsCancellationRequested)
- {
- item.Sent?.TrySetCanceled(item.CancellationToken);
- continue;
- }
- try
+ TransportMessage frame;
+ Action? sent = null;
+ active = null;
+ if ((controlBurst >= 8 || !_outbound.Reader.TryPeek(out _)) && TryWindowFrame(out frame, out sent))
{
- // Canceling ClientWebSocket.SendAsync aborts the shared connection,
- // so message cancellation only prevents writes that have not started.
- await _transport.SendAsync(frame, _lifetimeCts.Token).ConfigureAwait(false);
- item.Sent?.TrySetResult(true);
+ controlBurst = 0;
}
- catch (OperationCanceledException) when (_lifetimeCts.IsCancellationRequested)
+ else if (_outbound.Reader.TryRead(out var item))
{
- item.Sent?.TrySetException(new AhpClientClosedException());
- return;
+ active = item;
+ controlBurst++;
+ if (item.IsBarrier)
+ {
+ item.Sent?.TrySetResult(true);
+ continue;
+ }
+ if (item.CancellationToken.IsCancellationRequested)
+ {
+ item.Sent?.TrySetCanceled(item.CancellationToken);
+ continue;
+ }
+ frame = _serializer.EncodeMessage(item.Message);
+ if (item.CancellationToken.IsCancellationRequested)
+ {
+ item.Sent?.TrySetCanceled(item.CancellationToken);
+ continue;
+ }
+ if (item.WindowRequest is { } response && !response.TryStart())
+ {
+ item.Sent?.TrySetException(response.Failure ?? new AhpClientClosedException());
+ continue;
+ }
}
- catch (Exception ex)
+ else
{
- item.Sent?.TrySetException(ex);
- await ShutdownWithErrorAsync(new AhpTransportException("io", $"ahp: transport send: {ex.Message}", ex)).ConfigureAwait(false);
- return;
+ await _writerWake.WaitAsync(_lifetimeCts.Token).ConfigureAwait(false);
+ continue;
}
+ await _transport.SendAsync(frame, _lifetimeCts.Token).ConfigureAwait(false);
+ sent?.Invoke();
+ active?.Sent?.TrySetResult(true);
}
}
catch (OperationCanceledException) when (_lifetimeCts.IsCancellationRequested)
{
+ active?.Sent?.TrySetException(new AhpClientClosedException());
// Normal teardown canceled the channel enumeration.
}
catch (Exception ex)
{
- await ShutdownWithErrorAsync(new AhpTransportException("io", $"ahp: writer: {ex.Message}", ex)).ConfigureAwait(false);
+ active?.Sent?.TrySetException(ex);
+ await ShutdownWithErrorAsync(ex is AhpTransportException ? ex
+ : new AhpTransportException("io", $"ahp: writer: {ex.Message}", ex)).ConfigureAwait(false);
}
}
@@ -617,24 +668,28 @@ private async Task RunReaderAsync()
}
catch
{
+ lock (_windowGate)
+ if (_windows.Values.Any(window => window.Limits is not null))
+ throw Protocol("malformed JSON on a windowed connection");
// Skip malformed frames; protocol resync is the server's responsibility.
AhpTelemetry.MalformedFrames.Add(1);
continue;
}
AhpTelemetry.MessagesReceived.Add(1);
- Dispatch(msg);
+ Dispatch(msg, msg.Notification is not null && RequiresWindowSizes ? RawWindowBytes(frame) : 0);
}
}
catch (Exception ex) when (ex is not OperationCanceledException)
{
- await ShutdownWithErrorAsync(new AhpTransportException("io", $"ahp: reader: {ex.Message}", ex)).ConfigureAwait(false);
+ await ShutdownWithErrorAsync(ex is AhpTransportException ? ex
+ : new AhpTransportException("protocol", $"ahp: reader: {ex.Message}", ex)).ConfigureAwait(false);
}
}
// ── Dispatch ──────────────────────────────────────────────────────────
- private void Dispatch(JsonRpcMessage msg)
+ private void Dispatch(JsonRpcMessage msg, long wireBytes)
{
if (msg.SuccessResponse is not null)
{
@@ -647,7 +702,8 @@ private void Dispatch(JsonRpcMessage msg)
}
else if (msg.Notification is not null)
{
- HandleNotification(msg.Notification);
+ if (!HandleWindowNotification(msg.Notification, wireBytes))
+ HandleNotification(msg.Notification);
}
else if (msg.Request is not null)
{
@@ -660,6 +716,8 @@ private void Dispatch(JsonRpcMessage msg)
private void Deliver(ulong id, JsonElement result, AhpRpcException? rpcError)
{
+ if (_windowResponses.TryRemove(id, out var hook))
+ hook.Apply(rpcError is null ? result : null, rpcError);
if (_pending.TryRemove(id, out var tcs))
{
if (rpcError is not null)
@@ -734,6 +792,7 @@ private async Task EnqueueReplyAsync(JsonRpcMessage msg)
try
{
await _outbound.Writer.WriteAsync(new OutboundMessage(msg)).ConfigureAwait(false);
+ _writerWake.Release();
}
catch { /* shutting down — best effort */ }
}
@@ -894,6 +953,9 @@ private void FanOut(string channel, SubscriptionEvent ev)
}
JsonElement? paramsEl = parameters is null ? null : _serializer.SerializeToElement(parameters);
+ WindowOptions? requestedOptions = null;
+ if (parameters is SubscribeParams subscribe) _windowRequestOptions.TryGetValue(subscribe, out requestedOptions);
+ PrepareWindowResponse(id, method, paramsEl, requestedOptions, requestCts.Token);
var req = new JsonRpcMessage
{
@@ -933,6 +995,7 @@ private void FanOut(string channel, SubscriptionEvent ev)
}
catch (Exception ex)
{
+ AbandonWindowRequest(id, ex);
// Distinguish caller cancellation (the caller's token) from a request
// timeout (the configured default-timeout fired its linked token) from a
// genuine error — the metric and the span status differ for each.
@@ -987,12 +1050,27 @@ public async Task NotifyAsync(
private async Task SendMessageAsync(JsonRpcMessage msg, CancellationToken cancellationToken)
{
+ if (msg.Notification is { Params: { } p } notification && p.TryGetProperty("channel", out var channelEl)
+ && channelEl.GetString() is { } channel && GetWindowedSubscription(channel) is { } owner)
+ {
+ if (notification.Method is "channel/frame" or "channel/credit" or "channel/snapshot")
+ throw new InvalidOperationException("Window framing and credit are owned by the client.");
+ if ((owner.Limits is not null || owner.Negotiating) && WindowedDelivery.CountsAgainstWindow(notification.Method))
+ {
+ await SendWindowAsync(owner, notification, cancellationToken).ConfigureAwait(false);
+ return;
+ }
+ }
var sentTcs = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
- var item = new OutboundMessage(msg, sentTcs, cancellationToken);
+ var item = new OutboundMessage(msg, sentTcs, cancellationToken)
+ {
+ WindowRequest = msg.Request is { } request && _windowResponses.TryGetValue(request.Id, out var response) ? response : null,
+ };
try
{
await _outbound.Writer.WriteAsync(item, cancellationToken).ConfigureAwait(false);
+ _writerWake.Release();
}
catch (OperationCanceledException)
{
@@ -1190,6 +1268,7 @@ public Subscription AttachSubscription(string uri)
public async Task UnsubscribeAsync(string uri, CancellationToken cancellationToken = default)
{
Guard.ThrowIfNull(uri, nameof(uri));
+ StopWindow(uri);
List subs;
lock (_subsLock)
{
diff --git a/clients/dotnet/src/AgentHostProtocol/Hosts/HostClientHandle.cs b/clients/dotnet/src/AgentHostProtocol/Hosts/HostClientHandle.cs
index 733e50902..1f9ccfd57 100644
--- a/clients/dotnet/src/AgentHostProtocol/Hosts/HostClientHandle.cs
+++ b/clients/dotnet/src/AgentHostProtocol/Hosts/HostClientHandle.cs
@@ -57,6 +57,18 @@ private AhpClient CheckAlive()
///
public void CheckAliveOrThrow() => CheckAlive();
+ /// Returns the authoritative windowed owner on this connection, if any.
+ public WindowedSubscription? GetWindowedSubscription(string channel) => CheckAlive().GetWindowedSubscription(channel);
+
+ /// Waits for channel credit and queue capacity on this connection.
+ public ValueTask WaitWritableAsync(string channel, CancellationToken cancellationToken = default) =>
+ (CheckAlive().GetWindowedSubscription(channel)
+ ?? throw new InvalidOperationException("The channel has no windowed owner.")).WaitWritableAsync(cancellationToken);
+
+ /// Sends a notification through the connection's flow-controlled scheduler.
+ public Task NotifyAsync(string method, T parameters, CancellationToken cancellationToken = default) =>
+ CheckAlive().NotifyAsync(method, parameters, cancellationToken);
+
///
/// Dispatches an action through this connection on ,
/// refusing (throwing) if the host was removed or the connection has been
diff --git a/clients/dotnet/src/AgentHostProtocol/Hosts/IMultiHostWindowedClient.cs b/clients/dotnet/src/AgentHostProtocol/Hosts/IMultiHostWindowedClient.cs
new file mode 100644
index 000000000..ebd02b096
--- /dev/null
+++ b/clients/dotnet/src/AgentHostProtocol/Hosts/IMultiHostWindowedClient.cs
@@ -0,0 +1,24 @@
+#nullable enable
+
+using System.Threading;
+using System.Threading.Tasks;
+
+namespace Microsoft.AgentHostProtocol.Hosts;
+
+public interface IMultiHostWindowedClient : IMultiHostClient
+{
+ Task<(SubscribeResult Result, WindowedSubscription Sub)> SubscribeWindowedAsync(
+ HostId host, string uri, WindowOptions? options = null, CancellationToken cancellationToken = default);
+ WindowedSubscription? GetWindowedSubscription(HostId host, string uri);
+ System.Threading.Channels.ChannelReader FlowControlEvents();
+}
+
+public static class MultiHostWindowedExtensions
+{
+ public static Task<(SubscribeResult Result, WindowedSubscription Sub)> SubscribeWindowedAsync(
+ this IMultiHostClient client, HostId host, string uri, WindowOptions? options = null,
+ CancellationToken cancellationToken = default) =>
+ client is IMultiHostWindowedClient windowed
+ ? windowed.SubscribeWindowedAsync(host, uri, options, cancellationToken)
+ : throw new NotSupportedException("This runtime does not implement windowed delivery.");
+}
diff --git a/clients/dotnet/src/AgentHostProtocol/Hosts/MultiHostClient.cs b/clients/dotnet/src/AgentHostProtocol/Hosts/MultiHostClient.cs
index 2a629c551..a3f90a47a 100644
--- a/clients/dotnet/src/AgentHostProtocol/Hosts/MultiHostClient.cs
+++ b/clients/dotnet/src/AgentHostProtocol/Hosts/MultiHostClient.cs
@@ -17,6 +17,33 @@ namespace Microsoft.AgentHostProtocol.Hosts;
internal sealed class HostEntry : IDisposable
{
+ internal readonly ConcurrentDictionary WindowOwners = new(StringComparer.Ordinal);
+ internal void RegisterWindowOwner(WindowedSubscription owner)
+ {
+ lock (_gate)
+ {
+ WindowOwners[owner.Uri] = owner;
+ if (!_subscriptions.Contains(owner.Uri)) _subscriptions.Add(owner.Uri);
+ }
+ }
+
+ internal void ForgetWindowOwner(WindowedSubscription owner)
+ {
+ lock (_gate)
+ if (WindowOwners.TryGetValue(owner.Uri, out var current) && ReferenceEquals(current, owner))
+ {
+ WindowOwners.TryRemove(owner.Uri, out _);
+ _subscriptions.Remove(owner.Uri);
+ }
+ }
+ internal void ApplyWindowRoot(RootState root)
+ {
+ lock (_gate)
+ {
+ _agents = root.Agents is { } agents ? new List(agents) : new List();
+ _activeSessions = root.ActiveSessions;
+ }
+ }
public HostId Id { get; }
public HostConfig Config { get; }
public string ClientId { get; set; }
@@ -437,7 +464,7 @@ public bool ApplyRootAction(StateAction action)
/// Multi-host registry + reconnect supervisor. Manages N independent AHP hosts,
/// fans in their inbound events, and supervises reconnects per-host policy.
///
-public sealed class MultiHostClient : IMultiHostClient
+public sealed partial class MultiHostClient : IMultiHostWindowedClient
{
private readonly ConcurrentDictionary _hosts = new(StringComparer.Ordinal);
private volatile IClientIdStore _store;
@@ -580,6 +607,7 @@ public async Task AddHostAsync(
if (!_hosts.TryAdd(ownedConfig.Id.ToString(), entry))
{
entry.Dispose();
+ DropWindowOwners(entry);
throw new DuplicateHostException(ownedConfig.Id);
}
@@ -595,6 +623,7 @@ public async Task AddHostAsync(
{
entry.LifetimeCts.Cancel();
entry.Dispose();
+ DropWindowOwners(entry);
}
throw;
}
@@ -660,6 +689,7 @@ public async Task RemoveHostAsync(HostId id, CancellationToken cancellationToken
FinishPerHostListeners(id.ToString());
entry!.LifetimeCts.Cancel();
+ DropWindowOwners(entry);
await entry.ConnectionGate.WaitAsync(CancellationToken.None).ConfigureAwait(false);
try
{
@@ -727,6 +757,7 @@ public async Task ShutdownAsync(CancellationToken cancellationToken = default)
var entries = new List(_hosts.Values);
_hosts.Clear();
+ FinishFlowEvents();
// Finish per-host listener streams for every host so their consumers'
// `await foreach` loops exit (mirrors the perResourceListeners finish in
@@ -736,6 +767,7 @@ public async Task ShutdownAsync(CancellationToken cancellationToken = default)
foreach (var entry in entries)
{
entry.LifetimeCts.Cancel();
+ DropWindowOwners(entry);
// Wake a parked (failed/disabled) supervisor so it observes the
// cancellation and exits rather than blocking on its manual-reconnect
// wait forever.
@@ -805,6 +837,7 @@ private async Task OpenHostAsync(HostEntry entry, CancellationToken cancellation
await openTask.ConfigureAwait(false);
return;
}
+ await attempt.AbortClientAsync().ConfigureAwait(false);
if (openTask.IsCompleted)
{
await openTask.ConfigureAwait(false);
@@ -838,7 +871,24 @@ private async Task OpenHostCoreAsync(
transport,
entry.Config.ClientConfig,
null);
+ attempt.TrackClient(client, cancellationToken);
transport = null;
+ client.RetainManagedWindows = true;
+ client.WindowHostId = entry.Id;
+ client.ManagedFlowEvent = BroadcastFlowEvent;
+ client.ManagedWindowEvent = (owner, notification, value) => ApplyManagedWindowEvent(entry, owner, notification, value);
+ foreach (var pair in entry.WindowOwners)
+ {
+ if (pair.Value.Closed)
+ entry.ForgetWindowOwner(pair.Value);
+ else if (pair.Key.StartsWith("ahp-tcp:", StringComparison.Ordinal))
+ {
+ pair.Value.End(new HostNotConnectedException(entry.Id));
+ entry.WindowOwners.TryRemove(pair.Key, out _);
+ entry.RemoveSubscription(pair.Key);
+ }
+ else if (!client.BindManagedWindow(pair.Value)) entry.ForgetWindowOwner(pair.Value);
+ }
// Register before the first handshake request so notifications that
// race initialize/reconnect are buffered rather than discarded.
var stream = client.CreateEventStream();
@@ -856,9 +906,29 @@ private async Task OpenHostCoreAsync(
ReconnectResult? reconnectResult = null;
try
{
- reconnectResult = await client.ReconnectAsync(
- snap.ClientId, snap.ServerSeq, subscriptions, cancellationToken)
- .ConfigureAwait(false);
+ var items = new List();
+ foreach (var pair in entry.WindowOwners)
+ if (pair.Value.ResumeWindowed)
+ items.Add(new ChannelDeliveryResumeParams
+ {
+ Channel = pair.Key,
+ FlowControl = new SubscriptionFlowControlOptions { Receive = pair.Value.Offer },
+ LastAppliedServerSeq = pair.Value.LastAppliedServerSeq,
+ });
+ reconnectResult = await client.ResumeAsync(new ReconnectParams
+ {
+ Channel = ProtocolVersion.RootResourceUri,
+ ClientId = snap.ClientId,
+ LastSeenServerSeq = snap.ServerSeq,
+ Subscriptions = new List(subscriptions),
+ Windows = items.Count == 0 ? null : new ChannelDeliveryResumeOptions { Items = items },
+ }, cancellationToken).ConfigureAwait(false);
+ foreach (var pair in entry.WindowOwners)
+ if (pair.Value.Closed)
+ {
+ entry.WindowOwners.TryRemove(pair.Key, out _);
+ entry.RemoveSubscription(pair.Key);
+ }
}
catch (Exception) when (!cancellationToken.IsCancellationRequested)
{
@@ -881,7 +951,7 @@ private async Task OpenHostCoreAsync(
() =>
{
if (summaries is not null) entry.SeedSessionSummaries(summaries);
- ApplyReconnectReplay(entry, replay);
+ ApplyReconnectReplay(entry, replay, snap.ServerSeq);
});
installed = true;
}
@@ -919,9 +989,20 @@ private async Task OpenHostCoreAsync(
var result = await client.InitializeAsync(
entry.ClientId,
entry.Config.ProtocolVersions,
- subscriptions,
+ subscriptions is null ? null : new List(subscriptions).FindAll(uri =>
+ !entry.WindowOwners.ContainsKey(uri) && !uri.StartsWith("ahp-tcp:", StringComparison.Ordinal)),
cancellationToken)
.ConfigureAwait(false);
+ foreach (var pair in entry.WindowOwners)
+ {
+ if (pair.Value.Closed) continue;
+ try { await client.SubscribeWindowedAsync(pair.Key, pair.Value.Options, cancellationToken).ConfigureAwait(false); }
+ catch (Exception error) when (!cancellationToken.IsCancellationRequested && !client.Completion.IsCompleted)
+ {
+ pair.Value.End(error, true);
+ client.DropManagedWindow(pair.Value);
+ }
+ }
// Extract the root-state snapshot (agents + activeSessions) that the
// server returned for the root channel, mirroring the
@@ -953,7 +1034,7 @@ private async Task OpenHostCoreAsync(
{
if (client is not null)
{
- try { await client.ShutdownAsync(CancellationToken.None).ConfigureAwait(false); } catch { }
+ try { await client.AbortAsync().ConfigureAwait(false); } catch { }
}
else if (transport is not null)
{
@@ -1016,9 +1097,9 @@ private void CompleteOpenHost(HostEntry entry, EventStream stream)
/// subscribed before the drop observe the actions missed while disconnected.
/// URIs in Missing are pruned from the reconnect subscription set.
///
- private ulong ApplyReconnectReplay(HostEntry entry, ReconnectReplayResult replay)
+ private ulong ApplyReconnectReplay(HostEntry entry, ReconnectReplayResult replay, long? previousServerSeq = null)
{
- var initialSeq = entry.Snapshot().ServerSeq;
+ var initialSeq = previousServerSeq ?? entry.Snapshot().ServerSeq;
var previousSeq = initialSeq;
if (replay.Actions is { } seqScan)
{
@@ -1032,12 +1113,13 @@ private ulong ApplyReconnectReplay(HostEntry entry, ReconnectReplayResult replay
}
}
- var generation = entry.ApplyConnected(null, initialSeq);
+ var generation = entry.ApplyConnected(null, Math.Max(initialSeq, entry.Snapshot().ServerSeq));
if (replay.Actions is { } actions)
{
foreach (var env in actions)
{
+ if (entry.CurrentClient?.GetWindowedSubscription(env.Channel) is not null) continue;
var evt = new SubscriptionEventAction(env);
ApplyEventToHostState(entry, evt);
if (env.Channel == ProtocolVersion.RootResourceUri)
@@ -1074,9 +1156,12 @@ private ulong ApplyReconnectSnapshot(HostEntry entry, ReconnectSnapshotResult re
{
if (snapshot.FromSeq > lastSeq) lastSeq = snapshot.FromSeq;
}
- var generation = entry.ApplyConnected(ExtractRootSnapshot(result.Snapshots), lastSeq);
+ var generation = entry.ApplyConnected(
+ entry.CurrentClient?.GetWindowedSubscription(ProtocolVersion.RootResourceUri) is null
+ ? ExtractRootSnapshot(result.Snapshots) : null, lastSeq);
foreach (var snapshot in result.Snapshots)
{
+ if (entry.CurrentClient?.GetWindowedSubscription(snapshot.Resource) is not null) continue;
var evt = new SubscriptionEventSnapshot(snapshot);
var hostEv = new HostSubscriptionEvent(entry.Id, snapshot.Resource, evt);
List>? channels;
@@ -1155,6 +1240,7 @@ private async Task PumpEventsAsync(HostEntry entry, EventStream stream)
// normal-shutdown path.
await foreach (var ev in stream.Events.ReadAllAsync(entry.LifetimeCts.Token).ConfigureAwait(false))
{
+ if (entry.CurrentClient?.GetWindowedSubscription(ev.Channel) is not null) continue;
// Update per-host observable state BEFORE broadcasting so any
// observer reading the next snapshot sees the post-event state
// (mirrors the ordering in Swift HostRuntime.handleEvent).
@@ -1858,6 +1944,11 @@ public async Task UnsubscribeAsync(
if (ReferenceEquals(client, entry.CurrentClient))
{
entry.RemoveSubscription(uri);
+ if (entry.WindowOwners.TryGetValue(uri, out var window))
+ {
+ window.End();
+ entry.WindowOwners.TryRemove(uri, out _);
+ }
return;
}
}
@@ -1893,16 +1984,44 @@ internal sealed class OpenHostAttempt
private const int Active = 0;
private const int Committing = 1;
private const int Abandoned = 2;
+ private readonly object _gate = new();
private int _state;
+ private AhpClient? _client;
- public bool TryAbandon() =>
- Interlocked.CompareExchange(ref _state, Abandoned, Active) == Active;
+ public bool TryAbandon()
+ {
+ lock (_gate)
+ {
+ if (_state != Active) return false;
+ _state = Abandoned;
+ return true;
+ }
+ }
+
+ public Task AbortClientAsync()
+ {
+ AhpClient? client;
+ lock (_gate) client = _client;
+ return client?.AbortAsync() ?? Task.CompletedTask;
+ }
+
+ public void TrackClient(AhpClient client, CancellationToken cancellationToken)
+ {
+ lock (_gate)
+ {
+ if (_state == Abandoned) throw new OperationCanceledException(cancellationToken);
+ _client = client;
+ }
+ }
public void BeginCommit(CancellationToken cancellationToken)
{
- cancellationToken.ThrowIfCancellationRequested();
- if (Interlocked.CompareExchange(ref _state, Committing, Active) != Active)
- throw new OperationCanceledException(cancellationToken);
+ lock (_gate)
+ {
+ cancellationToken.ThrowIfCancellationRequested();
+ if (_state != Active) throw new OperationCanceledException(cancellationToken);
+ _state = Committing;
+ }
}
}
}
diff --git a/clients/dotnet/src/AgentHostProtocol/Hosts/MultiHostWindowed.cs b/clients/dotnet/src/AgentHostProtocol/Hosts/MultiHostWindowed.cs
new file mode 100644
index 000000000..aaf84a580
--- /dev/null
+++ b/clients/dotnet/src/AgentHostProtocol/Hosts/MultiHostWindowed.cs
@@ -0,0 +1,101 @@
+#nullable enable
+
+using System.Threading;
+using System.Threading.Channels;
+using System.Threading.Tasks;
+
+namespace Microsoft.AgentHostProtocol.Hosts;
+
+public sealed partial class MultiHostClient
+{
+ private readonly object _flowGate = new();
+ private readonly List> _flowReaders = new();
+
+ public ChannelReader FlowControlEvents()
+ {
+ var channel = Channel.CreateBounded(
+ new BoundedChannelOptions(256) { FullMode = BoundedChannelFullMode.DropOldest });
+ lock (_flowGate) _flowReaders.Add(channel);
+ return channel.Reader;
+ }
+
+ private void BroadcastFlowEvent(FlowControlEvent value)
+ {
+ lock (_flowGate)
+ foreach (var reader in _flowReaders) reader.Writer.TryWrite(value);
+ }
+
+ private void FinishFlowEvents()
+ {
+ lock (_flowGate)
+ foreach (var reader in _flowReaders) reader.Writer.TryComplete();
+ }
+
+ public WindowedSubscription? GetWindowedSubscription(HostId host, string uri)
+ {
+ if (!_hosts.TryGetValue(host.ToString(), out var entry)) throw new UnknownHostException(host);
+ return entry.WindowOwners.TryGetValue(uri, out var owner) ? owner : null;
+ }
+
+ public async Task<(SubscribeResult Result, WindowedSubscription Sub)> SubscribeWindowedAsync(
+ HostId host, string uri, WindowOptions? options = null, CancellationToken cancellationToken = default)
+ {
+ if (!_hosts.TryGetValue(host.ToString(), out var entry)) throw new UnknownHostException(host);
+ while (true)
+ {
+ var client = await entry.WaitForClientAsync(cancellationToken).ConfigureAwait(false);
+ var result = await client.SubscribeWindowedAsync(uri, options, cancellationToken).ConfigureAwait(false);
+ await entry.ConnectionGate.WaitAsync(cancellationToken).ConfigureAwait(false);
+ try
+ {
+ if (entry.LifetimeCts.IsCancellationRequested)
+ {
+ result.Sub.End(new HostShutDownException(host), true);
+ throw new HostShutDownException(host);
+ }
+ if (ReferenceEquals(client, entry.CurrentClient))
+ {
+ lock (result.Sub.Gate)
+ if (!result.Sub.Closed)
+ {
+ result.Sub.OnClosed = entry.ForgetWindowOwner;
+ entry.RegisterWindowOwner(result.Sub);
+ }
+ return result;
+ }
+ }
+ finally { entry.ConnectionGate.Release(); }
+ }
+ }
+
+ private void ApplyManagedWindowEvent(HostEntry entry, WindowedSubscription owner,
+ JsonRpcNotification notification, SubscriptionEvent value)
+ {
+ var summaryTouched = ApplyEventToHostState(entry, value);
+ var rootTouched = value is SubscriptionEventAction rootAction
+ && owner.Uri == ProtocolVersion.RootResourceUri && entry.ApplyRootAction(rootAction.Envelope.Action);
+ if (value is SubscriptionEventAction action) entry.AdvanceServerSeq(action.Envelope.ServerSeq);
+ if (value is SubscriptionEventSnapshot snapshot)
+ {
+ entry.AdvanceServerSeq(snapshot.Snapshot.FromSeq);
+ if (owner.Uri == ProtocolVersion.RootResourceUri && snapshot.Snapshot.State.Root is { } root)
+ {
+ entry.ApplyWindowRoot(root);
+ rootTouched = true;
+ }
+ }
+ var ev = new HostSubscriptionEvent(entry.Id, owner.Uri, value);
+ List> readers;
+ lock (_subsLock) readers = new List>(_subChannels);
+ foreach (var reader in readers) reader.Writer.TryWrite(ev);
+ BroadcastPerResourceEvent(entry.Id, owner.Uri, value);
+ if (summaryTouched || rootTouched) NotifyPerHostSnapshot(entry);
+ if (summaryTouched) NotifyPerHostSummaries(entry);
+ }
+
+ private static void DropWindowOwners(HostEntry entry)
+ {
+ foreach (var pair in entry.WindowOwners) pair.Value.End(new HostShutDownException(entry.Id));
+ entry.WindowOwners.Clear();
+ }
+}
diff --git a/clients/dotnet/src/AgentHostProtocol/Telemetry.cs b/clients/dotnet/src/AgentHostProtocol/Telemetry.cs
index 2ff70674a..bb4dc8a1e 100644
--- a/clients/dotnet/src/AgentHostProtocol/Telemetry.cs
+++ b/clients/dotnet/src/AgentHostProtocol/Telemetry.cs
@@ -11,6 +11,7 @@
using System.Diagnostics;
using System.Diagnostics.Metrics;
using System.Reflection;
+using System.Collections.Concurrent;
namespace Microsoft.AgentHostProtocol;
@@ -74,4 +75,51 @@ public static class AhpTelemetry
internal static readonly Counter MalformedFrames =
Meter.CreateCounter(AhpTelemetryNames.FramesMalformed, unit: AhpTelemetryNames.FramesMalformedUnit,
description: AhpTelemetryNames.FramesMalformedDescription);
+
+ internal static readonly Counter WindowDataBytesSent = Meter.CreateCounter("ahp.client.window.data.sent", "By");
+ internal static readonly Counter WindowDataBytesReceived = Meter.CreateCounter("ahp.client.window.data.received", "By");
+ internal static readonly Counter WindowCreditBytesSent = Meter.CreateCounter("ahp.client.window.credit.sent", "By");
+ internal static readonly Counter WindowCreditBytesReceived = Meter.CreateCounter("ahp.client.window.credit.received", "By");
+ internal static readonly Histogram WindowBlockedDuration = Meter.CreateHistogram("ahp.client.window.blocked.duration", "ms");
+ private static readonly ConcurrentDictionary> Windows = new();
+ private static long _nextWindow;
+ private static readonly ObservableGauge BufferedWindowBytes =
+ Meter.CreateObservableGauge("ahp.client.window.buffered", ObserveWindowBytes, "By");
+
+ internal static long TrackWindow(WindowedSubscription subscription)
+ {
+ foreach (var pair in Windows)
+ if (!pair.Value.TryGetTarget(out _)) Windows.TryRemove(pair.Key, out _);
+ var key = System.Threading.Interlocked.Increment(ref _nextWindow);
+ Windows.TryAdd(key, new WeakReference(subscription));
+ return key;
+ }
+
+ internal static void UntrackWindow(long key) => Windows.TryRemove(key, out _);
+ internal static int TrackedWindowCount => Windows.Count;
+
+ private static IEnumerable> ObserveWindowBytes()
+ {
+ long send = 0, receive = 0;
+ foreach (var pair in Windows)
+ {
+ if (!pair.Value.TryGetTarget(out var owner) || owner.Closed)
+ {
+ Windows.TryRemove(pair.Key, out _);
+ continue;
+ }
+ var status = owner.FlowControl;
+ send += status.BufferedSendBytes;
+ receive += status.BufferedReceiveBytes;
+ }
+ yield return new Measurement(send, new KeyValuePair("direction", "send"));
+ yield return new Measurement(receive, new KeyValuePair("direction", "receive"));
+ }
+
+ internal static void FlowEvent(FlowControlEvent value)
+ {
+ using var activity = Activity.Current is null ? ActivitySource.StartActivity("ahp.flowcontrol") : null;
+ (Activity.Current ?? activity)?.AddEvent(new ActivityEvent("ahp.flowcontrol." + value.Kind,
+ tags: new ActivityTagsCollection { { "ahp.flowcontrol.bytes", value.Bytes } }));
+ }
}
diff --git a/clients/dotnet/src/AgentHostProtocol/WebSocketTransport.cs b/clients/dotnet/src/AgentHostProtocol/WebSocketTransport.cs
index 64f6474ba..41ac00b4e 100644
--- a/clients/dotnet/src/AgentHostProtocol/WebSocketTransport.cs
+++ b/clients/dotnet/src/AgentHostProtocol/WebSocketTransport.cs
@@ -38,7 +38,7 @@ public sealed class WebSocketTransportOptions
/// or to wrap
/// an existing connection.
///
-public sealed class WebSocketTransport : ITransport
+public sealed class WebSocketTransport : ITransport, ITransportMessageLimits
{
private readonly WebSocket _ws;
private readonly SemaphoreSlim _sendLock = new(1, 1);
@@ -46,6 +46,8 @@ public sealed class WebSocketTransport : ITransport
private int _disposed;
internal long MaxMessageBytes => _maxMessageBytes;
+ /// Complete inbound WebSocket message cap, not the receive-buffer size.
+ public long? MaximumInboundMessageBytes => _maxMessageBytes > 0 ? _maxMessageBytes : null;
// Receive buffer: 64 KiB initial, grows as needed.
private byte[] _receiveBuffer = new byte[64 * 1024];
diff --git a/clients/dotnet/src/AgentHostProtocol/WindowOptions.cs b/clients/dotnet/src/AgentHostProtocol/WindowOptions.cs
new file mode 100644
index 000000000..c850ea7f3
--- /dev/null
+++ b/clients/dotnet/src/AgentHostProtocol/WindowOptions.cs
@@ -0,0 +1,135 @@
+#nullable enable
+
+using System.Threading;
+using System.Threading.Tasks;
+
+namespace Microsoft.AgentHostProtocol;
+
+/// Connection defaults; configuring them does not enable windowing.
+public sealed class WindowingConfig
+{
+ public long WindowBytes { get; set; } = 262144;
+ /// UTF-16 bytes; null derives a safe limit from the transport.
+ public long? MaximumFrameBytes { get; set; }
+ public long MaxBufferedSendBytes { get; set; } = 1048576;
+ public IList Coalescers { get; } = new List();
+
+ internal WindowingConfig Snapshot()
+ {
+ Validate(WindowBytes, 1, nameof(WindowBytes));
+ if (MaximumFrameBytes is { } maximum) Validate(maximum, 1024, nameof(MaximumFrameBytes));
+ Validate(MaxBufferedSendBytes, 1, nameof(MaxBufferedSendBytes));
+ var copy = new WindowingConfig
+ {
+ WindowBytes = WindowBytes,
+ MaximumFrameBytes = MaximumFrameBytes,
+ MaxBufferedSendBytes = MaxBufferedSendBytes,
+ };
+ foreach (var coalescer in Coalescers)
+ copy.Coalescers.Add(coalescer ?? throw new ArgumentException("A coalescer cannot be null."));
+ return copy;
+ }
+
+ internal static void Validate(long value, long minimum, string name)
+ {
+ if (value < minimum || value > 9007199254740991)
+ throw new ArgumentOutOfRangeException(name);
+ }
+}
+
+public enum ReceiveMode { Auto, Manual }
+
+/// Per-subscription opt-in; unspecified limits inherit connection defaults.
+public sealed class WindowOptions
+{
+ public long? WindowBytes { get; set; }
+ public long? MaximumFrameBytes { get; set; }
+ public ReceiveMode Receive { get; set; } = ReceiveMode.Auto;
+ public bool Coalescing { get; set; } = true;
+
+ internal WindowOptions Snapshot() => new()
+ {
+ WindowBytes = WindowBytes,
+ MaximumFrameBytes = MaximumFrameBytes,
+ Receive = Receive,
+ Coalescing = Coalescing,
+ };
+}
+
+/// Additive client capability; existing IAhpClient implementations need not implement it.
+public interface IWindowedAhpClient : IAhpClient
+{
+ Task<(SubscribeResult Result, WindowedSubscription Sub)> SubscribeWindowedAsync(
+ string uri, WindowOptions? options = null, CancellationToken cancellationToken = default);
+ WindowedSubscription? GetWindowedSubscription(string uri);
+ Task ResumeAsync(ReconnectParams parameters, CancellationToken cancellationToken = default);
+ System.Threading.Channels.ChannelReader FlowControlEvents();
+}
+
+public static class WindowedClientExtensions
+{
+ private static IWindowedAhpClient Capability(IAhpClient client) => client as IWindowedAhpClient
+ ?? throw new NotSupportedException("This client does not implement windowed delivery.");
+
+ public static Task<(SubscribeResult Result, WindowedSubscription Sub)> SubscribeWindowedAsync(
+ this IAhpClient client, string uri, WindowOptions? options = null,
+ CancellationToken cancellationToken = default) =>
+ client is IWindowedAhpClient windowed
+ ? windowed.SubscribeWindowedAsync(uri, options, cancellationToken)
+ : throw new NotSupportedException("This client does not implement windowed delivery.");
+
+ public static WindowedSubscription? GetWindowedSubscription(this IAhpClient client, string uri) =>
+ Capability(client).GetWindowedSubscription(uri);
+
+ public static Task ResumeAsync(this IAhpClient client, ReconnectParams parameters,
+ CancellationToken cancellationToken = default) => Capability(client).ResumeAsync(parameters, cancellationToken);
+
+ public static System.Threading.Channels.ChannelReader FlowControlEvents(this IAhpClient client) =>
+ Capability(client).FlowControlEvents();
+}
+
+public enum CoalescingDirection { Incoming, Outgoing }
+
+public sealed class CoalescingContext
+{
+ public string Channel { get; }
+ public CoalescingDirection Direction { get; }
+ internal CoalescingContext(string channel, CoalescingDirection direction)
+ {
+ Channel = channel;
+ Direction = direction;
+ }
+}
+
+/// Combines adjacent queued notifications; null means they cannot be combined.
+public interface IChannelMessageCoalescer
+{
+ JsonRpcNotification? Coalesce(JsonRpcNotification earlier, JsonRpcNotification later, CoalescingContext context);
+}
+
+public sealed class FlowControlStatus
+{
+ public string Mode { get; internal init; } = "ordinary";
+ public string? FallbackReason { get; internal init; }
+ public ChannelFlowControl? Limits { get; internal init; }
+ public long OutstandingSendBytes { get; internal init; }
+ public long BufferedSendBytes { get; internal init; }
+ public long BufferedReceiveBytes { get; internal init; }
+ public bool Writable { get; internal init; }
+}
+
+public sealed class FlowControlEvent
+{
+ public string Channel { get; }
+ public string Kind { get; }
+ public long Bytes { get; }
+ public FlowControlStatus Status { get; }
+ public Hosts.HostId? HostId { get; internal init; }
+ internal FlowControlEvent(string channel, string kind, FlowControlStatus status, long bytes = 0)
+ {
+ Channel = channel;
+ Kind = kind;
+ Status = status;
+ Bytes = bytes;
+ }
+}
diff --git a/clients/dotnet/src/AgentHostProtocol/WindowedDeliveryEngine.cs b/clients/dotnet/src/AgentHostProtocol/WindowedDeliveryEngine.cs
new file mode 100644
index 000000000..a00664c59
--- /dev/null
+++ b/clients/dotnet/src/AgentHostProtocol/WindowedDeliveryEngine.cs
@@ -0,0 +1,981 @@
+#nullable enable
+
+using System.Collections.Concurrent;
+using System.Linq;
+using System.Text;
+using System.Threading;
+using System.Threading.Channels;
+using System.Threading.Tasks;
+
+namespace Microsoft.AgentHostProtocol;
+
+public sealed partial class AhpClient
+{
+ private sealed class WindowResponse
+ {
+ internal readonly IReadOnlyList Owners;
+ internal Action Apply = null!;
+ internal readonly Action Cleanup;
+ private int _phase;
+ internal bool Abandoned;
+ internal Exception? Failure;
+
+ internal WindowResponse(IReadOnlyList owners, Action cleanup)
+ {
+ Owners = owners;
+ Cleanup = cleanup;
+ }
+
+ internal bool TryStart() => Interlocked.CompareExchange(ref _phase, 1, 0) == 0;
+ internal bool Abandon(Exception error)
+ {
+ Failure = error;
+ Abandoned = true;
+ var unsent = Interlocked.CompareExchange(ref _phase, 2, 0) == 0;
+ Cleanup(error);
+ return unsent;
+ }
+ }
+
+ private readonly object _windowGate = new();
+ private readonly Dictionary _windows = new(StringComparer.Ordinal);
+ private readonly ConcurrentDictionary _windowResponses = new();
+ private readonly LinkedList _closedWindowOrder = new();
+ private readonly Dictionary> _closedWindows = new(StringComparer.Ordinal);
+ private readonly System.Runtime.CompilerServices.ConditionalWeakTable _windowRequestOptions = new();
+ private readonly List> _flowEventReaders = new();
+ private readonly WindowWakeSignal _writerWake = new();
+ private int _windowTurn;
+ private WindowSend? _windowInFlight;
+ private bool RequiresWindowSizes
+ {
+ get { lock (_windowGate) return _windows.Values.Any(owner => owner.Limits is not null); }
+ }
+ internal WindowingConfig WindowConfig => _cfg.Windowing;
+ internal JsonElement WindowParameters(T parameters) => _serializer.SerializeToElement(parameters);
+ internal Action? ManagedWindowEvent { get; set; }
+ internal bool RetainManagedWindows { get; set; }
+ internal Hosts.HostId? WindowHostId { get; set; }
+ internal Action? ManagedFlowEvent { get; set; }
+
+ public WindowedSubscription? GetWindowedSubscription(string uri)
+ {
+ lock (_windowGate) return _windows.TryGetValue(uri, out var owner) ? owner : null;
+ }
+
+ public ChannelReader FlowControlEvents()
+ {
+ var channel = Channel.CreateBounded(
+ new BoundedChannelOptions(256) { FullMode = BoundedChannelFullMode.DropOldest });
+ lock (_windowGate) _flowEventReaders.Add(channel);
+ return channel.Reader;
+ }
+
+ internal void WindowEvent(WindowedSubscription owner, string kind, long bytes = 0)
+ {
+ var value = new FlowControlEvent(owner.Uri, kind, owner.FlowControl, bytes) { HostId = owner.HostId };
+ lock (_windowGate)
+ foreach (var reader in _flowEventReaders) reader.Writer.TryWrite(value);
+ AhpTelemetry.FlowEvent(value);
+ ManagedFlowEvent?.Invoke(value);
+ }
+
+ private ChannelReceiveLimits WindowOffer(WindowOptions options)
+ {
+ if (options.Receive is not ReceiveMode.Auto and not ReceiveMode.Manual)
+ throw new ArgumentOutOfRangeException(nameof(options));
+ var transportDefault = _transport is ITransportMessageLimits { MaximumInboundMessageBytes: > 0 } limited
+ ? 2 * (limited.MaximumInboundMessageBytes!.Value / 3)
+ : 1048576;
+ var result = new ChannelReceiveLimits
+ {
+ WindowBytes = options.WindowBytes ?? WindowConfig.WindowBytes,
+ MaximumFrameBytes = options.MaximumFrameBytes ?? WindowConfig.MaximumFrameBytes ?? transportDefault,
+ };
+ WindowingConfig.Validate(result.WindowBytes, 1, nameof(options.WindowBytes));
+ WindowingConfig.Validate(result.MaximumFrameBytes, 1024, nameof(options.MaximumFrameBytes));
+ return result;
+ }
+
+ public async Task<(SubscribeResult Result, WindowedSubscription Sub)> SubscribeWindowedAsync(
+ string uri, WindowOptions? options = null, CancellationToken cancellationToken = default)
+ {
+ Guard.ThrowIfNull(uri, nameof(uri));
+ cancellationToken.ThrowIfCancellationRequested();
+ var copy = (options ?? new WindowOptions()).Snapshot();
+ var offer = WindowOffer(copy);
+ WindowedSubscription owner;
+ lock (_windowGate)
+ {
+ if (_windows.TryGetValue(uri, out var closing) && closing.Closed)
+ throw new InvalidOperationException("The subscription is closing.");
+ if (!_windows.TryGetValue(uri, out owner!))
+ {
+ owner = new WindowedSubscription(this, uri, copy, offer);
+ _windows.Add(uri, owner);
+ }
+ }
+ var request = new SubscribeParams
+ {
+ Channel = uri,
+ FlowControl = new SubscriptionFlowControlOptions { Receive = offer },
+ };
+ _windowRequestOptions.Add(request, copy);
+ SubscribeResult result;
+ try
+ {
+ result = await RequestAsync("subscribe", request, cancellationToken).ConfigureAwait(false)
+ ?? throw Protocol("subscribe returned no result");
+ }
+ catch (Exception error)
+ {
+ CleanupUnacceptedWindow(owner, error);
+ throw;
+ }
+ finally { _windowRequestOptions.Remove(request); }
+ return (result, owner);
+ }
+
+ public async Task ResumeAsync(ReconnectParams parameters, CancellationToken cancellationToken = default)
+ {
+ try
+ {
+ var result = await RequestAsync("reconnect", parameters, cancellationToken).ConfigureAwait(false)
+ ?? throw Protocol("reconnect returned no result");
+ AhpTelemetry.Reconnects.Add(1, new KeyValuePair(AhpTelemetryNames.AttrOutcome, AhpTelemetryNames.OutcomeOk));
+ return result;
+ }
+ catch
+ {
+ AhpTelemetry.Reconnects.Add(1, new KeyValuePair(AhpTelemetryNames.AttrOutcome, AhpTelemetryNames.OutcomeError));
+ throw;
+ }
+ }
+
+ internal bool BindManagedWindow(WindowedSubscription owner)
+ {
+ lock (owner.Gate)
+ {
+ if (owner.Closed) return false;
+ owner.Client = this;
+ owner.HostId = WindowHostId;
+ owner.Epoch++;
+ owner.Closed = false;
+ owner.CloseError = null;
+ owner.Negotiating = true;
+ owner.Limits = null;
+ owner.Received = owner.Released = owner.CreditSent = owner.Sent = owner.PeerConsumed = 0;
+ owner.CreditTimer = false;
+ owner.BufferedReceive -= owner.FragmentBytes;
+ owner.Fragments = null;
+ owner.FragmentBytes = 0;
+ owner.Offer = WindowOffer(owner.Options);
+ }
+ lock (_windowGate) _windows[owner.Uri] = owner;
+ return true;
+ }
+
+ private void PrepareWindowResponse(ulong id, string method, JsonElement? parameters, WindowOptions? requestedOptions,
+ CancellationToken cancellationToken)
+ {
+ if (parameters is not { } p) return;
+ if (method == "subscribe" && p.TryGetProperty("channel", out var channelEl) && channelEl.GetString() is { } uri)
+ {
+ var owner = GetWindowedSubscription(uri);
+ ChannelReceiveLimits? offer = null;
+ if (p.TryGetProperty("flowControl", out var flow))
+ {
+ offer = _serializer.Deserialize(flow).Receive;
+ WindowingConfig.Validate(offer.WindowBytes, 1, "windowBytes");
+ WindowingConfig.Validate(offer.MaximumFrameBytes, 1024, "maximumFrameBytes");
+ if (owner is null)
+ {
+ owner = new WindowedSubscription(this, uri, new WindowOptions(), offer);
+ lock (_windowGate) _windows[uri] = owner;
+ }
+ }
+ if (owner is null) return;
+ var target = owner;
+ var pendingOptions = requestedOptions;
+ var initial = owner.Negotiating;
+ var response = new WindowResponse(new[] { target }, error => CleanupUnacceptedWindow(target, error));
+ response.Apply = (result, error) =>
+ {
+ if (result is null)
+ {
+ response.Cleanup(error ?? new AhpClientClosedException("Subscribe failed."));
+ return;
+ }
+ if (target.Closed || !ReferenceEquals(GetWindowedSubscription(uri), target))
+ {
+ if (GetWindowedSubscription(uri) is null) _ = UnsubscribeOwnerAsync(target);
+ return;
+ }
+ var decoded = _serializer.Deserialize(result.Value);
+ if (decoded.FlowControl is not null && decoded.Snapshot is not null)
+ throw Protocol("windowed subscribe returned a snapshot");
+ InstallWindow(target, decoded.FlowControl, offer, true);
+ lock (target.Gate)
+ {
+ if (pendingOptions is not null) target.Options = pendingOptions;
+ }
+ if (decoded.Snapshot is { } snapshot)
+ ReceiveWindowMessage(target, SnapshotNotification(snapshot), 0, 0);
+ if (initial && (!_pending.ContainsKey(id) || cancellationToken.IsCancellationRequested))
+ CloseWindowOwner(target);
+ };
+ _windowResponses[id] = response;
+ }
+ else if (method == "reconnect")
+ {
+ var windows = p.TryGetProperty("windows", out var windowsEl)
+ ? _serializer.Deserialize(windowsEl) : null;
+ var requested = new Dictionary(StringComparer.Ordinal);
+ foreach (var item in windows?.Items ?? new List())
+ {
+ if (item.Channel.StartsWith("ahp-tcp:", StringComparison.Ordinal))
+ throw Protocol("TCP subscriptions cannot reconnect");
+ if (!p.GetProperty("subscriptions").EnumerateArray().Any(x => x.GetString() == item.Channel)
+ || requested.ContainsKey(item.Channel))
+ throw Protocol("invalid reconnect window subscriptions");
+ var owner = GetWindowedSubscription(item.Channel);
+ if (owner is null)
+ {
+ owner = new WindowedSubscription(this, item.Channel, new WindowOptions(), item.FlowControl.Receive);
+ lock (_windowGate) _windows[item.Channel] = owner;
+ }
+ requested.Add(item.Channel, (owner, item.FlowControl.Receive));
+ }
+ List ordinary;
+ lock (_windowGate) ordinary = _windows.Values.Where(x => !x.ResumeWindowed && x.EverAccepted && !x.Closed).ToList();
+ if (requested.Count == 0 && ordinary.Count == 0) return;
+ var targets = requested.Values.Select(x => x.Owner).Concat(ordinary).ToList();
+ var response = new WindowResponse(targets, error =>
+ {
+ foreach (var target in targets) CleanupUnacceptedWindow(target, error);
+ });
+ response.Apply = (result, error) =>
+ {
+ if (result is null)
+ {
+ response.Cleanup(error ?? new AhpClientClosedException("Reconnect failed."));
+ return;
+ }
+ var retained = new HashSet(StringComparer.Ordinal);
+ if (result.Value.TryGetProperty("windows", out var returned))
+ {
+ foreach (var item in _serializer.Deserialize(returned).Items)
+ {
+ if (!requested.TryGetValue(item.Channel, out var state) || !retained.Add(item.Channel))
+ throw Protocol("unsolicited reconnect window");
+ if (state.Owner.Closed || !ReferenceEquals(GetWindowedSubscription(item.Channel), state.Owner))
+ {
+ if (GetWindowedSubscription(item.Channel) is null) _ = UnsubscribeOwnerAsync(state.Owner);
+ continue;
+ }
+ InstallWindow(state.Owner, item.FlowControl, state.Offer, item.Recovery == ChannelRecoveryKind.Snapshot);
+ WindowEvent(state.Owner, "resumed");
+ }
+ }
+ foreach (var item in requested)
+ if (!retained.Contains(item.Key))
+ {
+ item.Value.Owner.End(Protocol("subscription was not retained"), true);
+ WindowEvent(item.Value.Owner, "dropped");
+ RemoveWindow(item.Value.Owner);
+ }
+ RecoverOrdinaryWindows(_serializer.Deserialize(result.Value));
+ };
+ _windowResponses[id] = response;
+ }
+ }
+
+ private void InstallWindow(WindowedSubscription owner, ChannelFlowControl? limits, ChannelReceiveLimits? offer, bool snapshot)
+ {
+ if (limits is not null)
+ {
+ if (offer is null) throw Protocol("unsolicited window acceptance");
+ ValidateLimits(limits.ClientReceive);
+ ValidateLimits(limits.HostReceive);
+ if (limits.ClientReceive.WindowBytes > offer.WindowBytes || limits.ClientReceive.MaximumFrameBytes > offer.MaximumFrameBytes)
+ throw Protocol("accepted limits exceed the offer");
+ }
+ else if (owner.Uri.StartsWith("ahp-tcp:", StringComparison.Ordinal))
+ throw Protocol("TCP requires windowed delivery");
+ lock (owner.Gate)
+ {
+ if (owner.Limits is null || limits is null)
+ {
+ owner.Epoch++;
+ owner.Sent = owner.PeerConsumed = owner.Received = owner.Released = owner.CreditSent = 0;
+ owner.Fragments = null;
+ owner.FragmentBytes = 0;
+ }
+ ForgetClosedWindow(owner.Uri);
+ owner.Offer = offer ?? owner.Offer;
+ owner.Limits = limits;
+ owner.ResumeWindowed = limits is not null;
+ owner.Fallback = limits is null && offer is not null ? "declined" : null;
+ owner.Negotiating = false;
+ owner.EverAccepted = true;
+ owner.AwaitingSnapshot = limits is not null && snapshot;
+ }
+ owner.Publish();
+ WindowEvent(owner, limits is null && offer is not null ? "declined" : "negotiated");
+ _writerWake.Release();
+ }
+
+ private static void ValidateLimits(ChannelReceiveLimits limits)
+ {
+ if (limits.WindowBytes < 1 || limits.MaximumFrameBytes < 1024
+ || limits.WindowBytes > 9007199254740991 || limits.MaximumFrameBytes > 9007199254740991)
+ throw Protocol("invalid window limits");
+ }
+
+ internal static long RawWindowBytes(TransportMessage message) =>
+ checked((long)(message.Text ?? new UTF8Encoding(false, true).GetString(message.Binary.ToArray())).Length * 2);
+ private static AhpTransportException Protocol(string message) => new("protocol", "ahp: " + message);
+
+ private JsonRpcNotification SnapshotNotification(Snapshot snapshot) => new()
+ {
+ Method = "channel/snapshot",
+ Params = _serializer.SerializeToElement(new ChannelSnapshotParams { Channel = snapshot.Resource, Snapshot = snapshot }),
+ };
+
+ private bool HandleWindowNotification(JsonRpcNotification notification, long bytes)
+ {
+ if (notification.Params is not { } parameters
+ || !parameters.TryGetProperty("channel", out var channel)
+ || channel.GetString() is not { } uri) return false;
+ var owner = GetWindowedSubscription(uri);
+ if (owner is null)
+ {
+ if (IsClosedWindow(uri)) return true;
+ if (notification.Method is "channel/frame" or "channel/snapshot" or "channel/credit")
+ throw Protocol("window message without a subscription");
+ return false;
+ }
+ if (owner.Closed) return true;
+ if (owner.Negotiating && notification.Method is not "channel/credit")
+ {
+ if (IsClosedWindow(uri)) return true;
+ if (notification.Method is "channel/frame" or "channel/snapshot" || owner.ResumeWindowed)
+ throw Protocol("data before negotiation completed");
+ return false;
+ }
+ if (notification.Method == "channel/credit")
+ {
+ var consumed = _serializer.Deserialize(parameters).ConsumedBytes;
+ lock (owner.Gate)
+ {
+ if (owner.Limits is null) throw Protocol("credit on an ordinary subscription");
+ if (consumed < 0 || consumed > owner.Sent) throw Protocol("credit exceeds sent bytes");
+ if (consumed > owner.PeerConsumed) owner.PeerConsumed = consumed;
+ }
+ AhpTelemetry.WindowCreditBytesReceived.Add(bytes);
+ WindowEvent(owner, "creditReceived", bytes);
+ owner.Publish();
+ _writerWake.Release();
+ return true;
+ }
+ var counted = owner.Limits is not null && WindowedDelivery.CountsAgainstWindow(notification.Method);
+ if (counted)
+ {
+ if (bytes > owner.Limits!.ClientReceive.MaximumFrameBytes) throw Protocol("data exceeds maximumFrameBytes");
+ lock (owner.Gate)
+ {
+ owner.Received = checked(owner.Received + bytes);
+ owner.BufferedReceive = checked(owner.BufferedReceive + bytes);
+ }
+ AhpTelemetry.WindowDataBytesReceived.Add(bytes);
+ }
+ if (notification.Method == "channel/frame")
+ {
+ if (!counted) throw Protocol("fragment on an ordinary subscription");
+ var fragment = _serializer.Deserialize(parameters);
+ if (fragment.Data.Length == 0 || char.IsLowSurrogate(fragment.Data[0]) || char.IsHighSurrogate(fragment.Data[fragment.Data.Length - 1]))
+ throw Protocol("invalid fragment boundary");
+ string? joined = null;
+ long total;
+ lock (owner.Gate)
+ {
+ owner.Fragments ??= new StringBuilder();
+ owner.Fragments.Append(fragment.Data);
+ owner.FragmentBytes = checked(owner.FragmentBytes + bytes);
+ total = owner.FragmentBytes;
+ if (fragment.Final == true)
+ {
+ joined = owner.Fragments.ToString();
+ owner.Fragments = null;
+ owner.FragmentBytes = 0;
+ }
+ }
+ if (joined is null)
+ ReleaseWindowBytes(owner, bytes, owner.Epoch, true);
+ else
+ {
+ WindowEvent(owner, "fragmentedMessageReceived", total);
+ var message = _serializer.DecodeMessage(TransportMessage.FromText(joined));
+ var inner = message.Notification ?? throw Protocol("fragment did not contain a notification");
+ if (inner.Method == "channel/frame" || !WindowedDelivery.CountsAgainstWindow(inner.Method)
+ || inner.Params?.GetProperty("channel").GetString() != uri)
+ throw Protocol("invalid joined fragment channel or method");
+ ReceiveWindowMessage(owner, inner, total, bytes);
+ }
+ return true;
+ }
+ if (counted && owner.Fragments is not null) throw Protocol("data interleaved with fragments");
+ if (notification.Method == "tcp/reset")
+ {
+ lock (owner.Gate)
+ {
+ owner.Fragments = null;
+ foreach (var send in owner.Sends) send.Fail(Protocol("TCP reset"));
+ owner.Sends.Clear();
+ owner.BufferedSend = 0;
+ foreach (var delivery in owner.Queue) owner.BufferedReceive -= delivery.WireBytes;
+ owner.Queue.Clear();
+ owner.BufferedReceive -= owner.FragmentBytes;
+ owner.FragmentBytes = 0;
+ }
+ }
+ if (notification.Method == "tcp/data")
+ {
+ if (owner.ReceiveEof) throw Protocol("TCP data after EOF");
+ ValidateTcpData(parameters.GetProperty("data").GetString()!);
+ }
+ if (notification.Method == "tcp/eof") owner.ReceiveEof = true;
+ ReceiveWindowMessage(owner, notification, counted ? bytes : 0, counted ? bytes : 0);
+ if (notification.Method == "tcp/reset")
+ {
+ owner.End();
+ _ = UnsubscribeOwnerAsync(owner);
+ }
+ return true;
+ }
+
+ internal void RecoverOrdinaryWindows(ReconnectResult result)
+ {
+ List owners;
+ lock (_windowGate) owners = _windows.Values.Where(x => !x.ResumeWindowed && !x.Closed).ToList();
+ foreach (var owner in owners)
+ {
+ owner.Negotiating = false;
+ owner.Publish();
+ if (result.Value is ReconnectReplayResult replay)
+ foreach (var action in replay.Actions.Where(x => x.Channel == owner.Uri))
+ ReceiveWindowMessage(owner, new JsonRpcNotification
+ {
+ Method = "action",
+ Params = _serializer.SerializeToElement(action),
+ }, 0, 0);
+ if (result.Value is ReconnectSnapshotResult snapshot)
+ foreach (var state in snapshot.Snapshots.Where(x => x.Resource == owner.Uri))
+ ReceiveWindowMessage(owner, SnapshotNotification(state), 0, 0);
+ }
+ }
+
+ private SubscriptionEvent WindowEventValue(JsonRpcNotification notification)
+ {
+ var p = notification.Params ?? throw Protocol("notification has no params");
+ return notification.Method switch
+ {
+ "action" => new SubscriptionEventAction(_serializer.Deserialize(p)),
+ "channel/snapshot" => new SubscriptionEventSnapshot(_serializer.Deserialize(p).Snapshot),
+ "tcp/data" => new SubscriptionEventTcpData(_serializer.Deserialize(p)),
+ "tcp/eof" => new SubscriptionEventTcpEof(_serializer.Deserialize(p)),
+ "tcp/reset" => new SubscriptionEventTcpReset(_serializer.Deserialize(p)),
+ "root/sessionAdded" => new SubscriptionEventSessionAdded(_serializer.Deserialize(p)),
+ "root/sessionRemoved" => new SubscriptionEventSessionRemoved(_serializer.Deserialize(p)),
+ "root/sessionSummaryChanged" => new SubscriptionEventSessionSummaryChanged(_serializer.Deserialize(p)),
+ "root/progress" => new SubscriptionEventProgress(_serializer.Deserialize(p)),
+ "auth/required" => new SubscriptionEventAuthRequired(_serializer.Deserialize(p)),
+ _ => new SubscriptionEventNotification(notification),
+ };
+ }
+
+ [System.Diagnostics.CodeAnalysis.SuppressMessage("Reliability", "CA2000:Dispose objects before losing scope",
+ Justification = "The reliable queue transfers delivery ownership to its consumer; coalescing replaces leases without releasing their credit.")]
+ private void ReceiveWindowMessage(WindowedSubscription owner, JsonRpcNotification notification, long bytes, long credit, bool observe = true)
+ {
+ if (owner.AwaitingSnapshot && WindowedDelivery.CountsAgainstWindow(notification.Method))
+ {
+ if (notification.Method != "channel/snapshot") throw Protocol("first data is not a snapshot");
+ owner.AwaitingSnapshot = false;
+ }
+
+ if (notification.Method == "channel/snapshot"
+ && notification.Params!.Value.GetProperty("snapshot").GetProperty("resource").GetString() != owner.Uri)
+ throw Protocol("snapshot resource mismatch");
+ var value = WindowEventValue(notification);
+ var epoch = owner.Epoch;
+ var delivery = new ReceivedDelivery(value, notification, bytes, credit, epoch,
+ () => owner.Client.ReleaseDelivery(owner, bytes, credit, epoch));
+ lock (owner.Gate)
+ {
+ if (value is SubscriptionEventAction action) owner.LastSeq = Math.Max(owner.LastSeq, action.Envelope.ServerSeq);
+ if (value is SubscriptionEventSnapshot snapshot) owner.LastSeq = Math.Max(owner.LastSeq, snapshot.Snapshot.FromSeq);
+ if (owner.Limits is not null && owner.Options.Coalescing && owner.Queue.Last is { } tail && tail.Value.Epoch == epoch)
+ {
+ var merged = Coalesce(tail.Value.Notification, notification, owner, CoalescingDirection.Incoming);
+ if (merged is not null)
+ {
+ var total = checked(bytes + tail.Value.WireBytes);
+ var owed = checked(credit + tail.Value.CreditBytes);
+ delivery = new ReceivedDelivery(WindowEventValue(merged), merged, total, owed, epoch,
+ () => owner.Client.ReleaseDelivery(owner, total, owed, epoch));
+ owner.Queue.RemoveLast();
+ WindowEvent(owner, "coalesced", total);
+ }
+ }
+ if (owner.Limits is null && owner.Queue.Count >= _cfg.SubscriptionBufferCapacity)
+ owner.Queue.RemoveFirst();
+ owner.Queue.AddLast(delivery);
+ owner.Available.TrySetResult(true);
+ owner.Available = WindowedSubscription.Signal();
+ }
+ if (observe)
+ {
+ FanOut(owner.Uri, value);
+ ManagedWindowEvent?.Invoke(owner, notification, value);
+ }
+ owner.Publish();
+ }
+
+ private void ReleaseWindowBytes(WindowedSubscription owner, long bytes, long epoch, bool immediate = false)
+ {
+ if (bytes == 0) return;
+ bool flush;
+ lock (owner.Gate)
+ {
+ if (owner.Closed || owner.Epoch != epoch || owner.Limits is null || owner.Negotiating) return;
+ owner.Released = checked(owner.Released + bytes);
+ var threshold = Math.Max(1, Math.Min(16384, owner.Limits.ClientReceive.WindowBytes / 4));
+ flush = immediate || owner.Released - owner.CreditSent >= threshold
+ || owner.Received - owner.CreditSent >= owner.Limits.ClientReceive.WindowBytes;
+ if (!flush && !owner.CreditTimer)
+ {
+ owner.CreditTimer = true;
+ _ = FlushCreditLaterAsync(owner, epoch);
+ }
+
+ }
+ if (flush) _writerWake.Release();
+ owner.Publish();
+ }
+
+ private void ReleaseDelivery(WindowedSubscription owner, long bytes, long credit, long epoch)
+ {
+ lock (owner.Gate) owner.BufferedReceive -= bytes;
+ ReleaseWindowBytes(owner, credit, epoch);
+ owner.Publish();
+ }
+
+ private async Task FlushCreditLaterAsync(WindowedSubscription owner, long epoch)
+ {
+ using var cancellation = CancellationTokenSource.CreateLinkedTokenSource(_lifetimeCts.Token);
+ lock (owner.Gate)
+ {
+ if (owner.Closed || owner.Epoch != epoch) return;
+ owner.CreditCancellation = cancellation;
+ }
+ try
+ {
+ await TimeProviderCompatibility.DelayAsync(_cfg.TimeProvider, TimeSpan.FromMilliseconds(5), cancellation.Token).ConfigureAwait(false);
+ _writerWake.Release();
+ }
+ catch (OperationCanceledException) when (cancellation.IsCancellationRequested) { }
+ finally
+ {
+ lock (owner.Gate)
+ if (ReferenceEquals(owner.CreditCancellation, cancellation))
+ {
+ owner.CreditCancellation = null;
+ owner.CreditTimer = false;
+ }
+ }
+ }
+
+ private string EncodeWindow(JsonRpcNotification notification) =>
+ _serializer.EncodeMessage(new JsonRpcMessage { Notification = notification }).Text
+ ?? throw Protocol("serializer must encode text");
+
+ private JsonRpcNotification? Coalesce(JsonRpcNotification older, JsonRpcNotification newer,
+ WindowedSubscription owner, CoalescingDirection direction)
+ {
+ if (older.Method != newer.Method || !WindowedDelivery.CountsAgainstWindow(older.Method)
+ || older.Method is "channel/snapshot" or "dispatchAction" or "tcp/eof"
+ || older.Params?.TryGetProperty("origin", out _) == true || newer.Params?.TryGetProperty("origin", out _) == true)
+ return null;
+ foreach (var coalescer in WindowConfig.Coalescers)
+ {
+ var combined = coalescer.Coalesce(older, newer, new CoalescingContext(owner.Uri, direction));
+ if (combined is null) continue;
+ if (combined.Method != older.Method || combined.Params?.GetProperty("channel").GetString() != owner.Uri)
+ throw Protocol("custom coalescer changed method or channel");
+ return combined;
+ }
+ if (direction == CoalescingDirection.Incoming && older.Method == "action")
+ {
+ var first = _serializer.Deserialize(older.Params!.Value);
+ var last = _serializer.Deserialize(newer.Params!.Value);
+ if (first.RejectionReason is not null || last.RejectionReason is not null) return null;
+ StateAction? action = null;
+ if (first.Action.Value is TerminalDataAction a && last.Action.Value is TerminalDataAction b)
+ action = new StateAction(b with { Data = a.Data + b.Data });
+ if (first.Action.Value is ChatDeltaAction c && last.Action.Value is ChatDeltaAction d
+ && c.TurnId == d.TurnId && c.PartId == d.PartId
+ && MetadataEqual(older.Params.Value.GetProperty("action"), newer.Params.Value.GetProperty("action")))
+ action = new StateAction(d with { Content = c.Content + d.Content });
+ return action is null ? null : new JsonRpcNotification
+ {
+ Method = "action",
+ Params = _serializer.SerializeToElement(last with { Action = action }),
+ };
+ }
+ if (direction == CoalescingDirection.Outgoing && older.Method == "tcp/data")
+ {
+ var first = Convert.FromBase64String(older.Params!.Value.GetProperty("data").GetString()!);
+ var last = Convert.FromBase64String(newer.Params!.Value.GetProperty("data").GetString()!);
+ var combined = new byte[first.Length + last.Length];
+ first.CopyTo(combined, 0);
+ last.CopyTo(combined, first.Length);
+ return new JsonRpcNotification
+ {
+ Method = "tcp/data",
+ Params = _serializer.SerializeToElement(new TcpDataParams
+ {
+ Channel = owner.Uri,
+ Data = Convert.ToBase64String(combined),
+ }),
+ };
+ }
+ return null;
+ }
+
+ private static bool MetadataEqual(JsonElement first, JsonElement last)
+ {
+ var hasFirst = first.TryGetProperty("_meta", out var a);
+ var hasLast = last.TryGetProperty("_meta", out var b);
+ return hasFirst == hasLast && (!hasFirst || JsonEqual(a, b));
+ }
+
+ internal static bool JsonEqual(JsonElement a, JsonElement b)
+ {
+ if (a.ValueKind != b.ValueKind) return false;
+ if (a.ValueKind == JsonValueKind.Object)
+ return a.EnumerateObject().Count() == b.EnumerateObject().Count()
+ && a.EnumerateObject().All(p => b.TryGetProperty(p.Name, out var value) && JsonEqual(p.Value, value));
+ if (a.ValueKind == JsonValueKind.Array)
+ return a.GetArrayLength() == b.GetArrayLength() && a.EnumerateArray().Zip(b.EnumerateArray(), JsonEqual).All(x => x);
+ return a.ToString() == b.ToString();
+ }
+
+ private async Task SendWindowAsync(WindowedSubscription owner, JsonRpcNotification notification, CancellationToken token)
+ {
+ var text = EncodeWindow(notification);
+ var completion = WindowedSubscription.Signal();
+ var item = new WindowSend(notification, text, completion, token);
+ while (true)
+ {
+ Task wait;
+ token.ThrowIfCancellationRequested();
+ if (_lifetimeCts.IsCancellationRequested) throw new AhpClientClosedException();
+ lock (owner.Gate)
+ {
+ if (owner.Closed) throw owner.CloseError ?? new AhpClientClosedException();
+ if (owner.BufferedSend == 0 || owner.BufferedSend + item.Bytes <= WindowConfig.MaxBufferedSendBytes)
+ {
+ if (notification.Method == "tcp/data")
+ {
+ if (owner.SendEof) throw new InvalidOperationException("TCP data cannot follow EOF.");
+ ValidateTcpData(notification.Params!.Value.GetProperty("data").GetString()!);
+ }
+ if (notification.Method == "tcp/eof") owner.SendEof = true;
+ if (owner.Options.Coalescing && owner.Sends.Last is { } tail && !tail.Value.Started
+ && tail.Value.Waiters.All(x => x.Token == token))
+ {
+ var combined = Coalesce(tail.Value.Notification, notification, owner, CoalescingDirection.Outgoing);
+ if (combined is not null)
+ {
+ var encoded = EncodeWindow(combined);
+ var size = (long)encoded.Length * 2;
+ if (owner.BufferedSend - tail.Value.Bytes + size <= WindowConfig.MaxBufferedSendBytes)
+ {
+ owner.BufferedSend += size - tail.Value.Bytes;
+ tail.Value.Notification = combined;
+ tail.Value.Text = encoded;
+ tail.Value.Bytes = size;
+ tail.Value.Waiters.Add((completion, token));
+ WindowEvent(owner, "coalesced", size);
+ break;
+ }
+ }
+ }
+ owner.Sends.AddLast(item);
+ owner.BufferedSend += item.Bytes;
+ break;
+ }
+ wait = owner.Writable.Task;
+ }
+ await Task.WhenAny(wait, Completion).WaitAsync(token).ConfigureAwait(false);
+ }
+ owner.Publish();
+ _writerWake.Release();
+ using (token.Register(() => _writerWake.Release()))
+ await completion.Task.WaitAsync(token).ConfigureAwait(false);
+ }
+
+ private bool TryWindowFrame(out TransportMessage frame, out Action sent)
+ {
+ List owners;
+ lock (_windowGate) owners = new List(_windows.Values);
+ foreach (var owner in owners)
+ {
+ lock (owner.Gate)
+ {
+ if (owner.Closed || owner.Negotiating || owner.Released <= owner.CreditSent) continue;
+ var total = owner.Released;
+ var notification = new JsonRpcNotification
+ {
+ Method = "channel/credit",
+ Params = _serializer.SerializeToElement(new ChannelCreditParams { Channel = owner.Uri, ConsumedBytes = total }),
+ };
+ frame = TransportMessage.FromText(EncodeWindow(notification));
+ owner.CreditSent = total;
+ var creditBytes = RawWindowBytes(frame);
+ sent = () =>
+ {
+ AhpTelemetry.WindowCreditBytesSent.Add(creditBytes);
+ WindowEvent(owner, "creditSent", creditBytes);
+ };
+ return true;
+ }
+ }
+ for (var i = 0; i < owners.Count; i++)
+ {
+ var index = (_windowTurn + i) % owners.Count;
+ var owner = owners[index];
+ lock (owner.Gate)
+ {
+ if (owner.Closed || owner.Negotiating) continue;
+ while (owner.Sends.First is { } canceled && !canceled.Value.Started
+ && canceled.Value.Waiters.All(x => x.Token.IsCancellationRequested))
+ {
+ owner.BufferedSend -= canceled.Value.Bytes;
+ foreach (var waiter in canceled.Value.Waiters) waiter.Completion.TrySetCanceled(waiter.Token);
+ owner.Sends.RemoveFirst();
+ if (canceled.Value.Notification.Method == "tcp/eof") owner.SendEof = false;
+ owner.Publish();
+ }
+ if (owner.Sends.First is not { } node
+ || (owner.Limits is not null && owner.Sent - owner.PeerConsumed >= owner.Limits.HostReceive.WindowBytes)) continue;
+ var item = node.Value;
+ string output;
+ bool final;
+ if (owner.Limits is null || (!item.Started && item.Bytes <= owner.Limits.HostReceive.MaximumFrameBytes))
+ {
+ output = item.Text;
+ final = true;
+ }
+ else
+ {
+ var remaining = owner.Limits.HostReceive.WindowBytes - (owner.Sent - owner.PeerConsumed);
+ output = EncodeFragment(owner, item, remaining, out var count);
+ item.Offset += count;
+ final = item.Offset == item.Text.Length;
+ item.FragmentWireBytes += (long)output.Length * 2;
+ }
+ item.Started = true;
+ _windowInFlight = item;
+ var charge = checked((long)output.Length * 2);
+ if (owner.Limits is not null) owner.Sent = checked(owner.Sent + charge);
+ if (owner.Sent > 9007199254740991) throw Protocol("window counter overflow");
+ if (final)
+ {
+ owner.Sends.RemoveFirst();
+ owner.BufferedSend -= item.Bytes;
+ }
+ _windowTurn = (index + 1) % owners.Count;
+ frame = TransportMessage.FromText(output);
+ sent = () =>
+ {
+ AhpTelemetry.WindowDataBytesSent.Add(charge);
+ _windowInFlight = null;
+ if (final)
+ {
+ foreach (var waiter in item.Waiters) waiter.Completion.TrySetResult(true);
+ if (item.FragmentWireBytes > 0) WindowEvent(owner, "fragmentedMessageSent", item.FragmentWireBytes);
+ }
+ owner.Publish();
+ };
+ return true;
+ }
+ }
+ frame = null!;
+ sent = null!;
+ return false;
+ }
+
+ private string EncodeFragment(WindowedSubscription owner, WindowSend item, long remaining, out int count)
+ {
+ string Encode(int length) => EncodeWindow(new JsonRpcNotification
+ {
+ Method = "channel/frame",
+ Params = _serializer.SerializeToElement(new ChannelFrameParams
+ {
+ Channel = owner.Uri,
+ Data = item.Text.Substring(item.Offset, length),
+ Final = item.Offset + length == item.Text.Length ? true : null,
+ }),
+ });
+ int Boundary(int length) => length > 0 && item.Offset + length < item.Text.Length
+ && char.IsHighSurrogate(item.Text[item.Offset + length - 1])
+ && char.IsLowSurrogate(item.Text[item.Offset + length]) ? length - 1 : length;
+ var smallest = char.IsHighSurrogate(item.Text[item.Offset]) ? 2 : 1;
+ var minimum = Encode(smallest);
+ var limit = Math.Min(owner.Limits!.HostReceive.MaximumFrameBytes, Math.Max(remaining, (long)minimum.Length * 2));
+ if ((long)minimum.Length * 2 > limit) throw Protocol("maximumFrameBytes cannot carry a character");
+ var low = smallest;
+ var high = item.Text.Length - item.Offset;
+ count = smallest;
+ var output = minimum;
+ while (low <= high)
+ {
+ var midpoint = low + (high - low) / 2;
+ var candidateCount = Boundary(midpoint);
+ if (candidateCount == 0) { low = midpoint + 1; continue; }
+ var candidate = Encode(candidateCount);
+ if ((long)candidate.Length * 2 <= limit)
+ {
+ count = candidateCount;
+ output = candidate;
+ low = midpoint + 1;
+ }
+ else high = midpoint - 1;
+ }
+ return output;
+ }
+
+ internal void CloseWindowOwner(WindowedSubscription owner)
+ {
+ owner.End();
+ _ = UnsubscribeOwnerAsync(owner);
+ }
+
+ internal void DropManagedWindow(WindowedSubscription owner) => RemoveWindow(owner);
+
+ private static void ValidateTcpData(string data)
+ {
+ if (data.Length == 0 || Convert.ToBase64String(Convert.FromBase64String(data)) != data)
+ throw Protocol("invalid TCP base64");
+ }
+
+ private void RemoveWindow(WindowedSubscription owner)
+ {
+ lock (_windowGate)
+ if (_windows.TryGetValue(owner.Uri, out var current) && ReferenceEquals(current, owner))
+ _windows.Remove(owner.Uri);
+ RememberClosedWindow(owner.Uri);
+ }
+
+ private void CleanupUnacceptedWindow(WindowedSubscription owner, Exception error)
+ {
+ if (owner.EverAccepted) return;
+ if (_windowResponses.Values.Any(x => !x.Abandoned && x.Owners.Contains(owner))) return;
+ owner.End(error, true);
+ RemoveWindow(owner);
+ }
+
+ private void AbandonWindowRequest(ulong id, Exception error)
+ {
+ if (_windowResponses.TryGetValue(id, out var response) && response.Abandon(error))
+ _windowResponses.TryRemove(id, out _);
+ }
+
+ private void StopWindow(string uri)
+ {
+ var owner = GetWindowedSubscription(uri);
+ if (owner is not null)
+ {
+ owner.End();
+ RemoveWindow(owner);
+ }
+ else if (_windowResponses.Values.Any(x => x.Owners.Any(owner => owner.Uri == uri)))
+ RememberClosedWindow(uri);
+ foreach (var pending in _windowResponses)
+ if (pending.Value.Owners.Any(x => x.Uri == uri))
+ {
+ var error = new AhpClientClosedException("The subscription was unsubscribed.");
+ AbandonWindowRequest(pending.Key, error);
+ if (_pending.TryRemove(pending.Key, out var completion)) completion.TrySetException(error);
+ }
+ }
+
+ internal int PendingWindowResponseCount => _windowResponses.Count;
+ internal int ClosedWindowCount { get { lock (_windowGate) return _closedWindows.Count; } }
+
+ private void RememberClosedWindow(string uri)
+ {
+ lock (_windowGate)
+ {
+ if (_closedWindows.TryGetValue(uri, out var previous)) _closedWindowOrder.Remove(previous);
+ _closedWindows[uri] = _closedWindowOrder.AddLast(uri);
+ if (_closedWindows.Count > 1024 && _closedWindowOrder.First is { } oldest)
+ {
+ _closedWindows.Remove(oldest.Value);
+ _closedWindowOrder.RemoveFirst();
+ }
+ }
+ }
+
+ private bool IsClosedWindow(string uri) { lock (_windowGate) return _closedWindows.ContainsKey(uri); }
+
+ private void ForgetClosedWindow(string uri)
+ {
+ lock (_windowGate)
+ if (_closedWindows.TryGetValue(uri, out var node))
+ {
+ _closedWindowOrder.Remove(node);
+ _closedWindows.Remove(uri);
+ }
+ }
+
+ private async Task UnsubscribeOwnerAsync(WindowedSubscription owner)
+ {
+ try { await UnsubscribeAsync(owner.Uri).ConfigureAwait(false); }
+ catch (AhpException) { }
+ }
+
+ private void EndWindows(Exception? cause)
+ {
+ _windowInFlight?.Fail(cause ?? new AhpClientClosedException());
+ List owners;
+ lock (_windowGate)
+ {
+ owners = new List(_windows.Values);
+ foreach (var reader in _flowEventReaders) reader.Writer.TryComplete();
+ }
+ foreach (var owner in owners)
+ {
+ if (RetainManagedWindows && !owner.Closed && !owner.Uri.StartsWith("ahp-tcp:", StringComparison.Ordinal))
+ {
+ lock (owner.Gate)
+ {
+ owner.Epoch++;
+ owner.Negotiating = true;
+ owner.BufferedReceive -= owner.FragmentBytes;
+ owner.FragmentBytes = 0;
+ owner.Fragments = null;
+ foreach (var item in owner.Sends) item.Fail(cause ?? new AhpClientClosedException());
+ owner.Sends.Clear();
+ owner.BufferedSend = 0;
+ }
+ }
+ else owner.End(cause, owner.Uri.StartsWith("ahp-tcp:", StringComparison.Ordinal));
+ owner.Publish();
+ }
+ _windowResponses.Clear();
+ _writerWake.Release();
+ }
+}
diff --git a/clients/dotnet/src/AgentHostProtocol/WindowedSubscription.cs b/clients/dotnet/src/AgentHostProtocol/WindowedSubscription.cs
new file mode 100644
index 000000000..2049edefc
--- /dev/null
+++ b/clients/dotnet/src/AgentHostProtocol/WindowedSubscription.cs
@@ -0,0 +1,334 @@
+#nullable enable
+
+using System.Runtime.CompilerServices;
+using System.Threading;
+using System.Threading.Channels;
+using System.Threading.Tasks;
+
+namespace Microsoft.AgentHostProtocol;
+
+/// An owned delivery. Release after processing; release and disposal are idempotent.
+public sealed class ReceivedDelivery : IDisposable
+{
+ private Action? _release;
+ public SubscriptionEvent Event { get; }
+ public JsonRpcNotification Notification { get; }
+ public long WireBytes { get; }
+ internal long CreditBytes { get; }
+ internal long Epoch { get; }
+
+ internal ReceivedDelivery(SubscriptionEvent value, JsonRpcNotification notification,
+ long bytes, long creditBytes, long epoch, Action release)
+ {
+ Event = value;
+ Notification = notification;
+ WireBytes = bytes;
+ CreditBytes = creditBytes;
+ Epoch = epoch;
+ _release = release;
+ }
+
+ public void Release() => Interlocked.Exchange(ref _release, null)?.Invoke();
+ public void Dispose() => Release();
+}
+
+/// The sole credit-owning reader for a channel. Observational streams do not release credit.
+public sealed class WindowedSubscription : IDisposable, IAsyncDisposable
+{
+ internal readonly object Gate = new();
+ internal readonly LinkedList Queue = new();
+ internal TaskCompletionSource Available = Signal();
+ internal TaskCompletionSource Writable = Signal();
+ internal readonly TaskCompletionSource Finished = Signal();
+ internal readonly List> StatusReaders = new();
+ internal AhpClient Client;
+ internal WindowOptions Options;
+ internal Hosts.HostId? HostId;
+ internal ChannelReceiveLimits Offer;
+ internal ChannelFlowControl? Limits;
+ internal string? Fallback;
+ internal long Epoch;
+ internal long Received;
+ internal long Released;
+ internal long CreditSent;
+ internal long Sent;
+ internal long PeerConsumed;
+ internal long BufferedSend;
+ internal long BufferedReceive;
+ internal long LastSeq;
+ internal bool AwaitingSnapshot;
+ internal bool Negotiating = true;
+ internal bool EverAccepted;
+ internal bool ResumeWindowed;
+ internal bool Closed;
+ internal bool SendEof;
+ internal bool ReceiveEof;
+ internal Exception? CloseError;
+ internal bool CreditTimer;
+ internal CancellationTokenSource? CreditCancellation;
+ internal Action? OnClosed;
+ internal readonly long TelemetryKey;
+ internal System.Text.StringBuilder? Fragments;
+ internal long FragmentBytes;
+ internal readonly LinkedList Sends = new();
+ internal long? BlockedAt;
+ private readonly DeliveryReader _reader;
+
+ public string Uri { get; }
+ public FlowControlStatus FlowControl
+ {
+ get
+ {
+ lock (Gate)
+ return new FlowControlStatus
+ {
+ Mode = Limits is null ? "ordinary" : "windowed",
+ FallbackReason = Fallback,
+ Limits = Limits,
+ OutstandingSendBytes = Sent - PeerConsumed,
+ BufferedSendBytes = BufferedSend,
+ BufferedReceiveBytes = BufferedReceive,
+ Writable = !Closed && !Negotiating && (Limits is null || Sent - PeerConsumed < Limits.HostReceive.WindowBytes)
+ && BufferedSend < Client.WindowConfig.MaxBufferedSendBytes,
+ };
+ }
+ }
+ public long LastAppliedServerSeq { get { lock (Gate) return LastSeq; } }
+ public event EventHandler? FlowControlChanged;
+ /// Manual-mode reader. Use ReadEventsAsync for automatic release on advance.
+ public ChannelReader Deliveries =>
+ Options.Receive == ReceiveMode.Manual ? _reader
+ : throw new InvalidOperationException("Use ReadEventsAsync for an automatic subscription.");
+
+ internal WindowedSubscription(AhpClient client, string uri, WindowOptions options, ChannelReceiveLimits offer)
+ {
+ Client = client;
+ HostId = client.WindowHostId;
+ Uri = uri;
+ Options = options;
+ Offer = offer;
+ _reader = new DeliveryReader(this);
+ TelemetryKey = AhpTelemetry.TrackWindow(this);
+ }
+
+ internal static TaskCompletionSource Signal() => new(TaskCreationOptions.RunContinuationsAsynchronously);
+
+ public ChannelReader FlowControlChanges()
+ {
+ var channel = Channel.CreateBounded(new BoundedChannelOptions(1)
+ {
+ FullMode = BoundedChannelFullMode.DropOldest,
+ });
+ lock (Gate) StatusReaders.Add(channel);
+ channel.Writer.TryWrite(FlowControl);
+ return channel.Reader;
+ }
+
+ public async IAsyncEnumerable ReadEventsAsync(
+ [EnumeratorCancellation] CancellationToken cancellationToken = default)
+ {
+ if (Options.Receive != ReceiveMode.Auto)
+ throw new InvalidOperationException("Use Deliveries for a manual subscription.");
+ await foreach (var delivery in _reader.ReadAllAsync(cancellationToken).ConfigureAwait(false))
+ {
+ using (delivery) yield return delivery.Event;
+ }
+ }
+
+ public async ValueTask WaitWritableAsync(CancellationToken cancellationToken = default)
+ {
+ while (true)
+ {
+ Task wait;
+ lock (Gate)
+ {
+ if (Closed) throw CloseError ?? new AhpClientClosedException();
+ if (Client.Completion.IsCompleted) throw new AhpClientClosedException();
+ if (FlowControl.Writable) return;
+ wait = Writable.Task;
+ }
+ await wait.WaitAsync(cancellationToken).ConfigureAwait(false);
+ }
+ }
+
+ public Task NotifyAsync(string method, T parameters, CancellationToken cancellationToken = default)
+ {
+ if (Closed) throw CloseError ?? new AhpClientClosedException();
+ var encoded = Client.WindowParameters(parameters);
+ if (!encoded.TryGetProperty("channel", out var channel) || channel.GetString() != Uri)
+ throw new ArgumentException("The notification must target this subscription.", nameof(parameters));
+ return Client.NotifyAsync(method, encoded, cancellationToken);
+ }
+ public Task DispatchAsync(StateAction action, long? clientSeq = null,
+ CancellationToken cancellationToken = default)
+ {
+ if (Closed) throw CloseError ?? new AhpClientClosedException();
+ return Client.DispatchAsync(Uri, action, clientSeq, cancellationToken);
+ }
+ public Task UnsubscribeAsync(CancellationToken cancellationToken = default)
+ {
+ if (ReferenceEquals(Client.GetWindowedSubscription(Uri), this) && !Client.Completion.IsCompleted)
+ return Client.UnsubscribeAsync(Uri, cancellationToken);
+ End();
+ return Task.CompletedTask;
+ }
+ public async ValueTask DisposeAsync() => await UnsubscribeAsync().ConfigureAwait(false);
+ public void Dispose()
+ {
+ if (!Closed && ReferenceEquals(Client.GetWindowedSubscription(Uri), this)) Client.CloseWindowOwner(this);
+ }
+
+ internal void Publish()
+ {
+ var status = FlowControl;
+ List> readers;
+ lock (Gate)
+ {
+ Writable.TrySetResult(true);
+ Writable = Signal();
+ readers = new List>(StatusReaders);
+ if (!status.Writable && !Closed && !Negotiating && BlockedAt is null)
+ {
+ BlockedAt = System.Diagnostics.Stopwatch.GetTimestamp();
+ Client.WindowEvent(this, "blocked");
+ }
+ else if ((status.Writable || Closed) && BlockedAt is { } start)
+ {
+ BlockedAt = null;
+ AhpTelemetry.WindowBlockedDuration.Record(Compatibility.GetElapsedTime(start).TotalMilliseconds);
+ Client.WindowEvent(this, "unblocked");
+ }
+ }
+ foreach (var reader in readers) reader.Writer.TryWrite(status);
+ if (FlowControlChanged is { } changed)
+ foreach (EventHandler observer in changed.GetInvocationList())
+ {
+ try { observer(this, status); }
+ catch (Exception) { /* Observational callbacks must not interrupt the protocol driver. */ }
+ }
+ }
+
+ internal void End(Exception? error = null, bool discard = false)
+ {
+ lock (Gate)
+ {
+ if (Closed) return;
+ Closed = true;
+ Epoch++;
+ CreditTimer = false;
+ CreditCancellation?.Cancel();
+ CloseError = error;
+ if (discard)
+ {
+ foreach (var delivery in Queue) BufferedReceive -= delivery.WireBytes;
+ Queue.Clear();
+ BufferedReceive -= FragmentBytes;
+ FragmentBytes = 0;
+ Fragments = null;
+ }
+ Available.TrySetResult(true);
+ Writable.TrySetResult(true);
+ foreach (var reader in StatusReaders) reader.Writer.TryComplete(error);
+ foreach (var send in Sends) send.Fail(error ?? new AhpClientClosedException());
+ Sends.Clear();
+ BufferedSend = 0;
+ if (Queue.Count == 0) CompleteReader();
+ }
+ AhpTelemetry.UntrackWindow(TelemetryKey);
+ OnClosed?.Invoke(this);
+ Publish();
+ }
+
+ internal ChannelReader InternalDeliveries => _reader;
+
+ private void CompleteReader()
+ {
+ if (CloseError is { } error) Finished.TrySetException(error);
+ else Finished.TrySetResult(true);
+ }
+
+ private sealed class DeliveryReader(WindowedSubscription owner) : ChannelReader
+ {
+ public override Task Completion => owner.Finished.Task;
+ public override bool TryRead(out ReceivedDelivery item)
+ {
+ lock (owner.Gate)
+ {
+ if (owner.Queue.First is { } first)
+ {
+ item = first.Value;
+ owner.Queue.RemoveFirst();
+ if (owner.Closed && owner.Queue.Count == 0) owner.CompleteReader();
+ return true;
+ }
+ item = null!;
+ return false;
+ }
+ }
+ public override async ValueTask WaitToReadAsync(CancellationToken cancellationToken = default)
+ {
+ while (true)
+ {
+ Task wait;
+ lock (owner.Gate)
+ {
+ if (owner.Queue.Count > 0) return true;
+ if (owner.Closed)
+ {
+ if (owner.CloseError is { } error) throw error;
+ return false;
+ }
+ wait = owner.Available.Task;
+ }
+ await wait.WaitAsync(cancellationToken).ConfigureAwait(false);
+ }
+ }
+ }
+}
+
+/// A notification without a dedicated event variant, including future methods.
+public sealed class SubscriptionEventNotification(JsonRpcNotification notification) : SubscriptionEvent
+{
+ public JsonRpcNotification Notification { get; } = notification;
+}
+public sealed class SubscriptionEventTcpData(TcpDataParams parameters) : SubscriptionEvent
+{
+ public TcpDataParams Params { get; } = parameters;
+}
+public sealed class SubscriptionEventTcpEof(TcpEofParams parameters) : SubscriptionEvent
+{
+ public TcpEofParams Params { get; } = parameters;
+}
+public sealed class SubscriptionEventTcpReset(TcpResetParams parameters) : SubscriptionEvent
+{
+ public TcpResetParams Params { get; } = parameters;
+}
+
+internal sealed class WindowSend
+{
+ internal JsonRpcNotification Notification;
+ internal string Text;
+ internal int Offset;
+ internal bool Started;
+ internal long Bytes;
+ internal long FragmentWireBytes;
+ internal readonly List<(TaskCompletionSource Completion, CancellationToken Token)> Waiters = new();
+ internal WindowSend(JsonRpcNotification notification, string text, TaskCompletionSource completion, CancellationToken token)
+ {
+ Notification = notification;
+ Text = text;
+ Bytes = (long)text.Length * 2;
+ Waiters.Add((completion, token));
+ }
+
+ internal void Fail(Exception error) { foreach (var waiter in Waiters) waiter.Completion.TrySetException(error); }
+}
+
+internal sealed class WindowWakeSignal
+{
+ private readonly Channel _channel = Channel.CreateBounded(
+ new BoundedChannelOptions(1) { FullMode = BoundedChannelFullMode.DropWrite });
+ internal void Release() => _channel.Writer.TryWrite(true);
+ internal async Task WaitAsync(CancellationToken cancellationToken) =>
+ _ = await _channel.Reader.ReadAsync(cancellationToken).ConfigureAwait(false);
+}
diff --git a/clients/dotnet/tests/AgentHostProtocol.AotSmoke/Program.cs b/clients/dotnet/tests/AgentHostProtocol.AotSmoke/Program.cs
index d4aed706c..b9ed0e364 100644
--- a/clients/dotnet/tests/AgentHostProtocol.AotSmoke/Program.cs
+++ b/clients/dotnet/tests/AgentHostProtocol.AotSmoke/Program.cs
@@ -76,12 +76,73 @@ public static async Task Main()
"Generated AHP metadata resolver did not provide ActionEnvelope metadata.");
await RunTransportScenarioAsync(serializer);
+ await RunWindowScenarioAsync(serializer);
Console.WriteLine(
"Native AOT smoke passed: initialize, reconnect, ping, subscribe/action/reducer/unsubscribe, "
+ "custom request/notification metadata, and typed/raw/null inbound request results.");
}
+ static async Task RunWindowScenarioAsync(SystemTextJsonAhpSerializer serializer)
+ {
+ const string channel = "ahp-test:/aot-window";
+ using var timeout = new CancellationTokenSource(TimeSpan.FromSeconds(10));
+ var (clientTransport, server) = DuplexTransport.CreatePair();
+ await using var serverOwner = server;
+ await using var client = AhpClient.Connect(clientTransport, serializer: serializer);
+ var limits = new ChannelReceiveLimits { WindowBytes = 256, MaximumFrameBytes = 1024 };
+ var response = RespondToRequestAsync(server, serializer, "subscribe",
+ new SubscribeResult { FlowControl = new ChannelFlowControl { ClientReceive = limits, HostReceive = limits } },
+ timeout.Token);
+ var (_, subscription) = await client.SubscribeWindowedAsync(channel,
+ new WindowOptions { Receive = ReceiveMode.Manual }, timeout.Token);
+ await response;
+ var snapshot = serializer.EncodeMessage(new JsonRpcMessage
+ {
+ Notification = new JsonRpcNotification
+ {
+ Method = "channel/snapshot",
+ Params = serializer.SerializeToElement(new ChannelSnapshotParams
+ {
+ Channel = channel,
+ Snapshot = new Snapshot
+ {
+ Resource = channel,
+ FromSeq = 1,
+ State = new SnapshotState { Empty = new EmptyState() },
+ },
+ }),
+ },
+ }).Text!;
+ var split = snapshot.Length / 2;
+ for (var part = 0; part < 2; part++)
+ {
+ await server.SendAsync(serializer.EncodeMessage(new JsonRpcMessage
+ {
+ Notification = new JsonRpcNotification
+ {
+ Method = "channel/frame",
+ Params = serializer.SerializeToElement(new ChannelFrameParams
+ {
+ Channel = channel,
+ Data = part == 0 ? snapshot.Substring(0, split) : snapshot.Substring(split),
+ Final = part == 1 ? true : null,
+ }),
+ },
+ }), timeout.Token);
+ if (part == 0)
+ {
+ var credit = await ReceiveMessageAsync(server, serializer, timeout.Token);
+ Ensure(credit.Notification?.Method == "channel/credit", "Nonfinal fragment was not credited.");
+ }
+ }
+ using var delivery = await subscription.Deliveries.ReadAsync(timeout.Token);
+ Ensure(delivery.Event is SubscriptionEventSnapshot, "Window snapshot was not reassembled.");
+ delivery.Release();
+ var finalCredit = await ReceiveMessageAsync(server, serializer, timeout.Token);
+ Ensure(finalCredit.Notification?.Method == "channel/credit", "Final fragment was not credited on release.");
+ }
+
static async Task RunTransportScenarioAsync(SystemTextJsonAhpSerializer serializer)
{
const string envelopeChannel = "ahp-session:/native-aot";
diff --git a/clients/dotnet/tests/AgentHostProtocol.Tests/WindowedClientTests.cs b/clients/dotnet/tests/AgentHostProtocol.Tests/WindowedClientTests.cs
new file mode 100644
index 000000000..1a60e7671
--- /dev/null
+++ b/clients/dotnet/tests/AgentHostProtocol.Tests/WindowedClientTests.cs
@@ -0,0 +1,341 @@
+#nullable enable
+
+using System.Diagnostics;
+using System.Diagnostics.Metrics;
+using System.Linq;
+using System.Threading;
+using System.Threading.Tasks;
+using Microsoft.AgentHostProtocol.Hosts;
+using Xunit;
+
+namespace Microsoft.AgentHostProtocol.Tests;
+
+public sealed class WindowedClientTests
+{
+ private const string Uri = "ahp-test:/window";
+ private const string SnapshotText = "{\"jsonrpc\":\"2.0\",\"method\":\"channel/snapshot\",\"params\":{\"channel\":\"ahp-test:/window\",\"snapshot\":{\"resource\":\"ahp-test:/window\",\"state\":{},\"fromSeq\":7}}}";
+ private static readonly SystemTextJsonAhpSerializer Ser = SystemTextJsonAhpSerializer.Default;
+
+ [Fact]
+ public async Task BlockedChannelDoesNotHoldOtherChannelsOrCanceledSends() =>
+ await CheckFairnessAsync();
+
+ private static async Task CheckFairnessAsync()
+ {
+ var transport = new CorpusTransport();
+ RespondAutomatically(transport);
+ await using var client = AhpClient.Connect(transport);
+ var (_, blocked) = await client.SubscribeWindowedAsync(Uri,
+ new WindowOptions { WindowBytes = 512, MaximumFrameBytes = 1024 }).ConfigureAwait(false);
+ var (_, other) = await client.SubscribeWindowedAsync("ahp-test:/other").ConfigureAwait(false);
+ await blocked.NotifyAsync("future/data", new TcpDataParams { Channel = Uri, Data = new string('x', 220) }).ConfigureAwait(false);
+ Assert.False(blocked.FlowControl.Writable);
+ using var cancellation = new CancellationTokenSource();
+ var canceled = blocked.NotifyAsync("future/canceled", new TcpEofParams { Channel = Uri }, cancellation.Token);
+ await WindowedDeliveryCorpusTests.UntilAsync(() => blocked.FlowControl.BufferedSendBytes > 0).ConfigureAwait(false);
+ cancellation.Cancel();
+ await Assert.ThrowsAnyAsync(() => canceled).ConfigureAwait(false);
+ await other.NotifyAsync("future/data", new TcpEofParams { Channel = other.Uri }).WaitAsync(TimeSpan.FromSeconds(1)).ConfigureAwait(false);
+ await WindowedDeliveryCorpusTests.UntilAsync(() => blocked.FlowControl.BufferedSendBytes == 0).ConfigureAwait(false);
+ Assert.DoesNotContain(transport.Snapshot(), x => x.Json.GetProperty("method").GetString() == "future/canceled");
+ var writable = blocked.WaitWritableAsync().AsTask();
+ Assert.False(writable.IsCompleted);
+ transport.Inject(Ser.Serialize(new JsonRpcNotification
+ {
+ Method = "channel/credit",
+ Params = Ser.SerializeToElement(new ChannelCreditParams { Channel = Uri, ConsumedBytes = blocked.Sent }),
+ }));
+ await writable.WaitAsync(TimeSpan.FromSeconds(1)).ConfigureAwait(false);
+ }
+
+ [Fact]
+ public async Task MissingWindowClosesItsRetainedQueue() =>
+ await CheckMissingWindowAsync();
+
+ private static async Task CheckMissingWindowAsync()
+ {
+ var transport = new CorpusTransport();
+ RespondAutomatically(transport, keepWindows: false);
+ await using var client = AhpClient.Connect(transport);
+ var (_, owner) = await client.SubscribeWindowedAsync(Uri, new WindowOptions { Receive = ReceiveMode.Manual }).ConfigureAwait(false);
+ transport.Inject(SnapshotText);
+ await WindowedDeliveryCorpusTests.UntilAsync(() => owner.LastAppliedServerSeq == 7).ConfigureAwait(false);
+ await client.ResumeAsync(new ReconnectParams
+ {
+ Channel = ProtocolVersion.RootResourceUri,
+ ClientId = "missing-window",
+ LastSeenServerSeq = 7,
+ Subscriptions = new List { Uri },
+ Windows = new ChannelDeliveryResumeOptions
+ {
+ Items = new List
+ {
+ new() { Channel = Uri, FlowControl = new SubscriptionFlowControlOptions { Receive = owner.Offer } },
+ },
+ },
+ }).ConfigureAwait(false);
+ Assert.Null(client.GetWindowedSubscription(Uri));
+ await Assert.ThrowsAsync(() => owner.Deliveries.ReadAsync().AsTask()).ConfigureAwait(false);
+ Assert.Equal(0, owner.FlowControl.BufferedReceiveBytes);
+ }
+
+ [Fact]
+ public async Task TcpOwnerClosesAndIsExcludedFromManagedReconnect() =>
+ await CheckTcpReconnectAsync();
+
+ private static async Task CheckTcpReconnectAsync()
+ {
+ var transports = new List();
+ await using var multi = new MultiHostClient();
+ await multi.AddHostAsync(new HostConfig
+ {
+ Id = "tcp-host",
+ ReconnectPolicy = ReconnectPolicy.Disabled,
+ TransportFactory = (_, _) =>
+ {
+ var transport = new CorpusTransport();
+ RespondAutomatically(transport);
+ lock (transports) transports.Add(transport);
+ return Task.FromResult(transport);
+ },
+ }).ConfigureAwait(false);
+ var (_, tcp) = await multi.SubscribeWindowedAsync("tcp-host", "ahp-tcp:/test",
+ new WindowOptions { Receive = ReceiveMode.Manual }).ConfigureAwait(false);
+ var generation = multi.Host("tcp-host")!.Generation;
+ await multi.ReconnectAsync("tcp-host").ConfigureAwait(false);
+ await WindowedDeliveryCorpusTests.UntilAsync(() => multi.Host("tcp-host")!.Generation > generation).ConfigureAwait(false);
+ Assert.True(tcp.Closed);
+ var request = transports[1].Snapshot().First(x => x.Json.GetProperty("method").GetString() == "reconnect").Json;
+ Assert.DoesNotContain(request.GetProperty("params").GetProperty("subscriptions").EnumerateArray(),
+ x => x.GetString() == tcp.Uri);
+ Assert.False(request.GetProperty("params").TryGetProperty("windows", out _));
+ }
+
+ [Theory]
+ [InlineData(null, 1048576)]
+ [InlineData(33554432L, 22369620)]
+ public async Task TransportDefaultsAreEncodingSafe(long? cap, long expected) =>
+ await CheckDefaultAsync(cap, expected);
+
+ private static async Task CheckDefaultAsync(long? cap, long expected)
+ {
+ var transport = new CorpusTransport { MaximumInboundMessageBytes = cap };
+ await using var client = AhpClient.Connect(transport);
+ RespondAutomatically(transport);
+ var (_, owner) = await client.SubscribeWindowedAsync(Uri).ConfigureAwait(false);
+ Assert.Equal(expected, owner.Offer.MaximumFrameBytes);
+ Assert.Equal(262144, owner.Offer.WindowBytes);
+ }
+
+ [Fact]
+ public async Task AutomaticIteratorReleasesOnlyOnAdvanceAndDisposal() =>
+ await CheckAutomaticAsync();
+
+ private static async Task CheckAutomaticAsync()
+ {
+ var transport = new CorpusTransport();
+ RespondAutomatically(transport);
+ await using var client = AhpClient.Connect(transport);
+ var (_, owner) = await client.SubscribeWindowedAsync(Uri).ConfigureAwait(false);
+ var observers = client.CreateEventStream();
+ using (observers)
+ {
+ transport.Inject(SnapshotText);
+ await using var iterator = owner.ReadEventsAsync().GetAsyncEnumerator();
+ Assert.True(await iterator.MoveNextAsync().ConfigureAwait(false));
+ Assert.IsType(iterator.Current);
+ Assert.DoesNotContain(transport.Snapshot(), x => x.Json.GetProperty("method").GetString() == "channel/credit");
+ var waiting = iterator.MoveNextAsync().AsTask();
+ await WindowedDeliveryCorpusTests.UntilAsync(() => HasCredit(transport, (long)SnapshotText.Length * 2)).ConfigureAwait(false);
+ Assert.False(waiting.IsCompleted);
+ transport.Inject("{\"jsonrpc\":\"2.0\",\"method\":\"future/data\",\"params\":{\"channel\":\"ahp-test:/window\",\"value\":1}}");
+ Assert.True(await waiting.ConfigureAwait(false));
+ Assert.IsType(iterator.Current);
+ }
+ await WindowedDeliveryCorpusTests.UntilAsync(() => owner.FlowControl.BufferedReceiveBytes == 0).ConfigureAwait(false);
+ }
+
+ [Theory]
+ [InlineData("snapshot")]
+ [InlineData("replay")]
+ [InlineData("live")]
+ public async Task ManagedEpochRetainsUnreadItemsWithoutCreditingNewConnection(string recovery) =>
+ await CheckEpochAsync(recovery);
+
+ private static async Task CheckEpochAsync(string recovery)
+ {
+ var first = new CorpusTransport();
+ RespondAutomatically(first);
+ await using var oldClient = AhpClient.Connect(first);
+ oldClient.RetainManagedWindows = true;
+ var (_, owner) = await oldClient.SubscribeWindowedAsync(Uri, new WindowOptions { Receive = ReceiveMode.Manual }).ConfigureAwait(false);
+ first.Inject(SnapshotText);
+ await WindowedDeliveryCorpusTests.UntilAsync(() => owner.Queue.Count == 1).ConfigureAwait(false);
+ await oldClient.ShutdownAsync().ConfigureAwait(false);
+ var transport = new CorpusTransport();
+ RespondAutomatically(transport, recovery);
+ await using var client = AhpClient.Connect(transport);
+ client.BindManagedWindow(owner);
+ var result = await client.ResumeAsync(new ReconnectParams
+ {
+ Channel = ProtocolVersion.RootResourceUri,
+ ClientId = "window-client",
+ LastSeenServerSeq = 7,
+ Subscriptions = new List { Uri },
+ Windows = new ChannelDeliveryResumeOptions
+ {
+ Items = new List
+ {
+ new() { Channel = Uri, FlowControl = new SubscriptionFlowControlOptions { Receive = owner.Offer }, LastAppliedServerSeq = owner.LastAppliedServerSeq },
+ },
+ },
+ }).ConfigureAwait(false);
+ Assert.NotNull(result);
+ var old = await owner.Deliveries.ReadAsync().ConfigureAwait(false);
+ old.Dispose();
+ Assert.DoesNotContain(transport.Snapshot(), x => x.Json.GetProperty("method").GetString() == "channel/credit");
+ Assert.Equal(0, owner.FlowControl.BufferedReceiveBytes);
+ if (recovery == "snapshot") transport.Inject(SnapshotText);
+ else transport.Inject("{\"jsonrpc\":\"2.0\",\"method\":\"future/data\",\"params\":{\"channel\":\"ahp-test:/window\"}}");
+ using var current = await owner.Deliveries.ReadAsync().ConfigureAwait(false);
+ current.Release();
+ await WindowedDeliveryCorpusTests.UntilAsync(() => HasCredit(transport, current.WireBytes)).ConfigureAwait(false);
+ }
+
+ [Fact]
+ public async Task MultiHostKeepsOwnerAndOptionsAcrossReconnect() =>
+ await CheckMultiHostAsync();
+
+ private static async Task CheckMultiHostAsync()
+ {
+ var transports = new List();
+ await using var multi = new MultiHostClient();
+ await multi.AddHostAsync(new HostConfig
+ {
+ Id = "window-host",
+ ReconnectPolicy = ReconnectPolicy.Disabled,
+ TransportFactory = (_, _) =>
+ {
+ var transport = new CorpusTransport();
+ RespondAutomatically(transport);
+ lock (transports) transports.Add(transport);
+ return Task.FromResult(transport);
+ },
+ }).ConfigureAwait(false);
+ var (_, owner) = await multi.SubscribeWindowedAsync("window-host", Uri,
+ new WindowOptions { Receive = ReceiveMode.Manual, WindowBytes = 8192 }).ConfigureAwait(false);
+ transports[0].Inject(SnapshotText);
+ await WindowedDeliveryCorpusTests.UntilAsync(() => owner.LastAppliedServerSeq == 7).ConfigureAwait(false);
+ var generation = multi.Host("window-host")!.Generation;
+ await multi.ReconnectAsync("window-host").ConfigureAwait(false);
+ await WindowedDeliveryCorpusTests.UntilAsync(() => multi.Host("window-host")!.Generation > generation).ConfigureAwait(false);
+ Assert.Same(owner, multi.GetWindowedSubscription("window-host", Uri));
+ var request = transports[1].Snapshot().First(x => x.Json.GetProperty("method").GetString() == "reconnect").Json;
+ Assert.Equal(7, request.GetProperty("params").GetProperty("windows").GetProperty("items")[0].GetProperty("lastAppliedServerSeq").GetInt64());
+ Assert.Equal(8192, owner.Offer.WindowBytes);
+ using var old = await owner.Deliveries.ReadAsync().ConfigureAwait(false);
+ old.Release();
+ Assert.DoesNotContain(transports[1].Snapshot(), x => x.Json.GetProperty("method").GetString() == "channel/credit");
+ }
+
+ [Fact]
+ public async Task CustomCoalescerAndTelemetryAreObservable() =>
+ await CheckTelemetryAsync();
+
+ private static async Task CheckTelemetryAsync()
+ {
+ var measurements = new System.Collections.Concurrent.ConcurrentDictionary(StringComparer.Ordinal);
+ using var meter = new MeterListener
+ {
+ InstrumentPublished = (instrument, listener) =>
+ {
+ if (instrument.Meter.Name == AhpTelemetry.Name) listener.EnableMeasurementEvents(instrument);
+ },
+ };
+ meter.SetMeasurementEventCallback((instrument, value, _, _) =>
+ measurements.AddOrUpdate(instrument.Name, value, (_, previous) => previous + value));
+ meter.Start();
+ var events = new System.Collections.Concurrent.ConcurrentBag();
+ using var activities = new ActivityListener
+ {
+ ShouldListenTo = source => source.Name == AhpTelemetry.Name,
+ Sample = (ref ActivityCreationOptions _) => ActivitySamplingResult.AllData,
+ ActivityStopped = activity =>
+ {
+ foreach (var ev in activity.Events) events.Add(ev.Name);
+ },
+ };
+ ActivitySource.AddActivityListener(activities);
+ var transport = new CorpusTransport();
+ RespondAutomatically(transport);
+ var config = new ClientConfig();
+ config.Windowing.Coalescers.Add(new TestCoalescer());
+ await using var client = AhpClient.Connect(transport, config);
+ var flowEvents = client.FlowControlEvents();
+ var (_, owner) = await client.SubscribeWindowedAsync(Uri, new WindowOptions { Receive = ReceiveMode.Manual }).ConfigureAwait(false);
+ transport.Inject(SnapshotText);
+ using var snapshot = await owner.Deliveries.ReadAsync().ConfigureAwait(false);
+ snapshot.Release();
+ const string text = "{\"jsonrpc\":\"2.0\",\"method\":\"root/progress\",\"params\":{\"channel\":\"ahp-test:/window\",\"progressToken\":\"p\"}}";
+ transport.Inject(text);
+ transport.Inject(text);
+ await WindowedDeliveryCorpusTests.UntilAsync(() =>
+ {
+ lock (owner.Gate) return owner.Queue.First?.Value.WireBytes == text.Length * 4L;
+ }).ConfigureAwait(false);
+ meter.RecordObservableInstruments();
+ Assert.True(measurements.ContainsKey("ahp.client.window.buffered"));
+ using var merged = await owner.Deliveries.ReadAsync().ConfigureAwait(false);
+ merged.Release();
+ await WindowedDeliveryCorpusTests.UntilAsync(() => measurements.ContainsKey("ahp.client.window.credit.sent")).ConfigureAwait(false);
+ Assert.True(measurements["ahp.client.window.data.received"] > 0);
+ Assert.Contains(events, x => x == "ahp.flowcontrol.coalesced");
+ Assert.True(flowEvents.TryRead(out _));
+ }
+
+ private sealed class TestCoalescer : IChannelMessageCoalescer
+ {
+ public JsonRpcNotification? Coalesce(JsonRpcNotification earlier, JsonRpcNotification later, CoalescingContext context) =>
+ context.Direction == CoalescingDirection.Incoming && earlier.Method == "root/progress" ? later : null;
+ }
+
+ private static bool HasCredit(CorpusTransport transport, long amount) => transport.Snapshot().Any(x =>
+ x.Json.GetProperty("method").GetString() == "channel/credit"
+ && x.Json.GetProperty("params").GetProperty("consumedBytes").GetInt64() >= amount);
+
+ private static void RespondAutomatically(CorpusTransport transport, string recovery = "replay", bool keepWindows = true)
+ {
+ transport.OnSent = request =>
+ {
+ if (!request.TryGetProperty("id", out var id)) return;
+ var method = request.GetProperty("method").GetString();
+ object result;
+ if (method == "subscribe")
+ {
+ var offer = Ser.Deserialize(request.GetProperty("params")).FlowControl!.Receive;
+ result = new SubscribeResult { FlowControl = new ChannelFlowControl { ClientReceive = offer, HostReceive = offer } };
+ }
+ else if (method == "initialize")
+ result = new InitializeResult { ProtocolVersion = "1.0.0", ServerSeq = 0, Snapshots = new List() };
+ else if (method == "reconnect")
+ {
+ var parameters = Ser.Deserialize(request.GetProperty("params"));
+ var items = parameters.Windows?.Items.Where(_ => keepWindows).Select(x => new ResumedChannelSubscription
+ {
+ Channel = x.Channel,
+ FlowControl = new ChannelFlowControl { ClientReceive = x.FlowControl.Receive, HostReceive = x.FlowControl.Receive },
+ Recovery = recovery switch { "snapshot" => ChannelRecoveryKind.Snapshot, "live" => ChannelRecoveryKind.Live, _ => ChannelRecoveryKind.Replay },
+ }).ToList();
+ result = new ReconnectReplayResult
+ {
+ Type = ReconnectResultType.Replay,
+ Actions = new List(),
+ Missing = new List(),
+ Windows = items is null ? null : new ChannelDeliveryResumeResult { Items = items },
+ };
+ }
+ else result = new Dictionary();
+ transport.Inject("{\"jsonrpc\":\"2.0\",\"id\":" + id.GetRawText() + ",\"result\":" + Ser.Serialize(result) + "}");
+ };
+ }
+}
diff --git a/clients/dotnet/tests/AgentHostProtocol.Tests/WindowedDeliveryCorpusTests.cs b/clients/dotnet/tests/AgentHostProtocol.Tests/WindowedDeliveryCorpusTests.cs
new file mode 100644
index 000000000..b9ee8ee2c
--- /dev/null
+++ b/clients/dotnet/tests/AgentHostProtocol.Tests/WindowedDeliveryCorpusTests.cs
@@ -0,0 +1,314 @@
+#nullable enable
+
+using System.IO;
+using System.Linq;
+using System.Threading;
+using System.Threading.Channels;
+using System.Threading.Tasks;
+using Xunit;
+
+namespace Microsoft.AgentHostProtocol.Tests;
+
+public sealed class WindowedDeliveryCorpusTests
+{
+ private static readonly JsonSerializerOptions StatusJson = new() { PropertyNamingPolicy = JsonNamingPolicy.CamelCase };
+ private static string Corpus
+ {
+ get
+ {
+ var directory = new DirectoryInfo(AppContext.BaseDirectory);
+ while (directory is not null)
+ {
+ var candidate = Path.Combine(directory.FullName, "types", "test-cases", "windowed-delivery");
+ if (Directory.Exists(candidate)) return candidate;
+ directory = directory.Parent;
+ }
+ throw new DirectoryNotFoundException("Shared windowed delivery corpus was not found.");
+ }
+ }
+
+ public static IEnumerable