[RT-E2] refactor: terminals run on the client's TimeProvider

docs/architecture/client-runtime.md

@@ -261,7 +261,10 @@ upstream's behaviour. Each one is listed, with its reason, in the section it con
   three named seams: `ITerminalFrameDecoder<TFrame>` for what a message carries and
   `ITerminalClosePolicy` for what a close status means, both consumed by the core, and
   `ITerminalUpgradeFailurePolicy` for what a refused upgrade means, consumed by the shipped
-  adapter at connect — before a core exists at all.
+  adapter at connect — before a core exists at all. The pipeline publishes the client's
+  `TimeProvider` in `ConnectionSnapshot`; both connect paths pass it through their sessions to
+  the core's send deadlines and send queue. Send admission and close-output each arm their own
+  independent graceful-close budget on that same clock.
 
 ## Error channels
 

src/OpenCode.Sdk/Internal/Terminal/TerminalReceivePump.cs

@@ -14,7 +14,7 @@ internal sealed class TerminalReceivePump<TFrame> : IAsyncDisposable
         SingleReader = false,
         AllowSynchronousContinuations = false,
     });
-    private readonly TerminalCancellationSource _lifetime = new();
+    private readonly CancellationTokenSource _lifetime = new();
     private readonly Lock _sync = new();
     private readonly Task _completion;
     private bool _abandoned;
