clients: add opt-in windowed delivery and examples

clients/dotnet/AgentHostProtocol.slnx

@@ -2,6 +2,8 @@
   <Folder Name="/examples/">
     <Project Path="examples/ConnectWs/ConnectWs.csproj" />
     <Project Path="examples/ReducersDemo/ReducersDemo.csproj" />
+    <Project Path="examples/WindowingBackpressure/WindowingBackpressure.csproj" />
+    <Project Path="examples/WindowingBasic/WindowingBasic.csproj" />
   </Folder>
   <Folder Name="/src/">
     <Project Path="src/AgentHostProtocol.Abstractions/AgentHostProtocol.Abstractions.csproj" />

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`

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.

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 };
+
+    /// <summary>Quotes text the way JavaScript's JSON.stringify does.</summary>
+    public static string Quote(string text) => JsonSerializer.Serialize(text, QuoteOptions);
+
+    /// <summary>
+    /// 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
+    /// <see cref="WindowedSubscription.ReadEventsAsync"/> performs, but it also
+    /// exposes each delivery's charged size (<see cref="ReceivedDelivery.WireBytes"/>),
+    /// which the examples print. Subscribe with <see cref="ReceiveMode.Manual"/>.
+    /// </summary>
+    public static async IAsyncEnumerable<ReceivedDelivery> TakeDeliveriesAsync(
+        this WindowedSubscription subscription,
+        [EnumeratorCancellation] CancellationToken cancellationToken = default)
+    {
+        await foreach (var delivery in subscription.Deliveries.ReadAllAsync(cancellationToken).ConfigureAwait(false))
+        {
+            using (delivery) yield return delivery;
+        }
+    }
+}

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<TransportMessage> _inbox;
+    private readonly Channel<TransportMessage> _outbox;
+
+    private InMemoryTransport(Channel<TransportMessage> inbox, Channel<TransportMessage> outbox)
+    {
+        _inbox = inbox;
+        _outbox = outbox;
+    }
+
+    /// <summary>Creates a linked pair: what one half sends, the other half receives.</summary>
+    public static (InMemoryTransport Client, InMemoryTransport Host) CreatePair()
+    {
+        var toHost = Channel.CreateUnbounded<TransportMessage>(new UnboundedChannelOptions { SingleReader = true });
+        var toClient = Channel.CreateUnbounded<TransportMessage>(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<TransportMessage> 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();
+}