diff --git a/docs/architecture/client-runtime.md b/docs/architecture/client-runtime.md index 5774020d..4dba6f63 100644 --- a/docs/architecture/client-runtime.md +++ b/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` 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 diff --git a/src/OpenCode.Sdk/Internal/Terminal/ITerminalDeadlineFactory.cs b/src/OpenCode.Sdk/Internal/Terminal/ITerminalDeadlineFactory.cs deleted file mode 100644 index b9f28742..00000000 --- a/src/OpenCode.Sdk/Internal/Terminal/ITerminalDeadlineFactory.cs +++ /dev/null @@ -1,10 +0,0 @@ -namespace OpenCode.Sdk.Internal; - -/// Creates the connection's per-operation cancellation timers. -internal interface ITerminalDeadlineFactory -{ - /// Starts one fixed deadline; the operation owns the returned source. - /// The duration until cancellation. - /// The operation-owned cancellation source. - public CancellationTokenSource Create(TimeSpan timeout); -} diff --git a/src/OpenCode.Sdk/Internal/Terminal/TerminalCancellationSource.cs b/src/OpenCode.Sdk/Internal/Terminal/TerminalCancellationSource.cs deleted file mode 100644 index 7a26af8b..00000000 --- a/src/OpenCode.Sdk/Internal/Terminal/TerminalCancellationSource.cs +++ /dev/null @@ -1,17 +0,0 @@ -namespace OpenCode.Sdk.Internal; - -/// Owns terminal cancellation without blocking a worker while callbacks are scheduled. -internal sealed class TerminalCancellationSource : IDisposable -{ - private readonly CancellationTokenSource _source = new(); - - /// Gets the token observed by owned I/O or send admission. - public CancellationToken Token => _source.Token; - - /// Requests cancellation and observes all callbacks before completion. - /// Our lifetime owner prevents source disposal until this task and owned I/O finish. - public Task CancelAsync() => _source.CancelOnWorkerAsync(); - - /// Releases the source after cancellation callbacks and owned I/O have settled. - public void Dispose() => _source.Dispose(); -} diff --git a/src/OpenCode.Sdk/Internal/Terminal/TerminalDeadlineFactory.cs b/src/OpenCode.Sdk/Internal/Terminal/TerminalDeadlineFactory.cs deleted file mode 100644 index db1966df..00000000 --- a/src/OpenCode.Sdk/Internal/Terminal/TerminalDeadlineFactory.cs +++ /dev/null @@ -1,15 +0,0 @@ -namespace OpenCode.Sdk.Internal; - -/// Starts cancellation timers using the platform timer implementation. -internal sealed class TerminalDeadlineFactory : ITerminalDeadlineFactory -{ - private TerminalDeadlineFactory() - { - } - - /// Gets the shared timer factory. - public static TerminalDeadlineFactory Instance { get; } = new(); - - /// - public CancellationTokenSource Create(TimeSpan timeout) => new(timeout); -} diff --git a/src/OpenCode.Sdk/Internal/Terminal/TerminalReceivePump.cs b/src/OpenCode.Sdk/Internal/Terminal/TerminalReceivePump.cs index 321be20d..a7d2aa45 100644 --- a/src/OpenCode.Sdk/Internal/Terminal/TerminalReceivePump.cs +++ b/src/OpenCode.Sdk/Internal/Terminal/TerminalReceivePump.cs @@ -14,7 +14,7 @@ internal sealed class TerminalReceivePump : 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); diff --git a/src/OpenCode.Sdk/Internal/Terminal/TerminalSendDeadline.cs b/src/OpenCode.Sdk/Internal/Terminal/TerminalSendDeadline.cs index 45eb709e..2b2a1c38 100644 --- a/src/OpenCode.Sdk/Internal/Terminal/TerminalSendDeadline.cs +++ b/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 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. diff --git a/src/OpenCode.Sdk/Internal/Terminal/TerminalSendQueue.cs b/src/OpenCode.Sdk/Internal/Terminal/TerminalSendQueue.cs index d116bda3..a8b6081f 100644 --- a/src/OpenCode.Sdk/Internal/Terminal/TerminalSendQueue.cs +++ b/src/OpenCode.Sdk/Internal/Terminal/TerminalSendQueue.cs @@ -1,7 +1,7 @@ namespace OpenCode.Sdk.Internal; /// Serializes physical writes and graceful close behind one connection gate. -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 ( diff --git a/src/OpenCode.Sdk/Internal/Terminal/TerminalSocketCore.cs b/src/OpenCode.Sdk/Internal/Terminal/TerminalSocketCore.cs index 6261f254..e4e2a9c6 100644 --- a/src/OpenCode.Sdk/Internal/Terminal/TerminalSocketCore.cs +++ b/src/OpenCode.Sdk/Internal/Terminal/TerminalSocketCore.cs @@ -17,10 +17,11 @@ internal sealed class TerminalSocketCore : IAsyncDisposable private readonly TerminalReceivePump _receiver; private readonly Func _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,7 +33,8 @@ public TerminalSocketCore( ITerminalWebSocket socket, ITerminalFrameDecoder decoder, ITerminalClosePolicy closePolicy, - Type owner) + Type owner, + TimeProvider time) { ArgumentNullException.ThrowIfNull(socket); ArgumentNullException.ThrowIfNull(decoder); @@ -40,10 +42,11 @@ public TerminalSocketCore( ArgumentNullException.ThrowIfNull(owner); _socket = socket; - _sendGate = new TerminalSendQueue(socket); + _time = time; + _sendGate = new TerminalSendQueue(socket, time); _owner = owner; _sendEndedToken = _sendEnded.Token; - _receiver = new TerminalReceivePump(socket, decoder, closePolicy, _sendEnded.CancelAsync); + _receiver = new TerminalReceivePump(socket, decoder, closePolicy, _sendEnded.CancelOnWorkerAsync); _getOutcome = () => _receiver.Outcome; } @@ -53,9 +56,6 @@ public TerminalSocketCore( /// Gets the fixed per-send budget captured by the connection. public TimeSpan SendTimeout { get; init; } = TerminalSocketBounds.DefaultSendTimeout; - /// Gets the timer boundary used for per-send deadlines. - public ITerminalDeadlineFactory DeadlineFactory { get; init; } = TerminalDeadlineFactory.Instance; - /// /// 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(); } diff --git a/src/OpenCode.Sdk/Internal/Transport/ConnectionSnapshot.cs b/src/OpenCode.Sdk/Internal/Transport/ConnectionSnapshot.cs index d2122e94..7d9da26c 100644 --- a/src/OpenCode.Sdk/Internal/Transport/ConnectionSnapshot.cs +++ b/src/OpenCode.Sdk/Internal/Transport/ConnectionSnapshot.cs @@ -19,13 +19,14 @@ namespace OpenCode.Sdk.Internal; /// 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; } /// Gets the Basic credential every request carries, or null when no password was configured. @@ -36,4 +37,7 @@ public ConnectionSnapshot(string endpointBase, AuthenticationHeaderValue? author /// Gets the ambient location a per-call selector merges over member by member. public LocationSelector? Location { get; } + + /// Gets the client's clock for non-HTTP send and close deadlines. + public TimeProvider Time { get; } } diff --git a/src/OpenCode.Sdk/Internal/Transport/Pipeline.cs b/src/OpenCode.Sdk/Internal/Transport/Pipeline.cs index 92be2404..81d42784 100644 --- a/src/OpenCode.Sdk/Internal/Transport/Pipeline.cs +++ b/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); } /// Gets the construction-time connection facts a non-HTTP door addresses the server with. diff --git a/src/OpenCode.Sdk/PersistentPtys/PersistentPtyClient.cs b/src/OpenCode.Sdk/PersistentPtys/PersistentPtyClient.cs index 9e28f841..6a38920f 100644 --- a/src/OpenCode.Sdk/PersistentPtys/PersistentPtyClient.cs +++ b/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; diff --git a/src/OpenCode.Sdk/PersistentPtys/PersistentPtySession.cs b/src/OpenCode.Sdk/PersistentPtys/PersistentPtySession.cs index 91b60907..28835504 100644 --- a/src/OpenCode.Sdk/PersistentPtys/PersistentPtySession.cs +++ b/src/OpenCode.Sdk/PersistentPtys/PersistentPtySession.cs @@ -157,24 +157,28 @@ public virtual async ValueTask DisposeAsync() /// /// The upgraded socket; the session takes ownership of it. /// The terminal the socket addresses; it names the failures. + /// The client's clock for send and close deadlines. /// The cancellation token bounding the first read. /// The attached session. /// The server closed before attaching, sent another frame first, or negotiated a protocol this SDK does not speak. internal static Task AttachAsync( ITerminalWebSocket socket, string ptyId, + TimeProvider time, CancellationToken cancellationToken) => - AttachAsync(socket, ptyId, TerminalSocketBounds.DefaultSendTimeout, cancellationToken); + AttachAsync(socket, ptyId, time, TerminalSocketBounds.DefaultSendTimeout, cancellationToken); /// Attaches with the fixed send budget captured from the connection options. /// The upgraded socket whose ownership is transferred on success. /// The terminal identity used in failures. + /// The client's clock for send and close deadlines. /// The fixed per-send budget. /// The token bounding attachment. /// The attached session. internal static async Task AttachAsync( ITerminalWebSocket socket, string ptyId, + TimeProvider time, TimeSpan sendTimeout, CancellationToken cancellationToken) { @@ -188,7 +192,8 @@ internal static async Task AttachAsync( socket, PersistentPtyFrameDecoder.Instance, PersistentPtyClosePolicy.Instance, - typeof(PersistentPtySession)) + typeof(PersistentPtySession), + time) { SendTimeout = sendTimeout, }; diff --git a/src/OpenCode.Sdk/Ptys/PtyClient.cs b/src/OpenCode.Sdk/Ptys/PtyClient.cs index 7c4e5450..d6a4b4c9 100644 --- a/src/OpenCode.Sdk/Ptys/PtyClient.cs +++ b/src/OpenCode.Sdk/Ptys/PtyClient.cs @@ -144,7 +144,7 @@ public virtual async Task 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 { diff --git a/src/OpenCode.Sdk/Ptys/PtySession.cs b/src/OpenCode.Sdk/Ptys/PtySession.cs index fd6c5923..bca17dbc 100644 --- a/src/OpenCode.Sdk/Ptys/PtySession.cs +++ b/src/OpenCode.Sdk/Ptys/PtySession.cs @@ -14,7 +14,7 @@ public class PtySession : IAsyncDisposable { private readonly TerminalSocketCore? _core; - internal PtySession(ITerminalWebSocket socket, TimeSpan? sendTimeout = null) + internal PtySession(ITerminalWebSocket socket, TimeProvider time, TimeSpan? sendTimeout = null) { ArgumentNullException.ThrowIfNull(socket); @@ -22,7 +22,8 @@ internal PtySession(ITerminalWebSocket socket, TimeSpan? sendTimeout = null) socket, PtyFrameDecoder.Instance, PtyClosePolicy.Instance, - typeof(PtySession)) + typeof(PtySession), + time) { SendTimeout = sendTimeout ?? TerminalSocketBounds.DefaultSendTimeout, }; diff --git a/tests/OpenCode.Sdk.Performance.Tests/Benchmarks/PersistentPtySessionReadBenchmarks.cs b/tests/OpenCode.Sdk.Performance.Tests/Benchmarks/PersistentPtySessionReadBenchmarks.cs index 05c6c96a..7d2ef978 100644 --- a/tests/OpenCode.Sdk.Performance.Tests/Benchmarks/PersistentPtySessionReadBenchmarks.cs +++ b/tests/OpenCode.Sdk.Performance.Tests/Benchmarks/PersistentPtySessionReadBenchmarks.cs @@ -205,7 +205,7 @@ private static async Task CountFramesAsync(PersistentPtySession session) private async Task CountCoreFramesAsync(ITerminalFrameDecoder decoder) { using var socket = new CannedTerminalWebSocket(_messages); - await using var core = new TerminalSocketCore(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 NewCore(ITerminalWebSocket socket) => - new(socket, PersistentPtyFrameDecoder.Instance, PersistentPtyClosePolicy.Instance, typeof(PersistentPtySession)); + NewCore(socket, PersistentPtyFrameDecoder.Instance); + + private static TerminalSocketCore NewCore( + ITerminalWebSocket socket, ITerminalFrameDecoder decoder) => + new(socket, decoder, PersistentPtyClosePolicy.Instance, typeof(PersistentPtySession), TimeProvider.System); /// The single attached control frame, the size class the normal family's cursor case measures. private static (WireFixture Fixture, (WebSocketMessageType Type, byte[] Payload)[] Messages) AttachedCase() diff --git a/tests/OpenCode.Sdk.Performance.Tests/Benchmarks/PtySessionReadBenchmarks.cs b/tests/OpenCode.Sdk.Performance.Tests/Benchmarks/PtySessionReadBenchmarks.cs index 161d23c8..1a0e6d7d 100644 --- a/tests/OpenCode.Sdk.Performance.Tests/Benchmarks/PtySessionReadBenchmarks.cs +++ b/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 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 ReadFramesAsync() public async Task 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 CountFramesAsync(PtySession session) private async Task CountCoreFramesAsync(ITerminalFrameDecoder decoder) { using var socket = new CannedTerminalWebSocket(_messages); - await using var core = new TerminalSocketCore(socket, decoder, PtyClosePolicy.Instance, typeof(PtySession)); + await using var core = new TerminalSocketCore(socket, decoder, PtyClosePolicy.Instance, typeof(PtySession), TimeProvider.System); var frames = 0; await foreach (var frame in core.ReadAsync(CancellationToken.None).ConfigureAwait(false)) { diff --git a/tests/OpenCode.Sdk.Performance.Tests/Benchmarks/TerminalWriteBenchmarks.cs b/tests/OpenCode.Sdk.Performance.Tests/Benchmarks/TerminalWriteBenchmarks.cs index bf7922ed..2f872fb5 100644 --- a/tests/OpenCode.Sdk.Performance.Tests/Benchmarks/TerminalWriteBenchmarks.cs +++ b/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? core = null; try { core = new TerminalSocketCore( - _persistentSocket, PersistentPtyFrameDecoder.Instance, PersistentPtyClosePolicy.Instance, typeof(PersistentPtySession)); + _persistentSocket, PersistentPtyFrameDecoder.Instance, PersistentPtyClosePolicy.Instance, typeof(PersistentPtySession), TimeProvider.System); _persistent = new PersistentPtySession(core, Seed.Attachment()); core = null; } diff --git a/tests/OpenCode.Sdk.Tests/CancellationIdentityTests.cs b/tests/OpenCode.Sdk.Tests/CancellationIdentityTests.cs index 26cef063..4e94991f 100644 --- a/tests/OpenCode.Sdk.Tests/CancellationIdentityTests.cs +++ b/tests/OpenCode.Sdk.Tests/CancellationIdentityTests.cs @@ -304,7 +304,7 @@ private static async Task AtAttachAsync(Cancellation using var socket = new ScriptedTerminalWebSocket().Parking(); return await CancelAtParkedReadAsync(async () => { - await using var session = await PersistentPtySession.AttachAsync(socket, PtyId, caller.Token); + await using var session = await PersistentPtySession.AttachAsync(socket, PtyId, new ManualTimeProvider(), caller.Token); }, socket, caller); } @@ -326,12 +326,12 @@ private static async Task AtTerminalReadAsync(bool p if (persistent) { _ = socket.Text(PersistentPtyFrameData.AttachedJson).Parking(); - await using var session = await PersistentPtySession.AttachAsync(socket, PtyId, CancellationToken.None); + await using var session = await PersistentPtySession.AttachAsync(socket, PtyId, new ManualTimeProvider(), CancellationToken.None); return await AtParkedReadAsync(session.ReadAsync(MethodToken(entry, caller.Token)), entry, socket, caller); } _ = socket.Parking(); - await using var normal = new PtySession(socket); + await using var normal = new PtySession(socket, new ManualTimeProvider()); return await AtParkedReadAsync(normal.ReadAsync(MethodToken(entry, caller.Token)), entry, socket, caller); } @@ -366,7 +366,7 @@ private static async Task AtTerminalSendAsync(string if (row.StartsWith("persistent-", StringComparison.Ordinal)) { _ = socket.Text(PersistentPtyFrameData.AttachedJson); - await using var session = await PersistentPtySession.AttachAsync(socket, PtyId, CancellationToken.None); + await using var session = await PersistentPtySession.AttachAsync(socket, PtyId, new ManualTimeProvider(), CancellationToken.None); var failure = await AtSendGateAsync(token => row switch { "persistent-write" => session.WriteAsync(PersistentPtyFrameData.Output(Input), token), @@ -384,7 +384,7 @@ await Assert.That(socket.SentMessages[^1]).IsEquivalentTo( return failure; } - await using var normal = new PtySession(socket); + await using var normal = new PtySession(socket, new ManualTimeProvider()); return await AtSendGateAsync(token => row == "pty-write" ? normal.WriteAsync(Input, token) : normal.SubmitAsync(Input, token), queued ? () => normal.WriteAsync(Input, CancellationToken.None) : null, socket, caller); diff --git a/tests/OpenCode.Sdk.Tests/PersistentPtys/PersistentPtyConnectUriBuilderTests.cs b/tests/OpenCode.Sdk.Tests/PersistentPtys/PersistentPtyConnectUriBuilderTests.cs index 4a1b2cfa..ec312ae4 100644 --- a/tests/OpenCode.Sdk.Tests/PersistentPtys/PersistentPtyConnectUriBuilderTests.cs +++ b/tests/OpenCode.Sdk.Tests/PersistentPtys/PersistentPtyConnectUriBuilderTests.cs @@ -1,4 +1,5 @@ using OpenCode.Sdk.Internal; +using OpenCode.Sdk.Tests.Support; namespace OpenCode.Sdk.Tests.PersistentPtys; @@ -53,5 +54,5 @@ public async Task Build_Should_Refuse_A_Dot_Segment_Route_Value() } private static ConnectionSnapshot Snapshot(string endpoint) => - new(EndpointPolicy.Normalize(new Uri(endpoint)), authorization: null, location: null); + new(EndpointPolicy.Normalize(new Uri(endpoint)), authorization: null, location: null, new ManualTimeProvider()); } diff --git a/tests/OpenCode.Sdk.Tests/PersistentPtys/PersistentPtySessionTests.cs b/tests/OpenCode.Sdk.Tests/PersistentPtys/PersistentPtySessionTests.cs index 31d5d884..43d17a7e 100644 --- a/tests/OpenCode.Sdk.Tests/PersistentPtys/PersistentPtySessionTests.cs +++ b/tests/OpenCode.Sdk.Tests/PersistentPtys/PersistentPtySessionTests.cs @@ -15,7 +15,7 @@ public async Task AttachAsync_Should_Release_The_Receiver_When_Canceled_Before_A using var cancellation = new CancellationTokenSource(); // The finally below cancels and awaits this attachment before the source is disposed. #pragma warning disable CA2025 // The analyzer does not recognize this Task join in finally. - var pending = PersistentPtySession.AttachAsync(socket, PtyId, cancellation.Token); + var pending = PersistentPtySession.AttachAsync(socket, PtyId, new ManualTimeProvider(), cancellation.Token); #pragma warning restore CA2025 OperationCanceledException? failure = null; try @@ -46,7 +46,7 @@ public async Task AttachAsync_Should_Detach_The_Connect_Token_After_Returning_Th var socket = new ScriptedTerminalWebSocket().Text(PersistentPtyFrameData.AttachedJson) .Pausing().Binary(PersistentPtyFrameData.Output("after attach")).Closing(WebSocketCloseStatus.NormalClosure); using var cancellation = new CancellationTokenSource(); - await using var session = await PersistentPtySession.AttachAsync(socket, PtyId, cancellation.Token); + await using var session = await PersistentPtySession.AttachAsync(socket, PtyId, new ManualTimeProvider(), cancellation.Token); await cancellation.CancelOnWorkerAsync(); await session.WriteAsync(PersistentPtyFrameData.Output("input")); socket.ReleaseReceives(); @@ -64,7 +64,7 @@ public async Task ResizeAsync_Should_Use_The_Connections_Configured_Send_Timer() { var socket = new ScriptedTerminalWebSocket().Text(PersistentPtyFrameData.AttachedJson).GatingSends(); await using var session = await PersistentPtySession.AttachAsync( - socket, PtyId, TimeSpan.FromMilliseconds(50), CancellationToken.None); + socket, PtyId, TimeProvider.System, TimeSpan.FromMilliseconds(50), CancellationToken.None); using var watchdog = new CancellationTokenSource(TimeSpan.FromSeconds(5)); var failure = await Assert.That(async () => await session.ResizeAsync(100, 30, watchdog.Token)) @@ -139,7 +139,7 @@ public async Task AttachAsync_Should_Consume_The_Attached_Frame_And_Expose_The_A .Text(PersistentPtyFrameData.AttachedJson) .Closing(WebSocketCloseStatus.NormalClosure); - await using var session = await PersistentPtySession.AttachAsync(socket, PtyId, CancellationToken.None); + await using var session = await PersistentPtySession.AttachAsync(socket, PtyId, new ManualTimeProvider(), CancellationToken.None); await Assert.That(session.Attachment.AttachmentId).IsEqualTo("att_1"); await Assert.That(session.Attachment.Role).IsEqualTo(PersistentPtyRole.Controller); @@ -154,7 +154,7 @@ public async Task AttachAsync_Should_Refuse_A_Terminal_Unavailable_Close_Before_ var socket = new ScriptedTerminalWebSocket().Closing(TerminalUnavailable, PersistentPtyFrameData.TerminalUnavailableReason); var failure = await Assert - .That(async () => _ = await PersistentPtySession.AttachAsync(socket, PtyId, CancellationToken.None)) + .That(async () => _ = await PersistentPtySession.AttachAsync(socket, PtyId, new ManualTimeProvider(), CancellationToken.None)) .Throws(); await Assert.That(failure!.Message).Contains("4404"); @@ -167,7 +167,7 @@ public async Task AttachAsync_Should_Refuse_A_Raw_Input_Protocol() var socket = new ScriptedTerminalWebSocket().Text(PersistentPtyFrameData.AttachedRawProtocolJson); var failure = await Assert - .That(async () => _ = await PersistentPtySession.AttachAsync(socket, PtyId, CancellationToken.None)) + .That(async () => _ = await PersistentPtySession.AttachAsync(socket, PtyId, new ManualTimeProvider(), CancellationToken.None)) .Throws(); await Assert.That(failure!.Message).Contains("input protocol"); @@ -185,7 +185,7 @@ public async Task AttachAsync_Should_Refuse_A_First_Frame_That_Is_Not_Attached() var socket = new ScriptedTerminalWebSocket().Text(PersistentPtyFrameData.ReplayCompleteJson); var failure = await Assert - .That(async () => _ = await PersistentPtySession.AttachAsync(socket, PtyId, CancellationToken.None)) + .That(async () => _ = await PersistentPtySession.AttachAsync(socket, PtyId, new ManualTimeProvider(), CancellationToken.None)) .Throws(); await Assert.That(failure!.Message).Contains("attached"); @@ -202,7 +202,7 @@ public async Task AttachAsync_Should_Name_The_Frame_Kind_When_The_Attachment_Id_ var socket = new ScriptedTerminalWebSocket().Text(PersistentPtyFrameData.AttachedNullAttachmentIdJson); var failure = await Assert - .That(async () => _ = await PersistentPtySession.AttachAsync(socket, PtyId, CancellationToken.None)) + .That(async () => _ = await PersistentPtySession.AttachAsync(socket, PtyId, new ManualTimeProvider(), CancellationToken.None)) .Throws(); // A wire null for a promised string is a member-level failure of a frame whose type the @@ -218,7 +218,7 @@ public async Task AttachAsync_Should_Name_The_Frame_Kind_When_The_Role_Is_Null() var socket = new ScriptedTerminalWebSocket().Text(PersistentPtyFrameData.AttachedNullRoleJson); var failure = await Assert - .That(async () => _ = await PersistentPtySession.AttachAsync(socket, PtyId, CancellationToken.None)) + .That(async () => _ = await PersistentPtySession.AttachAsync(socket, PtyId, new ManualTimeProvider(), CancellationToken.None)) .Throws(); await Assert.That(failure!.Message).Contains("'attached'"); @@ -231,7 +231,7 @@ public async Task AttachAsync_Should_Name_The_Frame_Kind_When_The_Role_Is_Not_Re var socket = new ScriptedTerminalWebSocket().Text(PersistentPtyFrameData.AttachedUnknownRoleJson); var failure = await Assert - .That(async () => _ = await PersistentPtySession.AttachAsync(socket, PtyId, CancellationToken.None)) + .That(async () => _ = await PersistentPtySession.AttachAsync(socket, PtyId, new ManualTimeProvider(), CancellationToken.None)) .Throws(); // The wire spells 'controller' and 'observer' and nothing else, so a third spelling is a @@ -246,7 +246,7 @@ public async Task AttachAsync_Should_Refuse_A_Normal_Close_Before_Attached() var socket = new ScriptedTerminalWebSocket().Closing(WebSocketCloseStatus.NormalClosure); var failure = await Assert - .That(async () => _ = await PersistentPtySession.AttachAsync(socket, PtyId, CancellationToken.None)) + .That(async () => _ = await PersistentPtySession.AttachAsync(socket, PtyId, new ManualTimeProvider(), CancellationToken.None)) .Throws(); await Assert.That(failure!.Message).Contains("attached"); @@ -257,7 +257,7 @@ public async Task AttachAsync_Should_Expose_A_Truncated_Observer_Replay() { var socket = new ScriptedTerminalWebSocket().Text(PersistentPtyFrameData.AttachedObserverJson); - await using var session = await PersistentPtySession.AttachAsync(socket, PtyId, CancellationToken.None); + await using var session = await PersistentPtySession.AttachAsync(socket, PtyId, new ManualTimeProvider(), CancellationToken.None); await Assert.That(session.Attachment.Role).IsEqualTo(PersistentPtyRole.Observer); await Assert.That(session.Attachment.Replay.Truncated).IsTrue(); @@ -275,7 +275,7 @@ public async Task ReadAsync_Should_Yield_Output_As_Bytes_And_The_Replay_Bracket_ .Text(PersistentPtyFrameData.ReplayCompleteJson) .Binary(PersistentPtyFrameData.Output("hello\n")) .Closing(WebSocketCloseStatus.NormalClosure); - await using var session = await PersistentPtySession.AttachAsync(socket, PtyId, CancellationToken.None); + await using var session = await PersistentPtySession.AttachAsync(socket, PtyId, new ManualTimeProvider(), CancellationToken.None); var frames = await ReadAllAsync(session); @@ -296,7 +296,7 @@ public async Task ReadAsync_Should_Decode_Each_Control_Frame_Kind( .Text(PersistentPtyFrameData.AttachedJson) .Text(control.Control) .Closing(WebSocketCloseStatus.NormalClosure); - await using var session = await PersistentPtySession.AttachAsync(socket, PtyId, CancellationToken.None); + await using var session = await PersistentPtySession.AttachAsync(socket, PtyId, new ManualTimeProvider(), CancellationToken.None); var frames = await ReadAllAsync(session); @@ -310,7 +310,7 @@ public async Task ReadAsync_Should_Yield_An_Unknown_Control_Type_As_An_Unknown_F .Text(PersistentPtyFrameData.AttachedJson) .Text(PersistentPtyFrameData.UnknownTypeJson) .Closing(WebSocketCloseStatus.NormalClosure); - await using var session = await PersistentPtySession.AttachAsync(socket, PtyId, CancellationToken.None); + await using var session = await PersistentPtySession.AttachAsync(socket, PtyId, new ManualTimeProvider(), CancellationToken.None); var frames = await ReadAllAsync(session); @@ -325,7 +325,7 @@ public async Task ReadAsync_Should_Refuse_A_Control_Frame_Without_A_Type() var socket = new ScriptedTerminalWebSocket() .Text(PersistentPtyFrameData.AttachedJson) .Text(PersistentPtyFrameData.TypelessJson); - await using var session = await PersistentPtySession.AttachAsync(socket, PtyId, CancellationToken.None); + await using var session = await PersistentPtySession.AttachAsync(socket, PtyId, new ManualTimeProvider(), CancellationToken.None); var failure = await Assert.That(async () => _ = await ReadAllAsync(session)).Throws(); @@ -338,7 +338,7 @@ public async Task ReadAsync_Should_Name_The_Frame_Kind_When_A_Typed_Control_Fram var socket = new ScriptedTerminalWebSocket() .Text(PersistentPtyFrameData.AttachedJson) .Text(PersistentPtyFrameData.ResizedWithoutColsJson); - await using var session = await PersistentPtySession.AttachAsync(socket, PtyId, CancellationToken.None); + await using var session = await PersistentPtySession.AttachAsync(socket, PtyId, new ManualTimeProvider(), CancellationToken.None); var failure = await Assert.That(async () => _ = await ReadAllAsync(session)).Throws(); @@ -355,7 +355,7 @@ public async Task ReadAsync_Should_Name_The_Frame_Kind_When_A_Title_Changed_Fram var socket = new ScriptedTerminalWebSocket() .Text(PersistentPtyFrameData.AttachedJson) .Text(PersistentPtyFrameData.TitleChangedNullTitleJson); - await using var session = await PersistentPtySession.AttachAsync(socket, PtyId, CancellationToken.None); + await using var session = await PersistentPtySession.AttachAsync(socket, PtyId, new ManualTimeProvider(), CancellationToken.None); var failure = await Assert.That(async () => _ = await ReadAllAsync(session)).Throws(); @@ -368,7 +368,7 @@ public async Task ReadAsync_Should_Refuse_Truncated_Control_Json() var socket = new ScriptedTerminalWebSocket() .Text(PersistentPtyFrameData.AttachedJson) .Text(PersistentPtyFrameData.TruncatedJson); - await using var session = await PersistentPtySession.AttachAsync(socket, PtyId, CancellationToken.None); + await using var session = await PersistentPtySession.AttachAsync(socket, PtyId, new ManualTimeProvider(), CancellationToken.None); var failure = await Assert.That(async () => _ = await ReadAllAsync(session)).Throws(); @@ -383,7 +383,7 @@ public async Task ReadAsync_Should_Assemble_A_Fragmented_Output_Message_Once() .Text(PersistentPtyFrameData.AttachedJson) .BinaryFragments(output, splitAt: 5) .Closing(WebSocketCloseStatus.NormalClosure); - await using var session = await PersistentPtySession.AttachAsync(socket, PtyId, CancellationToken.None); + await using var session = await PersistentPtySession.AttachAsync(socket, PtyId, new ManualTimeProvider(), CancellationToken.None); var frames = await ReadAllAsync(session); @@ -397,7 +397,7 @@ public async Task ReadAsync_Should_Assemble_A_Fragmented_Control_Frame_Once() .Text(PersistentPtyFrameData.AttachedJson) .TextFragments(PersistentPtyFrameData.ResizedJson[..20], PersistentPtyFrameData.ResizedJson[20..]) .Closing(WebSocketCloseStatus.NormalClosure); - await using var session = await PersistentPtySession.AttachAsync(socket, PtyId, CancellationToken.None); + await using var session = await PersistentPtySession.AttachAsync(socket, PtyId, new ManualTimeProvider(), CancellationToken.None); var frames = await ReadAllAsync(session); @@ -413,7 +413,7 @@ public async Task ReadAsync_Should_Refuse_A_Terminal_Unavailable_Close_Mid_Strea .Text(PersistentPtyFrameData.AttachedJson) .Binary(PersistentPtyFrameData.Output("live")) .Closing(TerminalUnavailable, PersistentPtyFrameData.TerminalUnavailableReason); - await using var session = await PersistentPtySession.AttachAsync(socket, PtyId, CancellationToken.None); + await using var session = await PersistentPtySession.AttachAsync(socket, PtyId, new ManualTimeProvider(), CancellationToken.None); var frames = new List(); var failure = await Assert.That(async () => await ReadIntoAsync(session, frames)) @@ -430,7 +430,7 @@ public async Task ReadAsync_Should_Refuse_An_Abnormal_Close() var socket = new ScriptedTerminalWebSocket() .Text(PersistentPtyFrameData.AttachedJson) .Closing(WebSocketCloseStatus.InternalServerError); - await using var session = await PersistentPtySession.AttachAsync(socket, PtyId, CancellationToken.None); + await using var session = await PersistentPtySession.AttachAsync(socket, PtyId, new ManualTimeProvider(), CancellationToken.None); var failure = await Assert.That(async () => _ = await ReadAllAsync(session)).Throws(); @@ -444,7 +444,7 @@ public async Task ReadAsync_Should_Preserve_The_Local_Viewport_When_The_Server_R .Text(PersistentPtyFrameData.AttachedJson) .Text(PersistentPtyFrameData.ResizedJson) .Parking(); - await using var session = await PersistentPtySession.AttachAsync(socket, PtyId, CancellationToken.None); + await using var session = await PersistentPtySession.AttachAsync(socket, PtyId, new ManualTimeProvider(), CancellationToken.None); await using var frames = session.ReadAsync().GetAsyncEnumerator(); await Assert.That(await frames.MoveNextAsync()).IsTrue(); var resized = (PersistentPtyResizedFrame)frames.Current; @@ -465,7 +465,7 @@ public async Task WriteAsync_Should_Use_The_Resize_That_Completes_Before_Its_Soc .Text(PersistentPtyFrameData.ResizedJson) .Parking() .GatingSends(); - await using var session = await PersistentPtySession.AttachAsync(socket, PtyId, CancellationToken.None); + await using var session = await PersistentPtySession.AttachAsync(socket, PtyId, new ManualTimeProvider(), CancellationToken.None); var resize = session.ResizeAsync(100, 30); await socket.SendEntered; var write = session.WriteAsync(PersistentPtyFrameData.Output("ls\n")); @@ -490,7 +490,7 @@ public async Task WriteAsync_Should_Use_The_Resize_That_Completes_Before_Its_Soc public async Task WriteAsync_Should_Send_A_Framed_Binary_Input_Carrying_The_Attached_Viewport() { var socket = new ScriptedTerminalWebSocket().Text(PersistentPtyFrameData.AttachedJson); - await using var session = await PersistentPtySession.AttachAsync(socket, PtyId, CancellationToken.None); + await using var session = await PersistentPtySession.AttachAsync(socket, PtyId, new ManualTimeProvider(), CancellationToken.None); await session.WriteAsync(PersistentPtyFrameData.Output("ls\n")); @@ -503,7 +503,7 @@ await Assert.That(socket.SentMessages.Single()) public async Task WriteAsync_Should_Frame_At_The_Viewport_The_Server_Reported() { var socket = new ScriptedTerminalWebSocket().Text(PersistentPtyFrameData.AttachedWideViewportJson); - await using var session = await PersistentPtySession.AttachAsync(socket, PtyId, CancellationToken.None); + await using var session = await PersistentPtySession.AttachAsync(socket, PtyId, new ManualTimeProvider(), CancellationToken.None); await session.WriteAsync(PersistentPtyFrameData.Output("ls\n")); @@ -518,7 +518,7 @@ await Assert.That(socket.SentMessages.Single()) public async Task WriteAsync_Should_Throw_After_Dispose() { var socket = new ScriptedTerminalWebSocket().Text(PersistentPtyFrameData.AttachedJson); - var session = await PersistentPtySession.AttachAsync(socket, PtyId, CancellationToken.None); + var session = await PersistentPtySession.AttachAsync(socket, PtyId, new ManualTimeProvider(), CancellationToken.None); await session.DisposeAsync(); _ = await Assert.That(async () => await session.WriteAsync(PersistentPtyFrameData.Output("late"))) @@ -543,7 +543,7 @@ public static IEnumerable> LinesCarryingABreak() => public async Task SubmitAsync_Should_Send_A_Framed_Line_Terminated_By_Enter() { var socket = new ScriptedTerminalWebSocket().Text(PersistentPtyFrameData.AttachedJson); - await using var session = await PersistentPtySession.AttachAsync(socket, PtyId, CancellationToken.None); + await using var session = await PersistentPtySession.AttachAsync(socket, PtyId, new ManualTimeProvider(), CancellationToken.None); await session.SubmitAsync(SubmittedLine); @@ -556,7 +556,7 @@ await Assert.That(socket.SentMessages.Single()) public async Task SubmitAsync_Should_Frame_At_The_Viewport_A_Resize_Recorded() { var socket = new ScriptedTerminalWebSocket().Text(PersistentPtyFrameData.AttachedJson); - await using var session = await PersistentPtySession.AttachAsync(socket, PtyId, CancellationToken.None); + await using var session = await PersistentPtySession.AttachAsync(socket, PtyId, new ManualTimeProvider(), CancellationToken.None); await session.ResizeAsync(100, 30); await session.SubmitAsync(SubmittedLine); @@ -569,7 +569,7 @@ await Assert.That(socket.SentMessages[1]) public async Task SubmitAsync_Should_Send_A_Bare_Enter_For_An_Empty_Line() { var socket = new ScriptedTerminalWebSocket().Text(PersistentPtyFrameData.AttachedJson); - await using var session = await PersistentPtySession.AttachAsync(socket, PtyId, CancellationToken.None); + await using var session = await PersistentPtySession.AttachAsync(socket, PtyId, new ManualTimeProvider(), CancellationToken.None); await session.SubmitAsync(string.Empty); @@ -581,7 +581,7 @@ await Assert.That(socket.SentMessages.Single()) public async Task SubmitAsync_Should_Refuse_A_Null_Line() { var socket = new ScriptedTerminalWebSocket().Text(PersistentPtyFrameData.AttachedJson); - await using var session = await PersistentPtySession.AttachAsync(socket, PtyId, CancellationToken.None); + await using var session = await PersistentPtySession.AttachAsync(socket, PtyId, new ManualTimeProvider(), CancellationToken.None); _ = await Assert.That(async () => await session.SubmitAsync(null!)).Throws(); } @@ -591,7 +591,7 @@ public async Task SubmitAsync_Should_Refuse_A_Null_Line() public async Task SubmitAsync_Should_Refuse_A_Line_Carrying_A_Break(string line) { var socket = new ScriptedTerminalWebSocket().Text(PersistentPtyFrameData.AttachedJson); - await using var session = await PersistentPtySession.AttachAsync(socket, PtyId, CancellationToken.None); + await using var session = await PersistentPtySession.AttachAsync(socket, PtyId, new ManualTimeProvider(), CancellationToken.None); var failure = await Assert.That(async () => await session.SubmitAsync(line)).Throws(); @@ -604,7 +604,7 @@ public async Task SubmitAsync_Should_Refuse_A_Line_Carrying_A_Break(string line) public async Task SubmitAsync_Should_Throw_After_Dispose() { var socket = new ScriptedTerminalWebSocket().Text(PersistentPtyFrameData.AttachedJson); - var session = await PersistentPtySession.AttachAsync(socket, PtyId, CancellationToken.None); + var session = await PersistentPtySession.AttachAsync(socket, PtyId, new ManualTimeProvider(), CancellationToken.None); await session.DisposeAsync(); _ = await Assert.That(async () => await session.SubmitAsync(SubmittedLine)).Throws(); @@ -613,10 +613,8 @@ public async Task SubmitAsync_Should_Throw_After_Dispose() [Test] public async Task SubmitAsync_Should_Report_A_Failed_Send_As_A_Transport_Failure() { - var socket = new ScriptedTerminalWebSocket() - .Text(PersistentPtyFrameData.AttachedJson) - .FailingNextSendWith(new WebSocketException("the connection dropped")); - await using var session = await PersistentPtySession.AttachAsync(socket, PtyId, CancellationToken.None); + var socket = FailingSendSocket(); + await using var session = await PersistentPtySession.AttachAsync(socket, PtyId, new ManualTimeProvider(), CancellationToken.None); var failure = await Assert.That(async () => await session.SubmitAsync(SubmittedLine)) .Throws(); @@ -629,7 +627,7 @@ public async Task SubmitAsync_Should_Report_A_Failed_Send_As_A_Transport_Failure public async Task ResizeAsync_Should_Send_A_Control_Frame_And_Track_The_Viewport() { var socket = new ScriptedTerminalWebSocket().Text(PersistentPtyFrameData.AttachedJson); - await using var session = await PersistentPtySession.AttachAsync(socket, PtyId, CancellationToken.None); + await using var session = await PersistentPtySession.AttachAsync(socket, PtyId, new ManualTimeProvider(), CancellationToken.None); await session.ResizeAsync(100, 30); await session.WriteAsync(PersistentPtyFrameData.Output("ls\n")); @@ -647,7 +645,7 @@ private static async Task AssertResizedFramesAsync(ScriptedTerminalWebSocket soc public async Task ResizeAsync_Should_Refuse_A_Zero_Or_Oversized_Dimension() { var socket = new ScriptedTerminalWebSocket().Text(PersistentPtyFrameData.AttachedJson); - await using var session = await PersistentPtySession.AttachAsync(socket, PtyId, CancellationToken.None); + await using var session = await PersistentPtySession.AttachAsync(socket, PtyId, new ManualTimeProvider(), CancellationToken.None); _ = await Assert.That(async () => await session.ResizeAsync(0, 24)).Throws(); _ = await Assert.That(async () => await session.ResizeAsync(80, 0)).Throws(); @@ -661,10 +659,8 @@ public async Task ResizeAsync_Should_Refuse_A_Zero_Or_Oversized_Dimension() [Test] public async Task ResizeAsync_Should_End_The_Connection_When_The_Control_Send_Fails() { - var socket = new ScriptedTerminalWebSocket() - .Text(PersistentPtyFrameData.AttachedJson) - .FailingNextSendWith(new WebSocketException("the connection dropped")); - await using var session = await PersistentPtySession.AttachAsync(socket, PtyId, CancellationToken.None); + var socket = FailingSendSocket(); + await using var session = await PersistentPtySession.AttachAsync(socket, PtyId, new ManualTimeProvider(), CancellationToken.None); _ = await Assert.That(async () => await session.ResizeAsync(100, 30)).Throws(); _ = await Assert.That(async () => await session.WriteAsync(PersistentPtyFrameData.Output("ls\n"))) @@ -677,7 +673,7 @@ public async Task ResizeAsync_Should_End_The_Connection_When_The_Control_Send_Fa public async Task ResizeAsync_Should_Preserve_The_Viewport_When_Canceled_Before_Socket_Entry() { var socket = new ScriptedTerminalWebSocket().Text(PersistentPtyFrameData.AttachedJson); - await using var session = await PersistentPtySession.AttachAsync(socket, PtyId, CancellationToken.None); + await using var session = await PersistentPtySession.AttachAsync(socket, PtyId, new ManualTimeProvider(), CancellationToken.None); using var cancellation = new CancellationTokenSource(); await cancellation.CancelOnWorkerAsync(); @@ -697,7 +693,7 @@ public async Task ReadAsync_Should_Refuse_A_Second_Concurrent_Enumeration() .Text(PersistentPtyFrameData.AttachedJson) .Binary(PersistentPtyFrameData.Output("live")) .Parking(); - await using var session = await PersistentPtySession.AttachAsync(socket, PtyId, CancellationToken.None); + await using var session = await PersistentPtySession.AttachAsync(socket, PtyId, new ManualTimeProvider(), CancellationToken.None); var first = session.ReadAsync().GetAsyncEnumerator(CancellationToken.None); await using var enumeration = first.ConfigureAwait(false); await Assert.That(await first.MoveNextAsync()).IsTrue(); @@ -761,6 +757,11 @@ public async Task Frame_Constructors_Should_Be_Reachable_Without_Friend_Access() await Assert.That(unreachable).IsEmpty(); } + private static ScriptedTerminalWebSocket FailingSendSocket() => + new ScriptedTerminalWebSocket() + .Text(PersistentPtyFrameData.AttachedJson) + .FailingNextSendWith(new WebSocketException("the connection dropped")); + private static async Task> ReadAllAsync(PersistentPtySession session) { var frames = new List(); diff --git a/tests/OpenCode.Sdk.Tests/Ptys/PtyConnectUriBuilderTests.cs b/tests/OpenCode.Sdk.Tests/Ptys/PtyConnectUriBuilderTests.cs index b8d48716..0e240b5f 100644 --- a/tests/OpenCode.Sdk.Tests/Ptys/PtyConnectUriBuilderTests.cs +++ b/tests/OpenCode.Sdk.Tests/Ptys/PtyConnectUriBuilderTests.cs @@ -1,4 +1,5 @@ using OpenCode.Sdk.Internal; +using OpenCode.Sdk.Tests.Support; namespace OpenCode.Sdk.Tests.Ptys; @@ -104,5 +105,5 @@ public async Task Build_Should_Refuse_A_Dot_Segment_Route_Value() } private static ConnectionSnapshot Snapshot(string endpoint, LocationSelector? location = null) => - new(EndpointPolicy.Normalize(new Uri(endpoint)), authorization: null, location); + new(EndpointPolicy.Normalize(new Uri(endpoint)), authorization: null, location, new ManualTimeProvider()); } diff --git a/tests/OpenCode.Sdk.Tests/Ptys/PtySessionTests.cs b/tests/OpenCode.Sdk.Tests/Ptys/PtySessionTests.cs index 9bb1ad21..e276165e 100644 --- a/tests/OpenCode.Sdk.Tests/Ptys/PtySessionTests.cs +++ b/tests/OpenCode.Sdk.Tests/Ptys/PtySessionTests.cs @@ -14,7 +14,7 @@ public sealed class PtySessionTests public async Task WriteAsync_Should_Use_The_Connections_Configured_Send_Timer() { var socket = new ScriptedTerminalWebSocket().GatingSends(); - await using var session = new PtySession(socket, TimeSpan.FromMilliseconds(50)); + await using var session = new PtySession(socket, TimeProvider.System, TimeSpan.FromMilliseconds(50)); using var watchdog = new CancellationTokenSource(TimeSpan.FromSeconds(5)); var failure = await Assert.That(async () => await session.WriteAsync("input", watchdog.Token)) @@ -43,7 +43,7 @@ public async Task ReadAsync_Should_Yield_The_Replay_Chunks_In_Order() .Text("chunk-two") .Text("chunk-three") .Closing(WebSocketCloseStatus.NormalClosure); - await using var session = new PtySession(socket); + await using var session = new PtySession(socket, new ManualTimeProvider()); var frames = await ReadAllAsync(session); @@ -58,7 +58,7 @@ public async Task ReadAsync_Should_Read_The_Control_Frame_As_The_Exact_Cursor() .Text("replayed") .Binary(PtyFrameData.ControlFrame(PtyFrameData.CursorControlJson)) .Closing(WebSocketCloseStatus.NormalClosure); - await using var session = new PtySession(socket); + await using var session = new PtySession(socket, new ManualTimeProvider()); var frames = await ReadAllAsync(session); @@ -72,7 +72,7 @@ public async Task ReadAsync_Should_Read_The_Control_Frame_As_The_Exact_Cursor() public async Task ReadAsync_Should_Refuse_A_Malformed_Control_Frame(string body) { var socket = new ScriptedTerminalWebSocket().Binary(PtyFrameData.ControlFrame(body)); - await using var session = new PtySession(socket); + await using var session = new PtySession(socket, new ManualTimeProvider()); var failure = await Assert.That(async () => _ = await ReadAllAsync(session)) .Throws(); @@ -86,7 +86,7 @@ public async Task ReadAsync_Should_Assemble_A_Fragmented_Text_Message_Once() var socket = new ScriptedTerminalWebSocket() .TextFragments("frag-", "men", "ted") .Closing(WebSocketCloseStatus.NormalClosure); - await using var session = new PtySession(socket); + await using var session = new PtySession(socket, new ManualTimeProvider()); var frames = await ReadAllAsync(session); @@ -101,7 +101,7 @@ public async Task ReadAsync_Should_Assemble_A_Fragmented_Control_Frame_Once() var socket = new ScriptedTerminalWebSocket() .BinaryFragments(frame, splitAt: 4) .Closing(WebSocketCloseStatus.NormalClosure); - await using var session = new PtySession(socket); + await using var session = new PtySession(socket, new ManualTimeProvider()); var frames = await ReadAllAsync(session); @@ -115,7 +115,7 @@ public async Task ReadAsync_Should_Decode_A_Broken_Surrogate_With_Replacement() var socket = new ScriptedTerminalWebSocket() .Binary(PtyFrameData.UnpairedSurrogate) .Closing(WebSocketCloseStatus.NormalClosure); - await using var session = new PtySession(socket); + await using var session = new PtySession(socket, new ManualTimeProvider()); var frames = await ReadAllAsync(session); @@ -130,7 +130,7 @@ public async Task ReadAsync_Should_Read_A_Binary_Message_Without_The_Marker_As_O var socket = new ScriptedTerminalWebSocket() .Binary(Encoding.UTF8.GetBytes("raw-output")) .Closing(WebSocketCloseStatus.NormalClosure); - await using var session = new PtySession(socket); + await using var session = new PtySession(socket, new ManualTimeProvider()); var frames = await ReadAllAsync(session); @@ -143,7 +143,7 @@ public async Task ReadAsync_Should_End_The_Enumeration_On_A_Normal_Close() var socket = new ScriptedTerminalWebSocket() .Text("last") .Closing(WebSocketCloseStatus.NormalClosure); - await using var session = new PtySession(socket); + await using var session = new PtySession(socket, new ManualTimeProvider()); var frames = await ReadAllAsync(session); @@ -154,7 +154,7 @@ public async Task ReadAsync_Should_End_The_Enumeration_On_A_Normal_Close() public async Task ReadAsync_Should_Refuse_A_Session_Not_Found_Close() { var socket = new ScriptedTerminalWebSocket().Closing(SessionNotFound, SessionNotFoundReason); - await using var session = new PtySession(socket); + await using var session = new PtySession(socket, new ManualTimeProvider()); var failure = await AssertSessionNotFoundAsync(session); await Assert.That(failure.Message).Contains(SessionNotFoundReason); @@ -166,7 +166,7 @@ public async Task ReadAsync_Should_Refuse_A_Session_Not_Found_Close_With_An_Empt // A real ClientWebSocket reports an empty description, not null, when the peer sent no // reason text; the policy must treat both the same rather than rendering a hollow "4404 ()". var socket = new ScriptedTerminalWebSocket().Closing(SessionNotFound, string.Empty); - await using var session = new PtySession(socket); + await using var session = new PtySession(socket, new ManualTimeProvider()); var failure = await AssertSessionNotFoundAsync(session); await Assert.That(failure.Message).DoesNotContain("()", StringComparison.Ordinal); @@ -184,7 +184,7 @@ private static async Task AssertSessionNotFoundAsync public async Task ReadAsync_Should_Refuse_An_Abnormal_Close() { var socket = new ScriptedTerminalWebSocket().Closing(WebSocketCloseStatus.ProtocolError); - await using var session = new PtySession(socket); + await using var session = new PtySession(socket, new ManualTimeProvider()); var failure = await Assert.That(async () => _ = await ReadAllAsync(session)) .Throws(); @@ -196,7 +196,7 @@ public async Task ReadAsync_Should_Refuse_An_Abnormal_Close() public async Task ReadAsync_Should_Refuse_A_Second_Concurrent_Enumeration() { var socket = new ScriptedTerminalWebSocket().Text("first").Parking(); - await using var session = new PtySession(socket); + await using var session = new PtySession(socket, new ManualTimeProvider()); var first = session.ReadAsync().GetAsyncEnumerator(CancellationToken.None); await using var enumeration = first.ConfigureAwait(false); await Assert.That(await first.MoveNextAsync()).IsTrue(); @@ -213,7 +213,7 @@ public async Task ReadAsync_Should_Allow_A_Second_Enumeration_After_The_First_Br .Text("first") .Text("second") .Closing(WebSocketCloseStatus.NormalClosure); - await using var session = new PtySession(socket); + await using var session = new PtySession(socket, new ManualTimeProvider()); await using (var first = session.ReadAsync().GetAsyncEnumerator()) { @@ -230,7 +230,7 @@ public async Task ReadAsync_Should_Allow_A_Second_Enumeration_After_The_First_Br public async Task ReadAsync_Should_Report_A_Caller_Cancellation_As_Cancellation() { var socket = new ScriptedTerminalWebSocket().Parking(); - await using var session = new PtySession(socket); + await using var session = new PtySession(socket, new ManualTimeProvider()); using var cancellation = new CancellationTokenSource(); var enumerator = session.ReadAsync(cancellation.Token).GetAsyncEnumerator(cancellation.Token); var pending = enumerator.MoveNextAsync(); @@ -245,7 +245,7 @@ public async Task ReadAsync_Should_Report_A_Caller_Cancellation_As_Cancellation( public async Task ReadAsync_Should_Report_A_Socket_Fault_As_A_Transport_Failure() { var socket = new ScriptedTerminalWebSocket().Faulting(new WebSocketException("connection reset")); - await using var session = new PtySession(socket); + await using var session = new PtySession(socket, new ManualTimeProvider()); var failure = await Assert.That(async () => _ = await ReadAllAsync(session)) .Throws(); @@ -257,7 +257,7 @@ public async Task ReadAsync_Should_Report_A_Socket_Fault_As_A_Transport_Failure( public async Task DisposeAsync_Should_End_A_Pending_Read_As_A_Normal_End() { var socket = new ScriptedTerminalWebSocket().Text("live").Parking(); - var session = new PtySession(socket); + var session = new PtySession(socket, new ManualTimeProvider()); var enumerator = session.ReadAsync().GetAsyncEnumerator(CancellationToken.None); await using var enumeration = enumerator.ConfigureAwait(false); await Assert.That(await enumerator.MoveNextAsync()).IsTrue(); @@ -273,7 +273,7 @@ public async Task DisposeAsync_Should_End_A_Pending_Read_As_A_Normal_End() public async Task WriteAsync_Should_Encode_The_Input_As_Utf8() { var socket = new ScriptedTerminalWebSocket(); - await using var session = new PtySession(socket); + await using var session = new PtySession(socket, new ManualTimeProvider()); await session.WriteAsync("héllo→\n"); @@ -285,7 +285,7 @@ public async Task WriteAsync_Should_Encode_The_Input_As_Utf8() public async Task WriteAsync_Should_Send_A_Text_Message() { var socket = new ScriptedTerminalWebSocket(); - await using var session = new PtySession(socket); + await using var session = new PtySession(socket, new ManualTimeProvider()); await session.WriteAsync("ls\r"); @@ -297,7 +297,7 @@ public async Task WriteAsync_Should_Send_A_Text_Message() public async Task WriteAsync_Should_Serialize_Concurrent_Sends() { var socket = new ScriptedTerminalWebSocket().GatingSends(); - await using var session = new PtySession(socket); + await using var session = new PtySession(socket, new ManualTimeProvider()); var first = session.WriteAsync("first"); await socket.SendEntered; @@ -315,7 +315,7 @@ public async Task WriteAsync_Should_Serialize_Concurrent_Sends() public async Task WriteAsync_Should_Refuse_A_Null_Input() { var socket = new ScriptedTerminalWebSocket(); - await using var session = new PtySession(socket); + await using var session = new PtySession(socket, new ManualTimeProvider()); _ = await Assert.That(async () => await session.WriteAsync(null!)).Throws(); } @@ -324,8 +324,7 @@ public async Task WriteAsync_Should_Refuse_A_Null_Input() public async Task WriteAsync_Should_Refuse_A_Write_After_Disposal() { var socket = new ScriptedTerminalWebSocket(); - var session = new PtySession(socket); - await session.DisposeAsync(); + var session = await CreateDisposedSessionAsync(socket); _ = await Assert.That(async () => await session.WriteAsync("late")).Throws(); } @@ -348,7 +347,7 @@ public static IEnumerable> LinesCarryingABreak() => public async Task SubmitAsync_Should_Send_The_Line_Terminated_By_Enter() { var socket = new ScriptedTerminalWebSocket(); - await using var session = new PtySession(socket); + await using var session = new PtySession(socket, new ManualTimeProvider()); await session.SubmitAsync(SubmittedLine); @@ -360,7 +359,7 @@ public async Task SubmitAsync_Should_Send_The_Line_Terminated_By_Enter() public async Task SubmitAsync_Should_Send_A_Bare_Enter_For_An_Empty_Line() { var socket = new ScriptedTerminalWebSocket(); - await using var session = new PtySession(socket); + await using var session = new PtySession(socket, new ManualTimeProvider()); await session.SubmitAsync(string.Empty); @@ -371,7 +370,7 @@ public async Task SubmitAsync_Should_Send_A_Bare_Enter_For_An_Empty_Line() public async Task SubmitAsync_Should_Refuse_A_Null_Line() { var socket = new ScriptedTerminalWebSocket(); - await using var session = new PtySession(socket); + await using var session = new PtySession(socket, new ManualTimeProvider()); _ = await Assert.That(async () => await session.SubmitAsync(null!)).Throws(); } @@ -381,7 +380,7 @@ public async Task SubmitAsync_Should_Refuse_A_Null_Line() public async Task SubmitAsync_Should_Refuse_A_Line_Carrying_A_Break(string line) { var socket = new ScriptedTerminalWebSocket(); - await using var session = new PtySession(socket); + await using var session = new PtySession(socket, new ManualTimeProvider()); var failure = await Assert.That(async () => await session.SubmitAsync(line)).Throws(); @@ -394,8 +393,7 @@ public async Task SubmitAsync_Should_Refuse_A_Line_Carrying_A_Break(string line) public async Task SubmitAsync_Should_Refuse_A_Submit_After_Disposal() { var socket = new ScriptedTerminalWebSocket(); - var session = new PtySession(socket); - await session.DisposeAsync(); + var session = await CreateDisposedSessionAsync(socket); _ = await Assert.That(async () => await session.SubmitAsync(SubmittedLine)).Throws(); } @@ -404,7 +402,7 @@ public async Task SubmitAsync_Should_Refuse_A_Submit_After_Disposal() public async Task SubmitAsync_Should_Report_A_Failed_Send_As_A_Transport_Failure() { var socket = new ScriptedTerminalWebSocket().FailingNextSendWith(new WebSocketException("connection reset")); - await using var session = new PtySession(socket); + await using var session = new PtySession(socket, new ManualTimeProvider()); var failure = await Assert.That(async () => await session.SubmitAsync(SubmittedLine)) .Throws(); @@ -417,7 +415,7 @@ public async Task SubmitAsync_Should_Report_A_Failed_Send_As_A_Transport_Failure public async Task SubmitAsync_Should_Serialize_Behind_An_Outstanding_Write() { var socket = new ScriptedTerminalWebSocket().GatingSends(); - await using var session = new PtySession(socket); + await using var session = new PtySession(socket, new ManualTimeProvider()); var write = session.WriteAsync("partial"); await socket.SendEntered; @@ -435,7 +433,7 @@ public async Task SubmitAsync_Should_Serialize_Behind_An_Outstanding_Write() public async Task DisposeAsync_Should_Let_An_In_Flight_Write_Finish_Cleanly() { var socket = new ScriptedTerminalWebSocket().GatingSends(); - var session = new PtySession(socket); + var session = CreateSession(socket); var write = session.WriteAsync("in-flight"); await socket.SendEntered; @@ -455,7 +453,7 @@ public async Task DisposeAsync_Should_Let_An_In_Flight_Write_Finish_Cleanly() public async Task DisposeAsync_Should_Refuse_A_Queued_Write_Rather_Than_Strand_It() { var socket = new ScriptedTerminalWebSocket().GatingSends(); - var session = new PtySession(socket); + var session = new PtySession(socket, new ManualTimeProvider()); var first = session.WriteAsync("first"); await socket.SendEntered; var queued = session.WriteAsync("queued"); @@ -487,7 +485,7 @@ public async Task DisposeAsync_Should_Refuse_A_Queued_Write_Rather_Than_Strand_I public async Task DisposeAsync_Should_Not_Close_While_A_Send_Is_Outstanding() { var socket = new ScriptedTerminalWebSocket().GatingSends(); - var session = new PtySession(socket); + var session = CreateSession(socket); var write = session.WriteAsync("in-flight"); await socket.SendEntered; @@ -506,7 +504,7 @@ public async Task DisposeAsync_Should_Not_Close_While_A_Send_Is_Outstanding() public async Task DisposeAsync_Should_Tear_The_Socket_Down_When_The_Graceful_Close_Fails() { var socket = new ScriptedTerminalWebSocket().FailingCloseWith(new InvalidOperationException("wrong state")); - var session = new PtySession(socket); + var session = new PtySession(socket, new ManualTimeProvider()); await session.DisposeAsync(); @@ -520,7 +518,7 @@ public async Task DisposeAsync_Should_Tear_The_Socket_Down_When_The_Graceful_Clo public async Task DisposeAsync_Should_Close_Gracefully_And_Stay_Idempotent() { var socket = new ScriptedTerminalWebSocket(); - var session = new PtySession(socket); + var session = new PtySession(socket, new ManualTimeProvider()); await session.DisposeAsync(); await session.DisposeAsync(); @@ -575,6 +573,15 @@ public async Task Session_Mock_Seam_Should_Fail_Instructively_Without_An_Overrid await Assert.That(failure!.Message).Contains("mocking constructor"); } + private static PtySession CreateSession(ITerminalWebSocket socket) => new(socket, new ManualTimeProvider()); + + private static async Task CreateDisposedSessionAsync(ITerminalWebSocket socket) + { + var session = CreateSession(socket); + await session.DisposeAsync(); + return session; + } + private static async Task> ReadAllAsync(PtySession session) { var frames = new List(); diff --git a/tests/OpenCode.Sdk.Tests/Terminal/TerminalSendDeadlineTests.cs b/tests/OpenCode.Sdk.Tests/Terminal/TerminalSendDeadlineTests.cs index 9819c4ec..6b92fbcf 100644 --- a/tests/OpenCode.Sdk.Tests/Terminal/TerminalSendDeadlineTests.cs +++ b/tests/OpenCode.Sdk.Tests/Terminal/TerminalSendDeadlineTests.cs @@ -1,6 +1,7 @@ using System.Net.WebSockets; using NSubstitute; using OpenCode.Sdk.Internal; +using OpenCode.Sdk.Tests.Support; namespace OpenCode.Sdk.Tests.Terminal; @@ -9,15 +10,15 @@ public sealed class TerminalSendDeadlineTests [Test] public async Task Map_Should_Preserve_A_Connection_Outcome_Recorded_Before_Expiry() { - using var timer = new CancellationTokenSource(); - var timers = CreateTimers(timer); + var time = new ManualTimeProvider(); TerminalConnectionOutcome? recorded = null; using var deadline = new TerminalSendDeadline( - typeof(PtySession), TerminalSocketBounds.DefaultSendTimeout, timers, () => recorded, CancellationToken.None); + typeof(PtySession), TerminalSocketBounds.DefaultSendTimeout, time, () => recorded, CancellationToken.None); var reset = new OpenCodeTransportException("The connection reset before expiry."); recorded = new TerminalConnectionOutcome(reset); - await timer.CancelOnWorkerAsync(); + await time.WaitForArmedAsync(TerminalSocketBounds.DefaultSendTimeout, HangGuard.Bound); + time.Advance(TerminalSocketBounds.DefaultSendTimeout); var failure = deadline.Map(new OperationCanceledException(deadline.Token), queued: true, disposed: false, recorded); await Assert.That(failure).IsSameReferenceAs(reset); @@ -26,12 +27,12 @@ public async Task Map_Should_Preserve_A_Connection_Outcome_Recorded_Before_Expir [Test] public async Task Map_Should_Preserve_The_Callers_Token_After_Expiry() { - using var timer = new CancellationTokenSource(); + var time = new ManualTimeProvider(); using var caller = new CancellationTokenSource(); - var timers = CreateTimers(timer); using var deadline = new TerminalSendDeadline( - typeof(PtySession), TerminalSocketBounds.DefaultSendTimeout, timers, static () => null, caller.Token); - await timer.CancelOnWorkerAsync(); + typeof(PtySession), TerminalSocketBounds.DefaultSendTimeout, time, static () => null, caller.Token); + await time.WaitForArmedAsync(TerminalSocketBounds.DefaultSendTimeout, HangGuard.Bound); + time.Advance(TerminalSocketBounds.DefaultSendTimeout); await caller.CancelOnWorkerAsync(); var interrupted = new OperationCanceledException(deadline.Token); @@ -45,14 +46,13 @@ public async Task Map_Should_Preserve_The_Callers_Token_After_Expiry() [Test] public async Task Map_Should_Preserve_A_Reset_That_Precedes_Deadline_Interruption() { - using var timer = new CancellationTokenSource(); - var timers = CreateTimers(timer); + var time = new ManualTimeProvider(); using var socket = Substitute.For(); var send = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); _ = socket.SendAsync(Arg.Any>(), Arg.Any(), Arg.Any()) .Returns(send.Task); using var deadline = new TerminalSendDeadline( - typeof(PtySession), TerminalSocketBounds.DefaultSendTimeout, timers, static () => null, CancellationToken.None); + typeof(PtySession), TerminalSocketBounds.DefaultSendTimeout, time, static () => null, CancellationToken.None); var reset = new WebSocketException(WebSocketError.ConnectionClosedPrematurely); send.SetException(reset); WebSocketException? original = null; @@ -65,7 +65,8 @@ public async Task Map_Should_Preserve_A_Reset_That_Precedes_Deadline_Interruptio original = exception; } - await timer.CancelOnWorkerAsync(); + await time.WaitForArmedAsync(TerminalSocketBounds.DefaultSendTimeout, HangGuard.Bound); + time.Advance(TerminalSocketBounds.DefaultSendTimeout); await Assert.That(original).IsSameReferenceAs(reset); var failure = deadline.Map(original!, queued: false, disposed: false, outcome: null); @@ -73,10 +74,4 @@ public async Task Map_Should_Preserve_A_Reset_That_Precedes_Deadline_Interruptio await Assert.That(failure.InnerException).IsSameReferenceAs(reset); } - internal static ITerminalDeadlineFactory CreateTimers(CancellationTokenSource timer) - { - var timers = Substitute.For(); - _ = timers.Create(Arg.Any()).Returns(timer); - return timers; - } } diff --git a/tests/OpenCode.Sdk.Tests/Terminal/TerminalSendQueueTests.cs b/tests/OpenCode.Sdk.Tests/Terminal/TerminalSendQueueTests.cs index a219718a..f05bcb6b 100644 --- a/tests/OpenCode.Sdk.Tests/Terminal/TerminalSendQueueTests.cs +++ b/tests/OpenCode.Sdk.Tests/Terminal/TerminalSendQueueTests.cs @@ -1,3 +1,4 @@ +using NSubstitute; using OpenCode.Sdk.Internal; using OpenCode.Sdk.Tests.Support; @@ -16,4 +17,121 @@ await Assert.That(suppressions.Single().Reason).IsEqualTo( "Expected graceful-close failure is superseded by the owner's unconditional socket teardown " + "and receiver/send joins in DisposeAsync; disposal completion is the observable cleanup outcome."); } + + [Test] + public async Task CloseAsync_Should_Bound_Admission_Without_Leaving_A_Waiter() + { + var time = new ManualTimeProvider(); + using var socket = Substitute.For(); + using var queue = new TerminalSendQueue(socket, time); + await queue.WaitAsync(CancellationToken.None); + var close = queue.CloseAsync(); + try + { + await time.WaitForArmedAsync(TerminalSocketBounds.GracefulCloseTimeout, HangGuard.Bound); + time.Advance(TerminalSocketBounds.GracefulCloseTimeout - TimeSpan.FromMilliseconds(1)); + await Assert.That(close.IsCompleted).IsFalse(); + await socket.DidNotReceive().CloseOutputAsync(Arg.Any()); + time.Advance(TimeSpan.FromMilliseconds(1)); + await close.WaitAsync(HangGuard.Bound); + await socket.DidNotReceive().CloseOutputAsync(Arg.Any()); + } + finally + { + queue.Release(); + await close.WaitAsync(HangGuard.Bound); + } + + await queue.WaitAsync(CancellationToken.None).WaitAsync(HangGuard.Bound); + queue.Release(); + } + + [Test] + public async Task CloseAsync_Should_Bound_Close_Output_And_Release_Admission() + { + using var socket = Substitute.For(); + using var cleanup = new CancellationTokenSource(); + using var queue = CreateGatedCloseQueue(socket, cleanup.Token, out var time, out var entered); + var close = queue.CloseAsync(); + try + { + var token = await entered.WaitAsync(HangGuard.Bound); + await time.WaitForArmedAsync(TerminalSocketBounds.GracefulCloseTimeout, HangGuard.Bound); + time.Advance(TerminalSocketBounds.GracefulCloseTimeout - TimeSpan.FromMilliseconds(1)); + await Assert.That(token.IsCancellationRequested).IsFalse(); + await Assert.That(close.IsCompleted).IsFalse(); + time.Advance(TimeSpan.FromMilliseconds(1)); + await close.WaitAsync(HangGuard.Bound); + await AssertAdmissionReleasedAsync(queue, token); + } + finally + { + await cleanup.CancelOnWorkerAsync(); + await close.WaitAsync(HangGuard.Bound); + } + } + + [Test] + public async Task CloseAsync_Should_Give_Close_Output_A_Fresh_Budget_After_Admission() + { + using var socket = Substitute.For(); + using var cleanup = new CancellationTokenSource(); + using var queue = CreateGatedCloseQueue(socket, cleanup.Token, out var time, out var entered); + await queue.WaitAsync(CancellationToken.None); + var held = true; + var close = queue.CloseAsync(); + try + { + await time.WaitForArmedAsync(TerminalSocketBounds.GracefulCloseTimeout, HangGuard.Bound); + time.Advance(TerminalSocketBounds.GracefulCloseTimeout - TimeSpan.FromMilliseconds(1)); + queue.Release(); + held = false; + var token = await entered.WaitAsync(HangGuard.Bound); + await time.WaitForArmedAsync(TerminalSocketBounds.GracefulCloseTimeout, HangGuard.Bound); + time.Advance(TimeSpan.FromMilliseconds(1)); + await Assert.That(token.IsCancellationRequested).IsFalse(); + await Assert.That(close.IsCompleted).IsFalse(); + time.Advance(TerminalSocketBounds.GracefulCloseTimeout - TimeSpan.FromMilliseconds(2)); + await Assert.That(token.IsCancellationRequested).IsFalse(); + time.Advance(TimeSpan.FromMilliseconds(1)); + await close.WaitAsync(HangGuard.Bound); + await AssertAdmissionReleasedAsync(queue, token); + } + finally + { + if (held) + { + queue.Release(); + } + + await cleanup.CancelOnWorkerAsync(); + await close.WaitAsync(HangGuard.Bound); + } + } + + private static TerminalSendQueue CreateGatedCloseQueue( + ITerminalWebSocket socket, CancellationToken cleanup, out ManualTimeProvider time, out Task closeEntered) + { + var entered = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + _ = socket.CloseOutputAsync(Arg.Any()).Returns(async call => + { + var token = call.Arg(); + var release = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + using var cancellation = token.Register(() => release.TrySetCanceled(token)); + using var completion = cleanup.Register(() => release.TrySetResult(true)); + _ = entered.TrySetResult(token); + await release.Task; + }); + closeEntered = entered.Task; + time = new ManualTimeProvider(); + return new TerminalSendQueue(socket, time); + } + + private static async Task AssertAdmissionReleasedAsync(TerminalSendQueue queue, CancellationToken token) + { + await Assert.That(token.IsCancellationRequested).IsTrue(); + // The expired token is observed, not used to cancel the admission check. + await queue.WaitAsync(CancellationToken.None).WaitAsync(HangGuard.Bound, CancellationToken.None); + queue.Release(); + } } diff --git a/tests/OpenCode.Sdk.Tests/Terminal/TerminalSocketCoreTests.cs b/tests/OpenCode.Sdk.Tests/Terminal/TerminalSocketCoreTests.cs index a193ffe0..871f8716 100644 --- a/tests/OpenCode.Sdk.Tests/Terminal/TerminalSocketCoreTests.cs +++ b/tests/OpenCode.Sdk.Tests/Terminal/TerminalSocketCoreTests.cs @@ -16,8 +16,7 @@ public sealed class TerminalSocketCoreTests [Test] public async Task SendAsync_Should_Preserve_A_Reset_Completed_Before_Deadline_Interruption() { - using var timer = new CancellationTokenSource(); - var timers = TerminalSendDeadlineTests.CreateTimers(timer); + var time = new ManualTimeProvider(); using var socket = Substitute.For(); var send = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); var reset = new WebSocketException(WebSocketError.ConnectionClosedPrematurely); @@ -31,10 +30,7 @@ public async Task SendAsync_Should_Preserve_A_Reset_Completed_Before_Deadline_In _ = socket.SendAsync(Arg.Any>(), Arg.Any(), Arg.Any()) .Returns(send.Task); await using var core = new TerminalSocketCore( - socket, PtyFrameDecoder.Instance, PtyClosePolicy.Instance, typeof(PtySession)) - { - DeadlineFactory = timers, - }; + socket, PtyFrameDecoder.Instance, PtyClosePolicy.Instance, typeof(PtySession), time); var entered = 0; var resumed = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); using var release = new ManualResetEventSlim(); @@ -65,7 +61,8 @@ public async Task SendAsync_Should_Preserve_A_Reset_Completed_Before_Deadline_In { send.SetException(reset); await resumed.Task.WaitAsync(HangGuard.Bound); - await timer.CancelOnWorkerAsync(); + await time.WaitForArmedAsync(core.SendTimeout, HangGuard.Bound); + time.Advance(core.SendTimeout); release.Set(); var failure = await Assert.That(async () => { @@ -88,8 +85,7 @@ public async Task SendAsync_Should_Preserve_A_Reset_Completed_Before_Deadline_In [Test] public async Task SendAsync_Should_Report_A_Timeout_When_The_Deadline_Abort_Reaches_Receive_First() { - using var timer = new CancellationTokenSource(); - var timers = TerminalSendDeadlineTests.CreateTimers(timer); + var time = new ManualTimeProvider(); using var socket = Substitute.For(); var receive = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); var send = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); @@ -103,10 +99,7 @@ public async Task SendAsync_Should_Report_A_Timeout_When_The_Deadline_Abort_Reac operationToken = call.Arg(); }); await using var core = new TerminalSocketCore( - socket, PtyFrameDecoder.Instance, PtyClosePolicy.Instance, typeof(PtySession)) - { - DeadlineFactory = timers, - }; + socket, PtyFrameDecoder.Instance, PtyClosePolicy.Instance, typeof(PtySession), time); var pending = core.SendAsync(new ArraySegment([0x61]), WebSocketMessageType.Text, CancellationToken.None); using var abort = operationToken.Register(() => receive.TrySetException(readFault)); OpenCodeTransportException? attachmentFailure = null; @@ -114,7 +107,8 @@ public async Task SendAsync_Should_Report_A_Timeout_When_The_Deadline_Abort_Reac await using var reader = core.ReadAsync(CancellationToken.None).GetAsyncEnumerator(); try { - await timer.CancelOnWorkerAsync(); + await time.WaitForArmedAsync(core.SendTimeout, HangGuard.Bound); + time.Advance(core.SendTimeout); attachmentFailure = await Assert.That(async () => _ = await reader.MoveNextAsync().AsTask().WaitAsync(HangGuard.Bound)) .Throws(); } @@ -159,7 +153,7 @@ public async Task SendAsync_Should_Wake_A_Queued_Call_With_The_Connections_Termi using var socket = new ScriptedTerminalWebSocket().Pausing() .Closing(status).GatingSends(); await using var core = new TerminalSocketCore( - socket, PtyFrameDecoder.Instance, PtyClosePolicy.Instance, typeof(PtySession)); + socket, PtyFrameDecoder.Instance, PtyClosePolicy.Instance, typeof(PtySession), new ManualTimeProvider()); var first = core.SendAsync(new ArraySegment([0x61]), WebSocketMessageType.Text, CancellationToken.None); await socket.SendEntered; var queued = core.SendAsync(new ArraySegment([0x62]), WebSocketMessageType.Text, CancellationToken.None); @@ -190,7 +184,7 @@ public async Task SendAsync_Should_Preserve_The_First_Terminal_Outcome_After_A_L using var socket = new ScriptedTerminalWebSocket().Pausing() .Closing(WebSocketCloseStatus.InternalServerError).GatingSends(); await using var core = new TerminalSocketCore( - socket, PtyFrameDecoder.Instance, PtyClosePolicy.Instance, typeof(PtySession)); + socket, PtyFrameDecoder.Instance, PtyClosePolicy.Instance, typeof(PtySession), new ManualTimeProvider()); using var cancellation = new CancellationTokenSource(); var pending = core.SendAsync(new ArraySegment([0x61]), WebSocketMessageType.Text, cancellation.Token); await socket.SendEntered; @@ -219,7 +213,7 @@ public async Task SendAsync_Should_End_The_Attachment_After_Physical_Caller_Canc { using var socket = new ScriptedTerminalWebSocket().GatingSends(); await using var core = new TerminalSocketCore( - socket, PtyFrameDecoder.Instance, PtyClosePolicy.Instance, typeof(PtySession)); + socket, PtyFrameDecoder.Instance, PtyClosePolicy.Instance, typeof(PtySession), new ManualTimeProvider()); using var cancellation = new CancellationTokenSource(); var pending = core.SendAsync(new ArraySegment([0x61]), WebSocketMessageType.Text, cancellation.Token); await socket.SendEntered; @@ -247,7 +241,7 @@ public async Task ReadAsync_Should_Preserve_A_Fragmented_Message_Across_Consumer using var socket = new ScriptedTerminalWebSocket() .TextFragment("first ").Pausing().Text("second").Closing(WebSocketCloseStatus.NormalClosure); await using var core = new TerminalSocketCore( - socket, PtyFrameDecoder.Instance, PtyClosePolicy.Instance, typeof(PtySession)); + socket, PtyFrameDecoder.Instance, PtyClosePolicy.Instance, typeof(PtySession), new ManualTimeProvider()); using var cancellation = new CancellationTokenSource(); await using (var reader = core.ReadAsync(cancellation.Token).GetAsyncEnumerator()) { @@ -270,7 +264,7 @@ public async Task ReadAsync_Should_Leave_Queued_Data_And_The_Terminal_Error_When { using var socket = new ScriptedTerminalWebSocket().Text("prefix").Closing(WebSocketCloseStatus.InternalServerError); await using var core = new TerminalSocketCore( - socket, PtyFrameDecoder.Instance, PtyClosePolicy.Instance, typeof(PtySession)); + socket, PtyFrameDecoder.Instance, PtyClosePolicy.Instance, typeof(PtySession), new ManualTimeProvider()); using var cancellation = new CancellationTokenSource(); await cancellation.CancelOnWorkerAsync(); await using (var reader = core.ReadAsync(cancellation.Token).GetAsyncEnumerator()) @@ -289,7 +283,7 @@ public async Task ReadAsync_Should_Preserve_A_Connection_Failure_Across_Readers( { using var socket = new ScriptedTerminalWebSocket().Text("prefix").Closing(WebSocketCloseStatus.InternalServerError); await using var core = new TerminalSocketCore( - socket, PtyFrameDecoder.Instance, PtyClosePolicy.Instance, typeof(PtySession)); + socket, PtyFrameDecoder.Instance, PtyClosePolicy.Instance, typeof(PtySession), new ManualTimeProvider()); await using (var first = core.ReadAsync(CancellationToken.None).GetAsyncEnumerator()) { await Assert.That(await first.MoveNextAsync()).IsTrue(); @@ -307,7 +301,7 @@ public async Task ReadAsync_Should_Keep_Normal_Completion_After_The_Queued_Prefi { using var socket = new ScriptedTerminalWebSocket().Text("prefix").Closing(WebSocketCloseStatus.NormalClosure); await using var core = new TerminalSocketCore( - socket, PtyFrameDecoder.Instance, PtyClosePolicy.Instance, typeof(PtySession)); + socket, PtyFrameDecoder.Instance, PtyClosePolicy.Instance, typeof(PtySession), new ManualTimeProvider()); await using (var first = core.ReadAsync(CancellationToken.None).GetAsyncEnumerator()) { await Assert.That(await first.MoveNextAsync()).IsTrue(); @@ -324,7 +318,7 @@ public async Task DisposeAsync_Should_Hard_Close_And_Join_Io_When_Graceful_Close using var socket = new ScriptedTerminalWebSocket().Parking() .FailingCloseWith(new WebSocketException(WebSocketError.ConnectionClosedPrematurely)); await using var core = new TerminalSocketCore( - socket, PtyFrameDecoder.Instance, PtyClosePolicy.Instance, typeof(PtySession)); + socket, PtyFrameDecoder.Instance, PtyClosePolicy.Instance, typeof(PtySession), new ManualTimeProvider()); await socket.Parked; await core.SendAsync(new ArraySegment([0x61]), WebSocketMessageType.Text, CancellationToken.None); @@ -341,7 +335,7 @@ public async Task DisposeAsync_Should_Not_Require_A_Suspended_Consumer_To_Resume { using var socket = new ScriptedTerminalWebSocket().Text("prefix").Text("unread").Parking(); await using var core = new TerminalSocketCore( - socket, PtyFrameDecoder.Instance, PtyClosePolicy.Instance, typeof(PtySession)); + socket, PtyFrameDecoder.Instance, PtyClosePolicy.Instance, typeof(PtySession), new ManualTimeProvider()); await using var first = core.ReadAsync(CancellationToken.None).GetAsyncEnumerator(); await Assert.That(await first.MoveNextAsync()).IsTrue(); @@ -357,7 +351,7 @@ public async Task DisposeAsync_Should_Share_In_Progress_Cleanup() { using var socket = new ScriptedTerminalWebSocket().GatingSends(); await using var core = new TerminalSocketCore( - socket, PtyFrameDecoder.Instance, PtyClosePolicy.Instance, typeof(PtySession)); + socket, PtyFrameDecoder.Instance, PtyClosePolicy.Instance, typeof(PtySession), new ManualTimeProvider()); var send = core.SendAsync(new ArraySegment([0x61]), WebSocketMessageType.Text, CancellationToken.None); await socket.SendEntered; var firstDisposal = core.DisposeAsync(); @@ -380,34 +374,63 @@ public async Task DisposeAsync_Should_Share_In_Progress_Cleanup() [Test] public async Task SendAsync_Should_Expire_A_Queued_Call_Without_Closing_The_Socket() { - using var firstDeadline = new CancellationTokenSource(); - using var queuedDeadline = new CancellationTokenSource(); - using var nextDeadline = new CancellationTokenSource(); - var timers = Substitute.For(); - _ = timers.Create(Arg.Any()).Returns(firstDeadline, queuedDeadline, nextDeadline); + var time = new ManualTimeProvider(); using var socket = new ScriptedTerminalWebSocket().GatingSends(); + using var transport = Substitute.For(); + var physicalSend = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + var recordedSend = Task.CompletedTask; + _ = transport.ReceiveAsync(Arg.Any>(), Arg.Any()) + .Returns(call => socket.ReceiveAsync(call.Arg>(), call.Arg())); + _ = transport.SendAsync(Arg.Any>(), Arg.Any(), Arg.Any()) + .Returns(physicalSend.Task).AndDoes(call => + { + recordedSend = socket.SendAsync( + call.Arg>(), call.Arg(), call.Arg()); + }); + _ = transport.CloseOutputAsync(Arg.Any()) + .Returns(call => socket.CloseOutputAsync(call.Arg())); + transport.When(value => value.Dispose()).Do(_ => socket.Dispose()); await using var core = new TerminalSocketCore( - socket, PtyFrameDecoder.Instance, PtyClosePolicy.Instance, typeof(PtySession)) + transport, PtyFrameDecoder.Instance, PtyClosePolicy.Instance, typeof(PtySession), time); + var resumed = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + using var release = new ManualResetEventSlim(); + var released = false; + // Hold the first call in its own post-send step, before it releases the send gate, so only + // the queued call can expire; the hold never throws, and the test thread asserts its result. + var holdAfterSend = new TerminalSendActions(OnSent: () => { - DeadlineFactory = timers, - }; - var first = core.SendAsync(new ArraySegment([0x61]), WebSocketMessageType.Text, CancellationToken.None); + _ = resumed.TrySetResult(true); + released = release.Wait(HangGuard.Bound); + }); + var first = core.SendAsync(() => new ArraySegment([0x61]), WebSocketMessageType.Text, holdAfterSend, CancellationToken.None); + await socket.SendEntered; var queued = core.SendAsync(new ArraySegment([0x62]), WebSocketMessageType.Text, CancellationToken.None); - await queuedDeadline.CancelOnWorkerAsync(); - socket.ReleaseSends(); - await first; - OpenCodeTransportException? failure = null; try { + socket.ReleaseSends(); + await recordedSend.WaitAsync(HangGuard.Bound); + physicalSend.SetResult(true); + await resumed.Task.WaitAsync(HangGuard.Bound); + await time.WaitForArmedAsync(core.SendTimeout, HangGuard.Bound, count: 2); + time.Advance(core.SendTimeout); await queued; } catch (OpenCodeTransportException exception) { failure = exception; } + finally + { + release.Set(); + socket.ReleaseSends(); + await recordedSend.WaitAsync(HangGuard.Bound); + _ = physicalSend.TrySetResult(true); + await first.WaitAsync(HangGuard.Bound); + } + await Assert.That(released).IsTrue(); await Assert.That(failure).IsNotNull(); await Assert.That(failure!.InnerException).IsTypeOf(); await Assert.That(socket.SentMessages.Count).IsEqualTo(1); @@ -419,18 +442,14 @@ public async Task SendAsync_Should_Expire_A_Queued_Call_Without_Closing_The_Sock [Test] public async Task SendAsync_Should_End_The_Attachment_When_A_Physical_Send_Expires() { - using var deadline = new CancellationTokenSource(); - var timers = Substitute.For(); - _ = timers.Create(Arg.Any()).Returns(deadline); + var time = new ManualTimeProvider(); using var socket = new ScriptedTerminalWebSocket().GatingSends(); await using var core = new TerminalSocketCore( - socket, PtyFrameDecoder.Instance, PtyClosePolicy.Instance, typeof(PtySession)) - { - DeadlineFactory = timers, - }; + socket, PtyFrameDecoder.Instance, PtyClosePolicy.Instance, typeof(PtySession), time); var send = core.SendAsync(new ArraySegment([0x61]), WebSocketMessageType.Text, CancellationToken.None); await socket.SendEntered; - await deadline.CancelOnWorkerAsync(); + await time.WaitForArmedAsync(core.SendTimeout, HangGuard.Bound); + time.Advance(core.SendTimeout); socket.ReleaseSends(); OpenCodeTransportException? failure = null; @@ -456,7 +475,7 @@ public async Task SendAsync_Should_End_The_Attachment_When_A_Physical_Send_Expir public async Task Constructor_Should_Refuse_A_Null_Socket() { var failure = Assert.Throws(() => _ = new TerminalSocketCore( - socket: null!, PtyFrameDecoder.Instance, PtyClosePolicy.Instance, typeof(PtySession))); + socket: null!, PtyFrameDecoder.Instance, PtyClosePolicy.Instance, typeof(PtySession), new ManualTimeProvider())); await Assert.That(failure.ParamName).IsEqualTo("socket"); } @@ -467,7 +486,7 @@ public async Task Constructor_Should_Refuse_A_Null_Decoder() using var socket = new ScriptedTerminalWebSocket(); var failure = Assert.Throws(() => _ = new TerminalSocketCore( - socket, decoder: null!, PtyClosePolicy.Instance, typeof(PtySession))); + socket, decoder: null!, PtyClosePolicy.Instance, typeof(PtySession), new ManualTimeProvider())); await Assert.That(failure.ParamName).IsEqualTo("decoder"); } @@ -478,7 +497,7 @@ public async Task Constructor_Should_Refuse_A_Null_Close_Policy() using var socket = new ScriptedTerminalWebSocket(); var failure = Assert.Throws(() => _ = new TerminalSocketCore( - socket, PtyFrameDecoder.Instance, closePolicy: null!, typeof(PtySession))); + socket, PtyFrameDecoder.Instance, closePolicy: null!, typeof(PtySession), new ManualTimeProvider())); await Assert.That(failure.ParamName).IsEqualTo("closePolicy"); } @@ -489,7 +508,7 @@ public async Task Constructor_Should_Refuse_A_Null_Owner() using var socket = new ScriptedTerminalWebSocket(); var failure = Assert.Throws(() => _ = new TerminalSocketCore( - socket, PtyFrameDecoder.Instance, PtyClosePolicy.Instance, owner: null!)); + socket, PtyFrameDecoder.Instance, PtyClosePolicy.Instance, owner: null!, new ManualTimeProvider())); await Assert.That(failure.ParamName).IsEqualTo("owner"); } @@ -499,7 +518,7 @@ public async Task SendAsync_Should_Name_The_Owning_Door_Once_Disposed() { using var socket = new ScriptedTerminalWebSocket(); await using var core = new TerminalSocketCore( - socket, PtyFrameDecoder.Instance, PtyClosePolicy.Instance, typeof(PtySession)); + socket, PtyFrameDecoder.Instance, PtyClosePolicy.Instance, typeof(PtySession), new ManualTimeProvider()); await core.DisposeAsync(); var failure = await Assert diff --git a/tests/OpenCode.Sdk.Tests/Threading/CancellationTokenSourceExtensionsTests.cs b/tests/OpenCode.Sdk.Tests/Threading/CancellationTokenSourceExtensionsTests.cs new file mode 100644 index 00000000..58bc6717 --- /dev/null +++ b/tests/OpenCode.Sdk.Tests/Threading/CancellationTokenSourceExtensionsTests.cs @@ -0,0 +1,40 @@ +using OpenCode.Sdk.Internal; +using OpenCode.Sdk.Tests.Support; + +namespace OpenCode.Sdk.Tests.Threading; + +public sealed class CancellationTokenSourceExtensionsTests +{ + [Test] + public async Task CancelOnWorkerAsync_Should_Return_Without_Waiting_And_Complete_Only_After_Callbacks() + { + using var source = new CancellationTokenSource(); + using var release = new ManualResetEventSlim(); + var entered = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + var finished = false; + using var registration = source.Token.Register(() => + { + _ = entered.TrySetResult(true); + release.Wait(); + finished = true; + }); + // Publish the cancellation task as data: unwrapping it would hide a blocking invocation. + var invocation = Task.Factory.StartNew(source.CancelOnWorkerAsync, + CancellationToken.None, TaskCreationOptions.DenyChildAttach, TaskScheduler.Default); + try + { + await entered.Task.WaitAsync(HangGuard.Bound); + var cancellation = await invocation.WaitAsync(HangGuard.Bound); + await Assert.That(cancellation.IsCompleted).IsFalse(); + release.Set(); + await cancellation.WaitAsync(HangGuard.Bound); + await Assert.That(finished).IsTrue(); + } + finally + { + release.Set(); + var cancellation = await invocation.WaitAsync(HangGuard.Bound); + await cancellation.WaitAsync(HangGuard.Bound); + } + } +} diff --git a/tests/OpenCode.Sdk.Tests/Threading/TimedCancellationTests.cs b/tests/OpenCode.Sdk.Tests/Threading/TimedCancellationTests.cs index 570b6498..2af06857 100644 --- a/tests/OpenCode.Sdk.Tests/Threading/TimedCancellationTests.cs +++ b/tests/OpenCode.Sdk.Tests/Threading/TimedCancellationTests.cs @@ -37,7 +37,7 @@ public async Task Create_Should_Cancel_At_The_Deadline_And_Not_One_Tick_Before() #if !NET [Test] - public async Task Create_Should_Ignore_An_In_Flight_Timer_Callback_After_Disposal() + public async Task Create_Should_Ignore_A_Timer_Callback_Queued_After_Disposal() { var time = new InFlightTimeProvider(); using var deadline = TimedCancellation.Create(Timeout, time); @@ -50,12 +50,14 @@ public async Task Create_Should_Ignore_An_In_Flight_Timer_Callback_After_Disposa #endif [Test] - public async Task Create_Should_Allow_Timer_Cancellation_To_Race_With_Disposal() + public async Task Create_Should_Let_A_Timer_Callback_With_A_Registration_Race_Disposal() { for (var round = 0; round < 1_000; round++) { var time = new InFlightTimeProvider(); using var deadline = TimedCancellation.Create(Timeout, time); + // A registration makes the callback's Cancel run work under the lock while Dispose contends for it. + using var registration = deadline.Token.Register(static () => { }); using var barrier = new Barrier(2); var firing = Task.Run(() => { diff --git a/tests/OpenCode.Sdk.Tests/Transport/ConnectionSnapshotTests.cs b/tests/OpenCode.Sdk.Tests/Transport/ConnectionSnapshotTests.cs index bba9010f..04d11ee1 100644 --- a/tests/OpenCode.Sdk.Tests/Transport/ConnectionSnapshotTests.cs +++ b/tests/OpenCode.Sdk.Tests/Transport/ConnectionSnapshotTests.cs @@ -1,6 +1,7 @@ using System.Net.Http.Headers; using System.Text; using OpenCode.Sdk.Internal; +using OpenCode.Sdk.Tests.Support; namespace OpenCode.Sdk.Tests.Transport; @@ -15,7 +16,7 @@ public sealed class ConnectionSnapshotTests [Test] public async Task ToString_Should_Not_Carry_The_Credential() { - var snapshot = new ConnectionSnapshot(EndpointBase, BasicCredential(), Location); + var snapshot = new ConnectionSnapshot(EndpointBase, BasicCredential(), Location, new ManualTimeProvider()); var rendered = Render(snapshot); @@ -45,7 +46,7 @@ public async Task The_Snapshot_Should_Still_Carry_Every_Member_It_Was_Built_With { var credential = BasicCredential(); - var snapshot = new ConnectionSnapshot(EndpointBase, credential, Location); + var snapshot = new ConnectionSnapshot(EndpointBase, credential, Location, new ManualTimeProvider()); // Redacting the rendering must not quietly redact the value itself. await Assert.That(snapshot.EndpointBase).IsEqualTo(EndpointBase); @@ -56,7 +57,7 @@ public async Task The_Snapshot_Should_Still_Carry_Every_Member_It_Was_Built_With [Test] public async Task The_Snapshot_Should_Accept_An_Anonymous_Unlocated_Connection() { - var snapshot = new ConnectionSnapshot(EndpointBase, authorization: null, location: null); + var snapshot = new ConnectionSnapshot(EndpointBase, authorization: null, location: null, new ManualTimeProvider()); await Assert.That(snapshot.Authorization).IsNull(); await Assert.That(snapshot.Location).IsNull(); @@ -65,7 +66,7 @@ public async Task The_Snapshot_Should_Accept_An_Anonymous_Unlocated_Connection() [Test] public async Task The_Snapshot_Should_Refuse_A_Blank_Endpoint_Base() { - _ = Assert.Throws(() => _ = new ConnectionSnapshot(" ", authorization: null, location: null)); + _ = Assert.Throws(() => _ = new ConnectionSnapshot(" ", authorization: null, location: null, new ManualTimeProvider())); await Task.CompletedTask; } diff --git a/tools/OpenCode.Sdk.Tools/Generator/Binding/ReservedNamePolicy.cs b/tools/OpenCode.Sdk.Tools/Generator/Binding/ReservedNamePolicy.cs index 4754e508..1dd872e0 100644 --- a/tools/OpenCode.Sdk.Tools/Generator/Binding/ReservedNamePolicy.cs +++ b/tools/OpenCode.Sdk.Tools/Generator/Binding/ReservedNamePolicy.cs @@ -160,7 +160,6 @@ internal static class ReservedNamePolicy "IServicePtyShutdown", "IStreamAdapter", "ITerminalClosePolicy", - "ITerminalDeadlineFactory", "ITerminalFrameDecoder", "ITerminalUpgradeFailurePolicy", "ITerminalWebSocket", @@ -279,9 +278,7 @@ internal static class ReservedNamePolicy "StructuralUnionClaim", "StructuralUnionPrinter", "TaskkillTreeKill", - "TerminalCancellationSource", "TerminalConnectionOutcome", - "TerminalDeadlineFactory", "TerminalReceivePump", "TerminalSendActions", "TerminalSendDeadline", diff --git a/tools/jscpd/tests.baseline.json b/tools/jscpd/tests.baseline.json index 22ce90fa..37462ae4 100644 --- a/tools/jscpd/tests.baseline.json +++ b/tools/jscpd/tests.baseline.json @@ -7,6 +7,7 @@ "01417d72ad1b123c": 1, "01b50a050ff5cb57": 1, "02c449784785f5e1": 1, + "030445ef25b69be7": 1, "030b9be65c2b2225": 1, "032245d683ed5066": 1, "0341de072c280e0e": 2, @@ -20,7 +21,6 @@ "05bec1db549b3145": 1, "060d60ecdc435655": 1, "0646131256062ec7": 1, - "064bde5a422105ed": 1, "06963808d80f8635": 1, "06f41df1aeaf1cd9": 1, "073ba1f3be6c91d0": 1, @@ -44,7 +44,6 @@ "0c53a2dd00579986": 1, "0c6ad1b72d1c5abd": 1, "0cee267a64eb44dc": 1, - "0d1c1e62f027669f": 1, "0d494a1b361ea0f6": 1, "0d59c107c28482f6": 1, "0e3563f01a4c7340": 1, @@ -53,6 +52,7 @@ "0ef0fc8554074249": 1, "0f55f7d7adfa2f82": 1, "0faa099538e4e7a0": 1, + "0fcc7cca1b6402dd": 1, "0ffa9abf9d9e844e": 1, "10003a5ebdb1faa9": 1, "1035e50fc18d0089": 1, @@ -68,6 +68,7 @@ "1363ad233bb49d6b": 1, "139a4466552f7894": 1, "13ad4ac3b5dbd5d7": 1, + "13fa62aa808a28cb": 1, "141c9b35d283a1cc": 1, "14284878346f6672": 1, "144da272b2295a00": 1, @@ -101,6 +102,7 @@ "1fe764b3d181c412": 1, "203f96f808c528e3": 1, "20f47cefc98df1f0": 1, + "210bbefc7e3a3805": 1, "2168c4bc0179c8db": 1, "2179550e5407130d": 1, "21fecd279dd4bb07": 1, @@ -130,10 +132,10 @@ "29e2d3f4c6ced0a9": 1, "2a2ea864e10f10a0": 1, "2a6fba049050425f": 1, - "2ad053ce77d6dfb2": 1, "2bef63e501d18ed2": 1, "2cce43ebc1614860": 1, "2cd64c5a0bd7b129": 1, + "2d01f1bea1f9ea7e": 1, "2d0c7764aa30effc": 1, "2e10753f9a336f97": 1, "2e6446a9c3e54be2": 1, @@ -141,8 +143,10 @@ "2ef093de84b22985": 1, "2fa52323893c6990": 1, "2fce9f9be54d9810": 1, + "2fcfde38c702a305": 1, "30024875e03c93af": 1, "303aee7f7b8af88f": 1, + "3081885a80a22be3": 1, "3095017bef402cf6": 1, "30c6457c90ba767e": 1, "31079b601cabad85": 1, @@ -150,7 +154,6 @@ "32a22f84426c274c": 1, "32a53e20ee7c711b": 1, "338595f92b8c9b2c": 1, - "33df1e6ccf3c7b58": 1, "34812987b8a4a722": 1, "34ff66db689cfea4": 1, "350995582d58fa70": 1, @@ -172,7 +175,6 @@ "3b290eb15830d924": 1, "3b4ee00b9fd149f5": 1, "3b71e78270640ef3": 1, - "3bccb11c8653a298": 1, "3c605c52befef8a8": 1, "3c72e0db609061c0": 1, "3ca9e998947ac50c": 1, @@ -192,11 +194,11 @@ "40385632dadab4ac": 1, "41005cc85152f877": 1, "410855c256ca33cf": 1, - "410dc16d23b0fb80": 1, "411bf938cdbb8b64": 1, "411f04f073e9ad3f": 1, "4192fdb04dccceeb": 1, "41ff4bb4c970a18a": 1, + "4202e901bcb4f8ba": 1, "42e82776f30ff817": 1, "4372939ffb05b447": 1, "442313bd038e5928": 1, @@ -221,6 +223,7 @@ "4ccca39676639bbf": 1, "4e8e100a6ddc1999": 1, "4e9f7bbcbfcdf308": 1, + "50c75dddf97e272d": 1, "50c8491cd38b65bf": 1, "51fe16f43bda14c9": 1, "523b54f16719c5d5": 1, @@ -242,6 +245,7 @@ "580bb3a45cc1aa00": 1, "5890b6797611cdbf": 1, "58cf3aedc0f4f1c5": 1, + "58df2dce6a57ddc3": 1, "5977e81f73804251": 1, "5ae402699aab9b4f": 1, "5b5f89cf29e631cb": 1, @@ -261,7 +265,6 @@ "5fc40824eff9ddbe": 1, "5fce21230688a4cc": 1, "5fea5678f28a491f": 1, - "614c8cb7e327b799": 1, "6168f0be77f3d62e": 1, "61aec69f93ee31a3": 1, "61e86a6218889bef": 1, @@ -272,6 +275,7 @@ "634f7d1273f299c5": 1, "63726760a9f9dac3": 1, "638060fdb01ef26f": 1, + "63ccee6cade2a08c": 1, "63cfa7e6ae3eeca8": 1, "63d25b3e282d370c": 1, "645f364bb0fb02ea": 1, @@ -310,7 +314,6 @@ "6efa92670898d4c4": 1, "7015815c99e52132": 1, "70b546f19e61694e": 1, - "70bd27545e44e77c": 1, "70e78f68bbb3f8b1": 1, "710c8abd019e2510": 1, "714507ef060378c2": 1, @@ -342,7 +345,9 @@ "7b16854634035afe": 1, "7b29c98f174c4aad": 1, "7b66455caf235ad7": 1, + "7baca2d0e7c78a56": 1, "7c4d439b46ebc8bf": 1, + "7cb3773f240a7d3b": 1, "7da92939d4cdca00": 1, "7df00680786e1792": 1, "7e09a30a918c9d2f": 1, @@ -353,8 +358,8 @@ "801aa0d5a3670945": 1, "804568ff414ee4ab": 1, "80cc7036ff095cb9": 1, - "80cfe1c4da15558e": 1, "80df9c5748306399": 1, + "813356530d4e26c5": 1, "81ceb8987ba6c6e3": 1, "822934d2e089a207": 1, "82654bfda780d48f": 1, @@ -367,7 +372,6 @@ "8440cd90df9d40a2": 1, "8472d20a9e5c98f3": 1, "849beb4a35a19b02": 1, - "85079fc139806e83": 1, "852958b905b8447d": 1, "852e00262bbefd12": 1, "859240ba1081004a": 1, @@ -378,6 +382,7 @@ "86fe3739deb7f33a": 1, "87158a808785d646": 1, "882a97e633e8c29d": 1, + "8845f6d375146134": 1, "88490cc5463a3f5b": 1, "8990b0dcf557805f": 1, "89b40759e47ca8db": 1, @@ -401,7 +406,6 @@ "90832cdd704dea7f": 1, "90911bb723015f18": 1, "90e2d0b4e68471be": 1, - "9111880caf330ba5": 1, "911e3ae6c033fd79": 1, "9145cf3245018c5e": 1, "914c8fb49bc2d976": 1, @@ -410,6 +414,7 @@ "926d6280af25895e": 1, "92a23a8351ac6f5b": 1, "93b9870af2c85e3c": 1, + "93c7ffbef646d5bd": 1, "943a5cceaaa0085d": 1, "94c3b038871234a4": 1, "950d14b87d7527d0": 1, @@ -434,7 +439,6 @@ "9b74c90c792496fd": 1, "9b8eeada912f2fdb": 1, "9c209e8b6e1de814": 1, - "9c41941b90bc5dc4": 1, "9cb8a77612db0cc1": 1, "9cfd4908ab13a49c": 1, "9d6b6b61ef9c0916": 1, @@ -454,15 +458,15 @@ "a4dd036b8a4ee2c6": 1, "a4fe4a3bc845b057": 1, "a5397ab292d68e09": 1, + "a5830a7e1b5d7975": 1, "a58ea3d2f30dafb0": 1, "a60791ba6ca05ea3": 1, + "a644bdfbe6ec0044": 1, "a6ed47cf441cb578": 1, "a6f52664bc5db802": 1, "a7361a9996870f8f": 1, "a7842c59f67aee3e": 1, - "a7b53cc19a267d21": 1, "a7ce419852c3f4bb": 1, - "a86ca42b60a2d816": 1, "a888b3cdad3d452d": 1, "a9346b5654c885e9": 1, "a97d033336de820c": 1, @@ -481,10 +485,8 @@ "b03618789a68ee62": 1, "b04ce5425f684986": 1, "b0dcb57bb0a78e76": 1, - "b107b4bdff30ec7a": 1, "b11e207acaec82c8": 1, "b17bbc10f934e336": 1, - "b1f194514a39741f": 1, "b26ec09aa9c7d788": 1, "b4bc3f0719b7f773": 1, "b4f1d187378630d2": 1, @@ -504,10 +506,11 @@ "b914a2a0e162a6cf": 1, "b94d40e5ff48eda4": 1, "b9a2eed0186fe069": 1, + "ba1b81c7dacd299e": 1, + "ba50dc522914af62": 1, "ba8d50bcd5dc7b3d": 1, "bb134471f279fb9d": 1, "bb76a81ecfad3368": 1, - "bb9c534e42d8fbec": 1, "bc16db8fb010ac77": 1, "bd2d26d72be60ab9": 1, "bd433b802ec9b919": 1, @@ -526,7 +529,7 @@ "c194aa2d8a81a46b": 1, "c1a2f7a561dac4db": 1, "c1aefaf779852373": 1, - "c3312d7a632e3665": 1, + "c2a7b415518d2198": 1, "c377e5bc7e557f04": 1, "c3d10f9320c71fb3": 1, "c4e680d4321eb064": 1, @@ -539,9 +542,7 @@ "c9a68b22b15b2c4f": 1, "ca23d20b55a208df": 1, "ca52433f79ca9adf": 1, - "ca8b61da9bd303bb": 1, "cb72f4a5d460e4a8": 1, - "cbf3676e991f1146": 1, "cc7e06ec26f04567": 1, "ccb62a02a7741fc6": 1, "ccc29e43c6a46325": 1, @@ -552,7 +553,7 @@ "cdb94ac71ed58e49": 1, "ce8da66742f7122d": 1, "ced268f3432a0391": 1, - "ceefd01b00d7a785": 1, + "ceed549c863ce378": 1, "d13f1005c59d1243": 1, "d1debd6e7118e546": 1, "d20c4d175159aa90": 1, @@ -571,14 +572,12 @@ "d9240b039fa378ae": 1, "d9554bbd56ab21a4": 1, "d98319a346958e32": 1, - "d9ac5bae878d22b3": 1, "db4877ca7222c2ec": 1, "dbbafe3c109191e2": 1, "dbfef984a824b22e": 1, "ddab192463f1ade4": 1, "de1435ae5e214376": 1, "de25b177ad221b80": 1, - "dee72b571774c9d5": 1, "df712485fc9d33c5": 1, "e17ab17b76a1e87f": 1, "e1e2fd8c426ea2a9": 1, @@ -589,7 +588,6 @@ "e373307cca6d534f": 1, "e3dddcb1f62c19b8": 1, "e445d35591b55a12": 1, - "e45d1d2fb94c859b": 1, "e4bc846ba130e160": 1, "e4c2843bfae3d310": 1, "e539609f53bb27d3": 1, @@ -603,7 +601,6 @@ "e9a4f95d6cc5b939": 1, "eaee9a41a0b130d0": 1, "eb0bbed02f67f613": 1, - "eb7e225d5bfe0159": 1, "ebec59933509b551": 1, "ebfb2fa4553fc40c": 1, "ebfeb862351f9538": 1, @@ -615,6 +612,7 @@ "edce6ae1385c92b6": 1, "ee052f70f2294a6c": 1, "ee1129d359884ce0": 1, + "ee67ec3f21f2f9ce": 1, "ef106fb1d88f9f46": 1, "ef2bdbe0fd231d31": 1, "ef308469d0166e41": 1, @@ -628,6 +626,7 @@ "f279b2bb369cd7f1": 1, "f29076103ef900ac": 1, "f366e666acd97894": 1, + "f37319ef3f0e2c88": 1, "f3bfe6c708e83755": 1, "f3caece08093c884": 1, "f3d16a0490547ad5": 1,