@@ -103,7 +103,7 @@ public async ValueTask DisposeAsync()
     {
         try
         {
-            await _lifetime.CancelAsync().ConfigureAwait(false);
+            await _lifetime.CancelOnWorkerAsync().ConfigureAwait(false);
             // The receiver is owned here and every await in it declines context capture.
 #pragma warning disable VSTHRD003 // Joining this connection's receiver is the disposal contract.
             await _completion.ConfigureAwait(false);

src/OpenCode.Sdk/Internal/Terminal/TerminalSendDeadline.cs

@@ -19,14 +19,14 @@ internal sealed class TerminalSendDeadline : IDisposable
     public TerminalSendDeadline(
         Type owner,
         TimeSpan timeout,
-        ITerminalDeadlineFactory factory,
+        TimeProvider time,
         Func<TerminalConnectionOutcome?> getOutcome,
         CancellationToken caller)
     {
         _owner = owner;
         _caller = caller;
         _getOutcome = getOutcome;
-        _deadline = factory.Create(timeout);
+        _deadline = TimedCancellation.Create(timeout, time);
         _operation = CancellationTokenSource.CreateLinkedTokenSource(caller, _deadline.Token);
         // Cancellation callbacks run in LIFO order: latch before the link interrupts physical I/O.
         // A deadline already canceled during construction has no admitted operation to interrupt.

src/OpenCode.Sdk/Internal/Terminal/TerminalSendQueue.cs

@@ -1,7 +1,7 @@
 namespace OpenCode.Sdk.Internal;
 
 /// <summary>Serializes physical writes and graceful close behind one connection gate.</summary>
-internal sealed class TerminalSendQueue(ITerminalWebSocket socket) : IDisposable
+internal sealed class TerminalSendQueue(ITerminalWebSocket socket, TimeProvider time) : IDisposable
 {
     private readonly SemaphoreSlim _gate = new(1, 1);
 
@@ -20,14 +20,21 @@ internal sealed class TerminalSendQueue(ITerminalWebSocket socket) : IDisposable
         "and receiver/send joins in DisposeAsync; disposal completion is the observable cleanup outcome.")]
     public async Task CloseAsync()
     {
-        if (!await _gate.WaitAsync(TerminalSocketBounds.GracefulCloseTimeout, CancellationToken.None).ConfigureAwait(false))
+        using (var admission = TimedCancellation.Create(TerminalSocketBounds.GracefulCloseTimeout, time))
         {
-            return;
+            try
+            {
+                await _gate.WaitAsync(admission.Token).ConfigureAwait(false);
+            }
+            catch (OperationCanceledException) when (admission.IsCancellationRequested)
+            {
+                return;
+            }
         }
 
-        using var timeout = new CancellationTokenSource(TerminalSocketBounds.GracefulCloseTimeout);
         try
         {
+            using var timeout = TimedCancellation.Create(TerminalSocketBounds.GracefulCloseTimeout, time);
             await socket.CloseOutputAsync(timeout.Token).ConfigureAwait(false);
         }
         catch (Exception exception) when (

src/OpenCode.Sdk/Internal/Terminal/TerminalSocketCore.cs

@@ -17,10 +17,11 @@ internal sealed class TerminalSocketCore<TFrame> : IAsyncDisposable
     private readonly TerminalReceivePump<TFrame> _receiver;
     private readonly Func<TerminalConnectionOutcome?> _getOutcome;
     private readonly Type _owner;
-    private readonly TerminalCancellationSource _sendEnded = new();
+    private readonly CancellationTokenSource _sendEnded = new();
     private readonly CancellationToken _sendEndedToken;
     private readonly TerminalSendQueue _sendGate;
     private readonly ITerminalWebSocket _socket;
+    private readonly TimeProvider _time;
     private int _disposed;
     private int _reading;
     private int _socketDisposed;
@@ -32,18 +33,20 @@ public TerminalSocketCore(
         ITerminalWebSocket socket,
         ITerminalFrameDecoder<TFrame> decoder,
         ITerminalClosePolicy closePolicy,
-        Type owner)
+        Type owner,
+        TimeProvider time)
     {
         ArgumentNullException.ThrowIfNull(socket);
         ArgumentNullException.ThrowIfNull(decoder);
         ArgumentNullException.ThrowIfNull(closePolicy);
         ArgumentNullException.ThrowIfNull(owner);
 
         _socket = socket;
-        _sendGate = new TerminalSendQueue(socket);
+        _time = time;
+        _sendGate = new TerminalSendQueue(socket, time);
         _owner = owner;
         _sendEndedToken = _sendEnded.Token;
-        _receiver = new TerminalReceivePump<TFrame>(socket, decoder, closePolicy, _sendEnded.CancelAsync);
+        _receiver = new TerminalReceivePump<TFrame>(socket, decoder, closePolicy, _sendEnded.CancelOnWorkerAsync);
         _getOutcome = () => _receiver.Outcome;
     }
 
@@ -53,9 +56,6 @@ public TerminalSocketCore(
     /// <summary>Gets the fixed per-send budget captured by the connection.</summary>
     public TimeSpan SendTimeout { get; init; } = TerminalSocketBounds.DefaultSendTimeout;
 
-    /// <summary>Gets the timer boundary used for per-send deadlines.</summary>
-    public ITerminalDeadlineFactory DeadlineFactory { get; init; } = TerminalDeadlineFactory.Instance;
-
     /// <summary>
     /// Reads the frames the server sends until it closes the connection normally. One core
     /// carries one active enumeration: undelivered frames have one consumer, so a second
@@ -126,7 +126,7 @@ private async Task SendCoreAsync(
         CancellationToken cancellationToken)
     {
 
-        using var deadline = new TerminalSendDeadline(_owner, SendTimeout, DeadlineFactory, _getOutcome, cancellationToken);
+        using var deadline = new TerminalSendDeadline(_owner, SendTimeout, _time, _getOutcome, cancellationToken);
         var payload = createPayload();
         using var admission = CancellationTokenSource.CreateLinkedTokenSource(deadline.Token, _sendEndedToken);
         try
@@ -199,7 +199,7 @@ public ValueTask DisposeAsync()
 
     private async Task DisposeCoreAsync()
     {
-        await _sendEnded.CancelAsync().ConfigureAwait(false);
+        await _sendEnded.CancelOnWorkerAsync().ConfigureAwait(false);
 
         try
         {
@@ -254,7 +254,7 @@ private async Task FailSendAsync(Exception failure)
         var terminalFailure = failure as OpenCodeTransportException ?? new OpenCodeTransportException(
             "The opencode PTY WebSocket was interrupted during a send; delivery is uncertain.", failure);
         _receiver.Complete(terminalFailure);
-        await _sendEnded.CancelAsync().ConfigureAwait(false);
+        await _sendEnded.CancelOnWorkerAsync().ConfigureAwait(false);
         DisposeSocket();
     }
 

src/OpenCode.Sdk/Internal/Transport/ConnectionSnapshot.cs

@@ -19,13 +19,14 @@ namespace OpenCode.Sdk.Internal;
 /// </remarks>
 internal sealed class ConnectionSnapshot
 {
-    public ConnectionSnapshot(string endpointBase, AuthenticationHeaderValue? authorization, LocationSelector? location)
+    public ConnectionSnapshot(string endpointBase, AuthenticationHeaderValue? authorization, LocationSelector? location, TimeProvider time)
     {
         ArgumentException.ThrowIfNullOrWhiteSpace(endpointBase);
 
         EndpointBase = endpointBase;
         Authorization = authorization;
         Location = location;
+        Time = time;
     }
 
     /// <summary>Gets the Basic credential every request carries, or null when no password was configured.</summary>
@@ -36,4 +37,7 @@ public ConnectionSnapshot(string endpointBase, AuthenticationHeaderValue? author
 
     /// <summary>Gets the ambient location a per-call selector merges over member by member.</summary>
     public LocationSelector? Location { get; }
+
+    /// <summary>Gets the client's clock for non-HTTP send and close deadlines.</summary>
+    public TimeProvider Time { get; }
 }

src/OpenCode.Sdk/Internal/Transport/Pipeline.cs

@@ -91,7 +91,7 @@ internal Pipeline(HttpClient httpClient, bool ownsHttpClient, OpenCodeClientOpti
         // The same snapshot, published for the one door that cannot ride these policies: a PTY
         // WebSocket upgrade builds its own socket and therefore needs the endpoint, the
         // credential, and the ambient location the policies keep to themselves.
-        Connection = new ConnectionSnapshot(_endpointBase, authorization, options.Location);
+        Connection = new ConnectionSnapshot(_endpointBase, authorization, options.Location, _time);
     }
 
     /// <summary>Gets the construction-time connection facts a non-HTTP door addresses the server with.</summary>

src/OpenCode.Sdk/PersistentPtys/PersistentPtyClient.cs

@@ -157,7 +157,7 @@ await socket
                 .ConnectAsync(address, ptyId, PersistentPtyUpgradeFailurePolicy.Instance, cancellationToken)
                 .ConfigureAwait(false);
             var session = await PersistentPtySession.AttachAsync(
-                socket, ptyId, options?.SendTimeout ?? TerminalSocketBounds.DefaultSendTimeout, cancellationToken)
+                socket, ptyId, connection.Time, options?.SendTimeout ?? TerminalSocketBounds.DefaultSendTimeout, cancellationToken)
                 .ConfigureAwait(false);
             socket = null;
             return session;

src/OpenCode.Sdk/PersistentPtys/PersistentPtySession.cs

@@ -157,24 +157,28 @@ public virtual async ValueTask DisposeAsync()
     /// </summary>
     /// <param name="socket">The upgraded socket; the session takes ownership of it.</param>
     /// <param name="ptyId">The terminal the socket addresses; it names the failures.</param>
+    /// <param name="time">The client's clock for send and close deadlines.</param>
     /// <param name="cancellationToken">The cancellation token bounding the first read.</param>
     /// <returns>The attached session.</returns>
     /// <exception cref="OpenCodeTransportException">The server closed before attaching, sent another frame first, or negotiated a protocol this SDK does not speak.</exception>
     internal static Task<PersistentPtySession> AttachAsync(
         ITerminalWebSocket socket,
         string ptyId,
+        TimeProvider time,
         CancellationToken cancellationToken) =>
-        AttachAsync(socket, ptyId, TerminalSocketBounds.DefaultSendTimeout, cancellationToken);
+        AttachAsync(socket, ptyId, time, TerminalSocketBounds.DefaultSendTimeout, cancellationToken);
 
     /// <summary>Attaches with the fixed send budget captured from the connection options.</summary>
     /// <param name="socket">The upgraded socket whose ownership is transferred on success.</param>
     /// <param name="ptyId">The terminal identity used in failures.</param>
+    /// <param name="time">The client's clock for send and close deadlines.</param>
     /// <param name="sendTimeout">The fixed per-send budget.</param>
     /// <param name="cancellationToken">The token bounding attachment.</param>
     /// <returns>The attached session.</returns>
     internal static async Task<PersistentPtySession> AttachAsync(
         ITerminalWebSocket socket,
         string ptyId,
+        TimeProvider time,
         TimeSpan sendTimeout,
         CancellationToken cancellationToken)
     {
@@ -188,7 +192,8 @@ internal static async Task<PersistentPtySession> AttachAsync(
                 socket,
                 PersistentPtyFrameDecoder.Instance,
                 PersistentPtyClosePolicy.Instance,
-                typeof(PersistentPtySession))
+                typeof(PersistentPtySession),
+                time)
             {
                 SendTimeout = sendTimeout,
             };

src/OpenCode.Sdk/Ptys/PtyClient.cs

@@ -144,7 +144,7 @@ public virtual async Task<PtySession> ConnectAsync(PtyConnectOptions? options =
             await socket
                 .ConnectAsync(address, ptyId, PtyUpgradeFailurePolicy.Instance, cancellationToken)
                 .ConfigureAwait(false);
-            return new PtySession(socket, options?.SendTimeout);
+            return new PtySession(socket, connection.Time, options?.SendTimeout);
         }
         catch
         {

src/OpenCode.Sdk/Ptys/PtySession.cs

@@ -14,15 +14,16 @@ public class PtySession : IAsyncDisposable
 {
     private readonly TerminalSocketCore<PtyFrame>? _core;
 
-    internal PtySession(ITerminalWebSocket socket, TimeSpan? sendTimeout = null)
+    internal PtySession(ITerminalWebSocket socket, TimeProvider time, TimeSpan? sendTimeout = null)
     {
         ArgumentNullException.ThrowIfNull(socket);
 
         _core = new TerminalSocketCore<PtyFrame>(
             socket,
             PtyFrameDecoder.Instance,
             PtyClosePolicy.Instance,
-            typeof(PtySession))
+            typeof(PtySession),
+            time)
         {
             SendTimeout = sendTimeout ?? TerminalSocketBounds.DefaultSendTimeout,
         };

tests/OpenCode.Sdk.Performance.Tests/Benchmarks/PersistentPtySessionReadBenchmarks.cs

@@ -205,7 +205,7 @@ private static async Task<int> CountFramesAsync(PersistentPtySession session)
     private async Task<int> CountCoreFramesAsync(ITerminalFrameDecoder<PersistentPtyFrame> decoder)
     {
         using var socket = new CannedTerminalWebSocket(_messages);
-        await using var core = new TerminalSocketCore<PersistentPtyFrame>(socket, decoder, PersistentPtyClosePolicy.Instance, typeof(PersistentPtySession));
+        await using var core = NewCore(socket, decoder);
         var frames = 0;
         await foreach (var frame in core.ReadAsync(CancellationToken.None).ConfigureAwait(false))
         {
@@ -231,7 +231,11 @@ private static (WebSocketMessageType Type, byte[] Payload)[] MessagesFor(WireFix
         Cases.First(entry => string.Equals(entry.Fixture.Name, fixture.Name, StringComparison.Ordinal)).Messages;
 
     private static TerminalSocketCore<PersistentPtyFrame> NewCore(ITerminalWebSocket socket) =>
-        new(socket, PersistentPtyFrameDecoder.Instance, PersistentPtyClosePolicy.Instance, typeof(PersistentPtySession));
+        NewCore(socket, PersistentPtyFrameDecoder.Instance);
+
+    private static TerminalSocketCore<PersistentPtyFrame> NewCore(
+        ITerminalWebSocket socket, ITerminalFrameDecoder<PersistentPtyFrame> decoder) =>
+        new(socket, decoder, PersistentPtyClosePolicy.Instance, typeof(PersistentPtySession), TimeProvider.System);
 
     /// <summary>The single <c>attached</c> control frame, the size class the normal family's cursor case measures.</summary>
     private static (WireFixture Fixture, (WebSocketMessageType Type, byte[] Payload)[] Messages) AttachedCase()

tests/OpenCode.Sdk.Performance.Tests/Benchmarks/PtySessionReadBenchmarks.cs

@@ -60,7 +60,7 @@ public async Task SetupAsync()
     {
         _messages = MessagesFor(Fixture);
         _socket = new CannedTerminalWebSocket(_messages);
-        _session = new PtySession(_socket);
+        _session = new PtySession(_socket, TimeProvider.System);
 
         var delivered = await CollectFramesAsync(_session).ConfigureAwait(false);
         if (delivered.Count != Fixture.Items)
@@ -77,7 +77,7 @@ public async Task SetupAsync()
 
         using (var pooled = new CannedTerminalWebSocket(_messages, pooled: true))
         {
-            await using var session = new PtySession(pooled);
+            await using var session = new PtySession(pooled, TimeProvider.System);
             var pooledFrames = await CollectFramesAsync(session).ConfigureAwait(false);
             if (pooledFrames.Count != Fixture.Items || pooled.PooledReceives < Fixture.Items
                 || !pooledFrames.Select(static frame => frame.GetType()).SequenceEqual(delivered.Select(static frame => frame.GetType())))
@@ -109,7 +109,7 @@ public async Task SetupAsync()
     public async Task<int> ReadFramesAsync()
     {
         using var socket = new CannedTerminalWebSocket(_messages);
-        await using var session = new PtySession(socket);
+        await using var session = new PtySession(socket, TimeProvider.System);
         return await CountFramesAsync(session).ConfigureAwait(false);
     }
 
@@ -118,7 +118,7 @@ public async Task<int> ReadFramesAsync()
     public async Task<int> ReadFramesPooledAsync()
     {
         using var socket = new CannedTerminalWebSocket(_messages, pooled: true);
-        await using var session = new PtySession(socket);
+        await using var session = new PtySession(socket, TimeProvider.System);
         return await CountFramesAsync(session).ConfigureAwait(false);
     }
 
@@ -173,7 +173,7 @@ private static async Task<int> CountFramesAsync(PtySession session)
     private async Task<int> CountCoreFramesAsync(ITerminalFrameDecoder<PtyFrame> decoder)
     {
         using var socket = new CannedTerminalWebSocket(_messages);
-        await using var core = new TerminalSocketCore<PtyFrame>(socket, decoder, PtyClosePolicy.Instance, typeof(PtySession));
+        await using var core = new TerminalSocketCore<PtyFrame>(socket, decoder, PtyClosePolicy.Instance, typeof(PtySession), TimeProvider.System);
         var frames = 0;
         await foreach (var frame in core.ReadAsync(CancellationToken.None).ConfigureAwait(false))
         {

tests/OpenCode.Sdk.Performance.Tests/Benchmarks/TerminalWriteBenchmarks.cs

@@ -70,13 +70,13 @@ public async Task SetupAsync()
         }
 
         _ptySocket = new SendSinkTerminalWebSocket();
-        _pty = new PtySession(_ptySocket);
+        _pty = new PtySession(_ptySocket, TimeProvider.System);
         _persistentSocket = new SendSinkTerminalWebSocket();
         TerminalSocketCore<PersistentPtyFrame>? core = null;
         try
         {
             core = new TerminalSocketCore<PersistentPtyFrame>(
-                _persistentSocket, PersistentPtyFrameDecoder.Instance, PersistentPtyClosePolicy.Instance, typeof(PersistentPtySession));
+                _persistentSocket, PersistentPtyFrameDecoder.Instance, PersistentPtyClosePolicy.Instance, typeof(PersistentPtySession), TimeProvider.System);
             _persistent = new PersistentPtySession(core, Seed.Attachment());
             core = null;
         }