diff --git a/Directory.Packages.props b/Directory.Packages.props index 480f4b15..7707cd15 100644 --- a/Directory.Packages.props +++ b/Directory.Packages.props @@ -10,6 +10,7 @@ + diff --git a/README.md b/README.md index bbc88baf..f45bbfff 100644 --- a/README.md +++ b/README.md @@ -315,11 +315,11 @@ Architecture, decision records, and engineering policy live under [`docs/`](docs example); a larger-capacity pool is a measured, benchmark-gated follow-up rather than a speculative change. -- **`PtySession.ReadAsync()` allocates its receive buffer per call.** Each read rents nothing and - allocates a fresh 16 KiB buffer — measured at 16,776 bytes on the complete read path versus - 24 bytes for decoding alone. Correctness is unaffected; a tight read loop over a chatty terminal - will produce more garbage than it needs to. Pooling the buffer is a queued optimization, held - behind the same benchmark gate as the item above so the change ships with evidence. +- **Each terminal connection allocates a 16 KiB receive buffer.** The connection-owned receiver + reuses it across messages and consumer read enumerations. Pooling remains a benchmark-gated + follow-up; repeated reads on the same connection do not allocate another receive buffer. The + receiver also queues undelivered frames, so slow or absent consumers can grow memory; see the + [terminal lifetime contract](docs/architecture/client-runtime.md). - **Attaching to an existing background service is not implemented.** opencode's third connection mode — discovering a registered daemon through its registration file (`Service.discover` / diff --git a/docs/ROADMAP.md b/docs/ROADMAP.md index d2dc5189..4184e097 100644 --- a/docs/ROADMAP.md +++ b/docs/ROADMAP.md @@ -1,6 +1,6 @@ # Roadmap -Date: 2026-09-07 +Date: 2026-09-08 Operational state: what ships today, what is queued next, what is still open, and what is known to be incomplete. This file is a summary and shrinks as work lands. `../AGENTS.md` routes to the @@ -23,6 +23,11 @@ accepted OpenAPI snapshot and rides one hand-written transport runtime. - **Assurance** — the suite runs on `net8.0`, `net9.0`, and `net10.0` across Linux, Windows, and macOS, plus `net472` on Windows, the fullest leg; `engineering/quality-gates.md` owns the gate a change must pass before it is called done. +- **Terminal lifetime corrections (D)** — consumer cancellation, send deadlines, local viewport + ordering, and shared cleanup are implemented and verified on Windows across the four runnable + targets. Linux and macOS live verification also passed on net8/net9/net10, including the persistent daemon + round trip and normal PTY reuse after read cancellation. `architecture/client-runtime.md` and + ADR-0023 own the contract. - **Packages** — `OpenCode.Sdk` and `OpenCode.Sdk.Extensions` pack at the single-sourced `VersionPrefix 0.1.0`, and every `master` push publishes a `0.1.0-nightly.*` build to GitHub Packages. NuGet.org publication is currently blocked by an upstream prefix reservation dispute; @@ -103,9 +108,9 @@ is revisited at each boundary. path the call must carry is invisible to any generated client; admitting it would mean inventing a path parameter the document does not declare (ADR-0013), and the upstream report is drafted. - **Two allocation follow-ups are queued behind a benchmark gate** — on `net472` and - `netstandard2.0` a response body over 1 MB costs one wire-sized copy, and `PtySession.ReadAsync` - allocates a fresh 16 KiB receive buffer per call. Both are measured rather than suspected, and - both are described for consumers in the README's Known Issues. + `netstandard2.0` a response body over 1 MB costs one wire-sized copy, and each terminal connection + allocates one 16 KiB receive buffer, reused across consumer reads. Both are described for consumers + in the README's Known Issues; pooling requires evidence under the current connection-lifetime harness. - **A half-open event stream is not detected.** A successful SSE body stays live until caller cancellation, server completion, or failure, so a connection whose peer is gone without closing hangs a consumer that supplied no cancellation of its own; ordinary resets, server exits, and diff --git a/docs/adr/0023-terminal-consumer-cancellation.md b/docs/adr/0023-terminal-consumer-cancellation.md new file mode 100644 index 00000000..da75c944 --- /dev/null +++ b/docs/adr/0023-terminal-consumer-cancellation.md @@ -0,0 +1,13 @@ +# Terminal consumer cancellation is independent of connection lifetime + +Date: 2026-09-08 + +A terminal consumer needs to stop waiting for output and later read or write on the same +attachment. Passing its cancellation token to physical WebSocket receive can abort that +connection. Both terminal families therefore use a connection-owned receiver and an unbounded +internal channel; public enumeration cancellation affects only the consumer. This is an SDK +contract, not a guarantee implied by `IAsyncEnumerable` or an upstream cancel/resume API. +The queue retains undelivered frames and may grow with an absent or slow consumer. A memory +limit and overflow policy require workload evidence before changing this contract. Connection +termination preserves queued output before reporting its outcome; explicit disposal abandons +unread output and ends owned work without requiring consumer progress. diff --git a/docs/architecture/client-runtime.md b/docs/architecture/client-runtime.md index fd7ead99..2aa3a4ec 100644 --- a/docs/architecture/client-runtime.md +++ b/docs/architecture/client-runtime.md @@ -1,6 +1,6 @@ # Client Runtime Architecture -Date: 2026-09-07 +Date: 2026-09-08 Canonical current rules for client construction, transport ownership, API errors, streams, and the local server launcher. Protocol and generated-model rules live in @@ -89,8 +89,8 @@ local server launcher. Protocol and generated-model rules live in ### PTY WebSocket session - `PtyClient.ConnectAsync` opens `PtySession`, the family's live working object: `ReadAsync` - enumerates `PtyFrame` values, `WriteAsync` sends input, and `DisposeAsync` closes. The session - owns its socket, so disposing it is the only way to end the connection. + enumerates `PtyFrame` values, `WriteAsync` sends input, and `DisposeAsync` closes. The caller + owns disposal of the session; peer closure and transport failure can also end the connection. - **Transport divergence.** This is one of the two SDK doors that do not ride the HTTP pipeline (the persistent PTY session is the other). The upgrade builds its own `ClientWebSocket`, so a caller-supplied `HttpClient`, its proxy, its handler chain, the redirect policy, the @@ -137,12 +137,33 @@ local server launcher. Protocol and generated-model rules live in exited and throws with the reason; because an exited PTY still upgrades cleanly, that failure surfaces on the first read rather than on connect. Any other close is an abnormal close, and a socket fault maps through `FailureClassification`'s PTY WebSocket phases. -- **Concurrency and disposal.** One session carries one active read enumeration — message - reassembly cannot be shared — and a second concurrent enumeration is refused with - `InvalidOperationException`. Sends are serialized behind a semaphore because the socket allows - one outstanding send. Caller cancellation stays `OperationCanceledException`. Disposal closes - gracefully under a bounded wait, then tears the socket down; it is idempotent, and a dispose - racing a pending read completes that read as a normal end rather than a fault. +- **Concurrency and reading (both terminal families).** One active public read enumeration is + allowed; a concurrent second reader throws `InvalidOperationException`. A connection-owned + receiver assembles complete messages into an unbounded `Channel` independently of + public enumeration. Canceling or disposing an enumerator ends only that consumer; another + enumeration may continue on the same otherwise healthy connection. The queue retains only + undelivered frames, and can grow when consumers are absent or slow. Connection termination + delivers the valid queued prefix before normal completion or the recorded transport/protocol + error. Later readers observe the same terminal outcome; consumer cancellation does not erase it. + A frame already selected for delivery may win a race with cancellation. +- **Explicit disposal (both terminal families).** The session owns its socket and receiver. + Disposal rejects new sends, wakes waiters, attempts bounded graceful close, then tears down + the socket and awaits owned I/O completion. It releases unread queued references without + waiting for consumer processing. Concurrent disposal calls await the same cleanup. Reads + after disposal are empty; writes after disposal throw `ObjectDisposedException`. A pending + read ends normally unless its own cancellation was already observed first. Expected graceful + close failures do not prevent hard cleanup. The graceful-close bounds are separate waits + for send admission and close-output; they are not one total budget or the send-timeout setting. +- **Send timeout (both terminal families).** `PtyConnectOptions.SendTimeout` and + `PersistentPtyConnectOptions.SendTimeout` configure a fixed total budget for each WebSocket + input or resize send, including serialization wait. Their default is 30 seconds. The option + accepts 1 millisecond through `TimeSpan.FromMilliseconds(int.MaxValue)`, inclusive, is captured + at connect, and never travels on the wire. Progress does not restart the budget. Expiry before + socket entry fails only that call; interruption during physical send terminates the attachment + because delivery is uncertain. Caller cancellation reports `OperationCanceledException`; + SDK expiry reports `OpenCodeTransportException` with a timeout cause. No send is retried. + This setting does not govern connection establishment, HTTP operations, idle time, command + execution, or disposal. ### Persistent PTY family @@ -192,9 +213,12 @@ local server launcher. Protocol and generated-model rules live in - **Input.** `WriteAsync(ReadOnlyMemory)` and `ResizeAsync(cols, rows)` each send one binary message in the framed input protocol's layout — `[type u8][cols u16 BE][rows u16 BE][data]`, type 1 for input and type 0 for a viewport change — which the SDK negotiates on every connection - and is the only protocol it writes. The viewport is a wire fact, not a caller preference: the - session starts it at the attachment's size and follows every resize frame the read enumeration - yields, whoever caused it, so a later write carries the size the server believes. Input from a + and is the only protocol it writes. The outbound viewport expresses this attachment's local + size intent. It starts at the attachment's size, and a successful `ResizeAsync` send changes + the size carried by later input. An inbound resize report remains available to the consumer + for rendering but does not overwrite that local intent. Header preparation, socket send, and + successful resize-state publication share one serialization boundary, so each input carries + a coherent size in send order. Input from a connection the server attached as an observer is accepted here and dropped there. - **Cursor.** `PersistentPtyConnectOptions.Cursor` is a relay to the connect query and nothing more. There is no live-only mode here: null replays from the oldest retained byte, and zero means @@ -223,8 +247,9 @@ local server launcher. Protocol and generated-model rules live in an unknown id, `shutdown` a 204, and `connect` a 4404 close. `ShutdownAsync` ends every terminal the daemon owns, not one. - **Shared core.** Both families' sessions run on the internal, family-neutral - `TerminalSocketCore` — receive with fragment reassembly, serialized sends, a bounded - graceful close, idempotent disposal, and one active read enumeration — and differ only behind + `TerminalSocketCore` — a connection-owned receiver with fragment reassembly, an + unbounded consumer queue, serialized sends with per-call deadlines, bounded graceful close, idempotent disposal, + and one active public read enumeration — and differ only behind 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 diff --git a/docs/architecture/platform-and-packaging.md b/docs/architecture/platform-and-packaging.md index e80988a4..5fb89618 100644 --- a/docs/architecture/platform-and-packaging.md +++ b/docs/architecture/platform-and-packaging.md @@ -1,6 +1,6 @@ # Platform and Packaging Architecture -Date: 2026-08-27 +Date: 2026-09-08 Canonical current rules for target frameworks, package boundaries, repository shape, versioning, distribution, dependencies, and licensing. @@ -54,6 +54,10 @@ Two separate API vintages shape the downlevel legs, and conflating them would mi ## Dependencies +- Terminal sessions use `System.Threading.Channels` for their internal receive queue. The + SDK declares that bridge package directly for `net472` and `netstandard2.0`; modern targets + use the inbox implementation. Channel types are not part of the public SDK API. + - Declare explicitly every package this repository's source uses directly, that appears on a public surface, or that is version-pinned for behavior; trust the transitive graph otherwise. - Downlevel bridge packages (`System.Memory`, `System.Buffers`, `System.Collections.Immutable`, diff --git a/src/OpenCode.Sdk/Internal/Abstractions/ITerminalDeadlineFactory.cs b/src/OpenCode.Sdk/Internal/Abstractions/ITerminalDeadlineFactory.cs new file mode 100644 index 00000000..c2a84f86 --- /dev/null +++ b/src/OpenCode.Sdk/Internal/Abstractions/ITerminalDeadlineFactory.cs @@ -0,0 +1,10 @@ +namespace OpenCode.Sdk.Internal.Abstractions; + +/// 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/PersistentPtyInputFrame.cs b/src/OpenCode.Sdk/Internal/PersistentPtyInputFrame.cs index 0005350d..285f2478 100644 --- a/src/OpenCode.Sdk/Internal/PersistentPtyInputFrame.cs +++ b/src/OpenCode.Sdk/Internal/PersistentPtyInputFrame.cs @@ -33,9 +33,24 @@ public static byte[] Encode(byte type, long cols, long rows, ReadOnlySpan var frame = new byte[HeaderLength + data.Length]; frame[0] = type; - BinaryPrimitives.WriteUInt16BigEndian(frame.AsSpan(1, 2), (ushort)cols); - BinaryPrimitives.WriteUInt16BigEndian(frame.AsSpan(3, 2), (ushort)rows); + WriteViewport(frame, cols, rows); data.CopyTo(frame.AsSpan(HeaderLength)); return frame; } + + /// Writes a validated viewport into an already owned frame under send serialization. + /// The complete frame buffer, including its five-byte header. + /// The viewport's column count. + /// The viewport's row count. + public static void WriteViewport(byte[] frame, long cols, long rows) + { + ArgumentNullException.ThrowIfNull(frame); + ArgumentOutOfRangeException.ThrowIfLessThan(cols, 1); + ArgumentOutOfRangeException.ThrowIfGreaterThan(cols, ushort.MaxValue); + ArgumentOutOfRangeException.ThrowIfLessThan(rows, 1); + ArgumentOutOfRangeException.ThrowIfGreaterThan(rows, ushort.MaxValue); + + BinaryPrimitives.WriteUInt16BigEndian(frame.AsSpan(1, 2), (ushort)cols); + BinaryPrimitives.WriteUInt16BigEndian(frame.AsSpan(3, 2), (ushort)rows); + } } diff --git a/src/OpenCode.Sdk/Internal/TerminalCancellationSource.cs b/src/OpenCode.Sdk/Internal/TerminalCancellationSource.cs new file mode 100644 index 00000000..0a379141 --- /dev/null +++ b/src/OpenCode.Sdk/Internal/TerminalCancellationSource.cs @@ -0,0 +1,26 @@ +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. + public Task CancelAsync() + { +#if NET8_0_OR_GREATER + return _source.CancelAsync(); +#else + // Polyfill 10.3's CancelAsync busy-waits for a queued Task.Run to start. Concurrent + // terminal shutdown can occupy every worker with that spin and starve cancellation. + // Our lifetime owner prevents source disposal until this task and owned I/O finish. + return Task.Run(_source.Cancel, CancellationToken.None); +#endif + } + + /// Releases the source after cancellation callbacks and owned I/O have settled. + public void Dispose() => _source.Dispose(); +} diff --git a/src/OpenCode.Sdk/Internal/TerminalConnectionOutcome.cs b/src/OpenCode.Sdk/Internal/TerminalConnectionOutcome.cs new file mode 100644 index 00000000..65a2c5d1 --- /dev/null +++ b/src/OpenCode.Sdk/Internal/TerminalConnectionOutcome.cs @@ -0,0 +1,8 @@ +namespace OpenCode.Sdk.Internal; + +/// The immutable first connection outcome shared by readers and writers. +internal sealed record TerminalConnectionOutcome(Exception? Failure) +{ + /// Gets the failure a send observes after this connection ended. + public Exception SendFailure => Failure ?? new OpenCodeTransportException("The opencode PTY WebSocket connection has ended."); +} diff --git a/src/OpenCode.Sdk/Internal/TerminalDeadlineFactory.cs b/src/OpenCode.Sdk/Internal/TerminalDeadlineFactory.cs new file mode 100644 index 00000000..e54a68bd --- /dev/null +++ b/src/OpenCode.Sdk/Internal/TerminalDeadlineFactory.cs @@ -0,0 +1,17 @@ +using OpenCode.Sdk.Internal.Abstractions; + +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/TerminalReceivePump.cs b/src/OpenCode.Sdk/Internal/TerminalReceivePump.cs new file mode 100644 index 00000000..321be20d --- /dev/null +++ b/src/OpenCode.Sdk/Internal/TerminalReceivePump.cs @@ -0,0 +1,192 @@ +using System.Diagnostics.CodeAnalysis; +using System.Net.WebSockets; +using System.Threading.Channels; + +namespace OpenCode.Sdk.Internal; + +/// Owns physical receive, fragment assembly, and undelivered frames for one connection. +/// The decoded frame type. +internal sealed class TerminalReceivePump : IAsyncDisposable + where TFrame : class +{ + private readonly Channel _frames = Channel.CreateUnbounded(new UnboundedChannelOptions + { + SingleReader = false, + AllowSynchronousContinuations = false, + }); + private readonly TerminalCancellationSource _lifetime = new(); + private readonly Lock _sync = new(); + private readonly Task _completion; + private bool _abandoned; + private TerminalConnectionOutcome? _outcome; + + public TerminalReceivePump( + ITerminalWebSocket socket, + ITerminalFrameDecoder decoder, + ITerminalClosePolicy closePolicy, + Func onEnded) + { + _completion = ReceiveAsync(socket, decoder, closePolicy, onEnded); + } + + /// Gets whether the connection has reached its terminal outcome. + public TerminalConnectionOutcome? Outcome + { + get + { + lock (_sync) + { + return _outcome; + } + } + } + + /// Gets the first terminal failure, if any. + public Exception? Failure + { + get + { + lock (_sync) + { + return _outcome?.Failure; + } + } + } + + /// Selects a frame atomically with cancellation observation and disposal. + public bool TryRead(CancellationToken cancellationToken, [NotNullWhen(true)] out TFrame? frame) + { + lock (_sync) + { + cancellationToken.ThrowIfCancellationRequested(); + frame = null; + return !_abandoned && _frames.Reader.TryRead(out frame); + } + } + + /// Waits only for consumer progress; this token never reaches physical receive. + public ValueTask WaitToReadAsync(CancellationToken cancellationToken) => + _frames.Reader.WaitToReadAsync(cancellationToken); + + /// Records the first terminal outcome and wakes consumers after the queued prefix. + public void Complete(Exception? failure) + { + lock (_sync) + { + if (_outcome is not null) + { + return; + } + + _outcome = new TerminalConnectionOutcome(failure); + _ = _frames.Writer.TryComplete(); + } + } + + /// Prevents publication and releases unread frames without waiting for a consumer. + public void Abandon() + { + lock (_sync) + { + _abandoned = true; + _outcome = new TerminalConnectionOutcome(null); + _ = _frames.Writer.TryComplete(); + while (_frames.Reader.TryRead(out _)) + { + // Explicit disposal abandons only frames that no reader has selected. + } + } + } + + /// Stops and observes the owned receiver before releasing its cancellation source. + public async ValueTask DisposeAsync() + { + try + { + await _lifetime.CancelAsync().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); +#pragma warning restore VSTHRD003 + } + finally + { + _lifetime.Dispose(); + } + } + + private async Task ReceiveAsync( + ITerminalWebSocket socket, + ITerminalFrameDecoder decoder, + ITerminalClosePolicy closePolicy, + Func onEnded) + { + try + { + await ReceiveMessagesAsync(socket, decoder, closePolicy).ConfigureAwait(false); + } + catch (Exception exception) when (FailureClassification.Handles(exception, FailurePhase.PtyWebSocketRead)) + { + Complete(FailureClassification.Map(exception, FailurePhase.PtyWebSocketRead, CancellationToken.None)); + } + catch (OpenCodeTransportException exception) + { + Complete(exception); + } + catch (Exception exception) + { + Complete(exception); + throw; + } + finally + { + await onEnded().ConfigureAwait(false); + } + } + + private async Task ReceiveMessagesAsync( + ITerminalWebSocket socket, + ITerminalFrameDecoder decoder, + ITerminalClosePolicy closePolicy) + { + var buffer = new byte[TerminalSocketBounds.ReceiveBufferSize]; + var segment = new ArraySegment(buffer); + PtyMessageAssembler? assembly = null; + while (Outcome is null) + { + var received = await socket.ReceiveAsync(segment, _lifetime.Token).ConfigureAwait(false); + if (received.MessageType is WebSocketMessageType.Close) + { + Complete(closePolicy.Map(socket.CloseStatus, socket.CloseStatusDescription)); + return; + } + + if (!received.EndOfMessage) + { + assembly ??= new PtyMessageAssembler(); + assembly.Append(buffer, received.Count); + continue; + } + + TFrame frame; + if (assembly is null || assembly.Length is 0) + { + frame = decoder.Decode(received.MessageType, buffer, received.Count); + } + else + { + assembly.Append(buffer, received.Count); + frame = decoder.Decode(received.MessageType, assembly.Buffer, assembly.Length); + assembly.Reset(); + } + + lock (_sync) + { + if (_outcome is null) + { + _ = _frames.Writer.TryWrite(frame); + } + } + } + } +} diff --git a/src/OpenCode.Sdk/Internal/TerminalSendActions.cs b/src/OpenCode.Sdk/Internal/TerminalSendActions.cs new file mode 100644 index 00000000..a9c717e6 --- /dev/null +++ b/src/OpenCode.Sdk/Internal/TerminalSendActions.cs @@ -0,0 +1,9 @@ +namespace OpenCode.Sdk.Internal; + +/// +/// Family-owned work that shares the socket's send ordering: prepare a message immediately +/// before its send, and publish local state only after that send succeeds. +/// +/// Prepares the payload while holding the send gate. +/// Publishes state after a successful send, before releasing the gate. +internal sealed record TerminalSendActions(Action? Prepare = null, Action? OnSent = null); diff --git a/src/OpenCode.Sdk/Internal/TerminalSendDeadline.cs b/src/OpenCode.Sdk/Internal/TerminalSendDeadline.cs new file mode 100644 index 00000000..17630c0e --- /dev/null +++ b/src/OpenCode.Sdk/Internal/TerminalSendDeadline.cs @@ -0,0 +1,59 @@ +using OpenCode.Sdk.Internal.Abstractions; + +namespace OpenCode.Sdk.Internal; + +/// Owns one fixed send budget and the caller/connection/timeout failure precedence. +internal sealed class TerminalSendDeadline : IDisposable +{ + private readonly CancellationToken _caller; + private readonly CancellationTokenSource _deadline; + private readonly CancellationTokenSource _operation; + private readonly Type _owner; + + public TerminalSendDeadline(Type owner, TimeSpan timeout, ITerminalDeadlineFactory factory, CancellationToken caller) + { + _owner = owner; + _caller = caller; + _deadline = factory.Create(timeout); + _operation = CancellationTokenSource.CreateLinkedTokenSource(caller, _deadline.Token); + } + + /// Gets the combined physical-send cancellation token. + public CancellationToken Token => _operation.Token; + + /// Classifies a failed send without replacing an earlier connection outcome. + public Exception Map(Exception exception, bool queued, bool disposed, TerminalConnectionOutcome? outcome) + { + if (_caller.IsCancellationRequested) + { + return new OperationCanceledException("The opencode PTY WebSocket send was canceled.", exception, _caller); + } + + if (queued && disposed) + { + return new ObjectDisposedException(_owner.FullName); + } + + if (outcome is not null) + { + return outcome.SendFailure; + } + + if (_deadline.IsCancellationRequested) + { + var message = queued + ? "The opencode PTY WebSocket send timed out while waiting for send admission." + : "The opencode PTY WebSocket send timed out during the socket write; delivery is uncertain."; + return new OpenCodeTransportException(message, new TimeoutException(message, exception)); + } + + return FailureClassification.Map(exception, FailurePhase.PtyWebSocketWrite, _caller); + } + + /// Releases the operation's registrations and timer once its send settles. + public void Dispose() + { + _operation.Dispose(); + _deadline.Dispose(); + } +} diff --git a/src/OpenCode.Sdk/Internal/TerminalSendQueue.cs b/src/OpenCode.Sdk/Internal/TerminalSendQueue.cs new file mode 100644 index 00000000..238eb9e9 --- /dev/null +++ b/src/OpenCode.Sdk/Internal/TerminalSendQueue.cs @@ -0,0 +1,43 @@ +namespace OpenCode.Sdk.Internal; + +/// Serializes physical writes and graceful close behind one connection gate. +internal sealed class TerminalSendQueue(ITerminalWebSocket socket) : IDisposable +{ + private readonly SemaphoreSlim _gate = new(1, 1); + + /// Waits for the next physical-send turn. + public Task WaitAsync(CancellationToken cancellationToken) => _gate.WaitAsync(cancellationToken); + + /// Releases the physical-send turn. + public void Release() => _gate.Release(); + + /// Releases the gate after its owner has joined every admitted send and close. + public void Dispose() => _gate.Dispose(); + + /// Attempts close with separate send-admission and close-output phase budgets. + public async Task TryCloseAsync() + { + if (!await _gate.WaitAsync(TerminalSocketBounds.GracefulCloseTimeout, CancellationToken.None).ConfigureAwait(false)) + { + return false; + } + + using var timeout = new CancellationTokenSource(TerminalSocketBounds.GracefulCloseTimeout); + try + { + await socket.CloseOutputAsync(timeout.Token).ConfigureAwait(false); + return true; + } + catch (Exception exception) when ( + FailureClassification.Handles(exception, FailurePhase.PtyWebSocketWrite) || + exception is InvalidOperationException) + { + // Expected graceful-close failures leave unconditional hard teardown to the owner. + return false; + } + finally + { + Release(); + } + } +} diff --git a/src/OpenCode.Sdk/Internal/TerminalSendTimeout.cs b/src/OpenCode.Sdk/Internal/TerminalSendTimeout.cs new file mode 100644 index 00000000..682a8129 --- /dev/null +++ b/src/OpenCode.Sdk/Internal/TerminalSendTimeout.cs @@ -0,0 +1,18 @@ +namespace OpenCode.Sdk.Internal; + +/// A finite connection send budget representable by every supported cancellation timer. +internal sealed record TerminalSendTimeout +{ + public TerminalSendTimeout(TimeSpan value) + { + ArgumentOutOfRangeException.ThrowIfLessThan(value, TimeSpan.FromMilliseconds(1)); + ArgumentOutOfRangeException.ThrowIfGreaterThan(value, TimeSpan.FromMilliseconds(int.MaxValue)); + Value = value; + } + + /// Gets the validated duration. + public TimeSpan Value { get; } + + /// Gets the shared immutable default budget. + public static TerminalSendTimeout Default { get; } = new(TerminalSocketBounds.DefaultSendTimeout); +} diff --git a/src/OpenCode.Sdk/Internal/TerminalSocketBounds.cs b/src/OpenCode.Sdk/Internal/TerminalSocketBounds.cs index 5b0aa889..29a09aca 100644 --- a/src/OpenCode.Sdk/Internal/TerminalSocketBounds.cs +++ b/src/OpenCode.Sdk/Internal/TerminalSocketBounds.cs @@ -7,6 +7,9 @@ namespace OpenCode.Sdk.Internal; /// internal static class TerminalSocketBounds { + /// The fixed total budget for one send, including serialization wait. + public static readonly TimeSpan DefaultSendTimeout = TimeSpan.FromSeconds(30); + /// /// A message either family sends can be far larger than one receive. Taking it in fixed /// slices keeps the per-session buffer small whatever the message size is; the read loop diff --git a/src/OpenCode.Sdk/Internal/TerminalSocketCore.cs b/src/OpenCode.Sdk/Internal/TerminalSocketCore.cs index 7c23d7ed..cc828282 100644 --- a/src/OpenCode.Sdk/Internal/TerminalSocketCore.cs +++ b/src/OpenCode.Sdk/Internal/TerminalSocketCore.cs @@ -1,5 +1,6 @@ using System.Net.WebSockets; using System.Runtime.CompilerServices; +using OpenCode.Sdk.Internal.Abstractions; namespace OpenCode.Sdk.Internal; @@ -13,13 +14,19 @@ namespace OpenCode.Sdk.Internal; internal sealed class TerminalSocketCore : IAsyncDisposable where TFrame : class { - private readonly ITerminalClosePolicy _closePolicy; - private readonly ITerminalFrameDecoder _decoder; + private readonly Lock _lifecycle = new(); + private readonly TerminalReceivePump _receiver; private readonly Type _owner; - private readonly SemaphoreSlim _sendGate = new(1, 1); + private readonly TerminalCancellationSource _sendEnded = new(); + private readonly CancellationToken _sendEndedToken; + private readonly TerminalSendQueue _sendGate; private readonly ITerminalWebSocket _socket; private int _disposed; private int _reading; + private int _socketDisposed; + private Task? _disposal; + private TaskCompletionSource? _sendsDrained; + private int _activeSends; public TerminalSocketCore( ITerminalWebSocket socket, @@ -33,19 +40,27 @@ public TerminalSocketCore( ArgumentNullException.ThrowIfNull(owner); _socket = socket; - _decoder = decoder; - _closePolicy = closePolicy; + _sendGate = new TerminalSendQueue(socket); _owner = owner; + _sendEndedToken = _sendEnded.Token; + _receiver = new TerminalReceivePump(socket, decoder, closePolicy, _sendEnded.CancelAsync); } /// Gets whether the core has been disposed. public bool IsDisposed => Volatile.Read(ref _disposed) is 1; + /// 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: message reassembly cannot be shared, so a second + /// carries one active enumeration: undelivered frames have one consumer, so a second /// concurrent enumeration is refused. Reading after disposal is not an error: unlike - /// , which throws once disposed, the enumeration simply ends empty. + /// , + /// which throws once disposed, the enumeration simply ends empty. /// /// The cancellation token ending the read. /// The frames, in the order the server sent them. @@ -60,14 +75,68 @@ public IAsyncEnumerable ReadAsync(CancellationToken cancellationToken) = /// The message type to send them as. /// The cancellation token. /// A task that completes once the message is sent. - public async Task SendAsync( + public Task SendAsync( ArraySegment payload, WebSocketMessageType messageType, + CancellationToken cancellationToken) => SendAsync(() => payload, messageType, null, cancellationToken); + + /// Sends a message with family-owned preparation and publication under the same gate. + /// Copies or encodes the owned message before the first asynchronous wait. + /// The message's wire type. + /// The work that shares the message's send order. + /// The cancellation token. + /// A task that completes after the send and its successful publication. + public async Task SendAsync( + Func> createPayload, + WebSocketMessageType messageType, + TerminalSendActions? actions, CancellationToken cancellationToken) { - ObjectDisposedException.ThrowIf(IsDisposed, _owner); + lock (_lifecycle) + { + ObjectDisposedException.ThrowIf(IsDisposed, _owner); + ThrowIfSendFailed(); + if (_activeSends++ is 0) + { + _sendsDrained = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + } + } + + try + { + await SendCoreAsync(createPayload, messageType, actions, cancellationToken).ConfigureAwait(false); + } + finally + { + lock (_lifecycle) + { + if (--_activeSends is 0) + { + _ = _sendsDrained!.TrySetResult(true); + } + } + } + } + + private async Task SendCoreAsync( + Func> createPayload, + WebSocketMessageType messageType, + TerminalSendActions? actions, + CancellationToken cancellationToken) + { + + using var deadline = new TerminalSendDeadline(_owner, SendTimeout, DeadlineFactory, cancellationToken); + var payload = createPayload(); + using var admission = CancellationTokenSource.CreateLinkedTokenSource(deadline.Token, _sendEndedToken); + try + { + await _sendGate.WaitAsync(admission.Token).ConfigureAwait(false); + } + catch (OperationCanceledException exception) + { + throw deadline.Map(exception, queued: true, IsDisposed, _receiver.Outcome); + } - await _sendGate.WaitAsync(cancellationToken).ConfigureAwait(false); try { // Re-checked behind the gate: a disposal can land while a queued send waits. It sits @@ -75,19 +144,34 @@ public async Task SendAsync( // phase's fault set, so a refusal raised inside it would be remapped into a transport // failure instead of reaching the caller as the misuse it is. ObjectDisposedException.ThrowIf(IsDisposed, _owner); + ThrowIfSendFailed(); + try + { + deadline.Token.ThrowIfCancellationRequested(); + } + catch (OperationCanceledException exception) + { + throw deadline.Map(exception, queued: true, IsDisposed, _receiver.Outcome); + } + + actions?.Prepare?.Invoke(); try { - await _socket.SendAsync(payload, messageType, cancellationToken).ConfigureAwait(false); + await _socket.SendAsync(payload, messageType, deadline.Token).ConfigureAwait(false); } catch (Exception exception) when (FailureClassification.Handles(exception, FailurePhase.PtyWebSocketWrite)) { - throw FailureClassification.Map(exception, FailurePhase.PtyWebSocketWrite, cancellationToken); + var failure = deadline.Map(exception, queued: false, IsDisposed, _receiver.Outcome); + await FailSendAsync(failure).ConfigureAwait(false); + throw failure; } + + actions?.OnSent?.Invoke(); } finally { - _ = _sendGate.Release(); + _sendGate.Release(); } } @@ -97,127 +181,123 @@ public async Task SendAsync( /// waiting on the socket ends as a normal end rather than a fault. /// /// A task that completes once the connection is closed. - public async ValueTask DisposeAsync() + public ValueTask DisposeAsync() { - if (Interlocked.Exchange(ref _disposed, 1) is 1) + lock (_lifecycle) { - return; + if (_disposal is null) + { + Volatile.Write(ref _disposed, 1); + _receiver.Abandon(); + _disposal = DisposeCoreAsync(); + } + + return new ValueTask(_disposal); } + } + + private async Task DisposeCoreAsync() + { + await _sendEnded.CancelAsync().ConfigureAwait(false); try { - _ = await TryCloseAsync().ConfigureAwait(false); + if (Volatile.Read(ref _socketDisposed) is 0) + { + _ = await _sendGate.TryCloseAsync().ConfigureAwait(false); + } } finally { // Unconditional: whatever the graceful close does, a disposal that has already // latched _disposed must never leave the socket alive behind it. - _socket.Dispose(); + try + { + DisposeSocket(); + } + finally + { + try + { + await _receiver.DisposeAsync().ConfigureAwait(false); + } + finally + { + await AwaitSendsAsync().ConfigureAwait(false); + _sendGate.Dispose(); + _sendEnded.Dispose(); + } + } } - // _sendGate is deliberately not disposed. SemaphoreSlim only needs disposal once - // AvailableWaitHandle has been read, which this class never does, and disposing it here - // would break the two things disposal must not break: an in-flight write's Release would - // throw over its own mapped failure, and a queued writer would never be released at all, - // because Dispose does not complete pending async waiters. } - private async Task TryCloseAsync() + private Task AwaitSendsAsync() { - // A close-output frame is a send, and the socket allows one outstanding send, so the - // graceful close takes the same gate every write takes instead of racing an in-flight - // one. The wait is bounded so a stuck send cannot stall a disposal indefinitely. - if (!await _sendGate.WaitAsync(TerminalSocketBounds.GracefulCloseTimeout).ConfigureAwait(false)) + lock (_lifecycle) { - return false; + return _sendsDrained?.Task ?? Task.CompletedTask; } + } - using var timeout = new CancellationTokenSource(TerminalSocketBounds.GracefulCloseTimeout); - try - { - await _socket.CloseOutputAsync(timeout.Token).ConfigureAwait(false); - return true; - } - catch (Exception exception) when ( - FailureClassification.Handles(exception, FailurePhase.PtyWebSocketWrite) || - exception is InvalidOperationException) + private void ThrowIfSendFailed() + { + if (_receiver.Outcome is { } outcome) { - // Best effort, and deliberately wider than the write plane's fault set: a socket that - // refuses a close for a state reason it reports as InvalidOperationException must not - // escape a disposal. The hard teardown follows unconditionally, and a caller closing a - // connection has nothing left to do about a close frame that never left. - return false; + throw outcome.SendFailure; } - finally + } + + 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); + DisposeSocket(); + } + + private void DisposeSocket() + { + if (Interlocked.Exchange(ref _socketDisposed, 1) is 0) { - _ = _sendGate.Release(); + _socket.Dispose(); } } private async IAsyncEnumerable ReadCoreAsync( [EnumeratorCancellation] CancellationToken cancellationToken) { + if (IsDisposed) + { + yield break; + } + if (Interlocked.Exchange(ref _reading, 1) is 1) { throw new InvalidOperationException( - $"A '{_owner.Name}' carries one active read enumeration; message reassembly cannot be shared across two."); + $"A '{_owner.Name}' carries one active read enumeration; undelivered frames have one consumer."); } - var buffer = new byte[TerminalSocketBounds.ReceiveBufferSize]; - var segment = new ArraySegment(buffer); - PtyMessageAssembler? assembly = null; try { while (true) { - PtyReceiveResult received; - - // A yield cannot sit inside a try that catches, so the receive is guarded alone. - try - { - received = await _socket.ReceiveAsync(segment, cancellationToken).ConfigureAwait(false); - } - catch (Exception exception) when (FailureClassification.Handles(exception, FailurePhase.PtyWebSocketRead)) + if (_receiver.TryRead(cancellationToken, out var frame)) { - if (!cancellationToken.IsCancellationRequested && IsDisposed) - { - // The caller tore the session down; the socket it disposed going quiet is - // the end it asked for, not a fault to report back. - break; - } - - throw FailureClassification.Map(exception, FailurePhase.PtyWebSocketRead, cancellationToken); + yield return frame; + continue; } - if (received.MessageType is WebSocketMessageType.Close) + if (!await _receiver.WaitToReadAsync(cancellationToken).ConfigureAwait(false)) { - var failure = _closePolicy.Map(_socket.CloseStatus, _socket.CloseStatusDescription); - if (failure is not null) + if (_receiver.Failure is { } failure) { throw failure; } - break; - } - - if (!received.EndOfMessage) - { - assembly ??= new PtyMessageAssembler(); - assembly.Append(buffer, received.Count); - continue; + yield break; } - - // An unfragmented message decodes straight from the receive buffer. - if (assembly is null || assembly.Length is 0) - { - yield return _decoder.Decode(received.MessageType, buffer, received.Count); - continue; - } - - assembly.Append(buffer, received.Count); - var assembled = _decoder.Decode(received.MessageType, assembly.Buffer, assembly.Length); - assembly.Reset(); - yield return assembled; } } finally diff --git a/src/OpenCode.Sdk/OpenCode.Sdk.csproj b/src/OpenCode.Sdk/OpenCode.Sdk.csproj index 2890c914..83b32a04 100644 --- a/src/OpenCode.Sdk/OpenCode.Sdk.csproj +++ b/src/OpenCode.Sdk/OpenCode.Sdk.csproj @@ -46,5 +46,11 @@ + + + + + + diff --git a/src/OpenCode.Sdk/PersistentPtys/PersistentPtyClient.cs b/src/OpenCode.Sdk/PersistentPtys/PersistentPtyClient.cs index 2a7cbe78..2d65060f 100644 --- a/src/OpenCode.Sdk/PersistentPtys/PersistentPtyClient.cs +++ b/src/OpenCode.Sdk/PersistentPtys/PersistentPtyClient.cs @@ -136,7 +136,9 @@ public virtual async Task ConnectAsync(PersistentPtyConnec await socket .ConnectAsync(address, ptyId, PersistentPtyUpgradeFailurePolicy.Instance, cancellationToken) .ConfigureAwait(false); - var session = await PersistentPtySession.AttachAsync(socket, ptyId, cancellationToken).ConfigureAwait(false); + var session = await PersistentPtySession.AttachAsync( + socket, ptyId, options?.SendTimeout ?? TerminalSocketBounds.DefaultSendTimeout, cancellationToken) + .ConfigureAwait(false); socket = null; return session; } diff --git a/src/OpenCode.Sdk/PersistentPtys/PersistentPtyConnectOptions.cs b/src/OpenCode.Sdk/PersistentPtys/PersistentPtyConnectOptions.cs index efaad985..b643c9be 100644 --- a/src/OpenCode.Sdk/PersistentPtys/PersistentPtyConnectOptions.cs +++ b/src/OpenCode.Sdk/PersistentPtys/PersistentPtyConnectOptions.cs @@ -1,12 +1,26 @@ +using OpenCode.Sdk.Internal; + namespace OpenCode.Sdk; /// -/// Shapes one persistent PTY WebSocket connection. Every member rides the upgrade URL's query; -/// the framed input protocol is not a member because this SDK speaks only that protocol and -/// negotiates it on every connection. +/// Shapes one persistent PTY WebSocket connection and its local send budget. The send timeout +/// stays in the SDK; the other options ride the upgrade query. The SDK always negotiates framed input. /// public sealed record PersistentPtyConnectOptions { + private readonly TerminalSendTimeout _sendTimeout = TerminalSendTimeout.Default; + + /// + /// Gets the fixed total budget for one WebSocket input or resize send, including serialization wait. + /// The default is 30 seconds; this does not limit connection establishment or command execution. + /// + /// The budget is below one millisecond or above int.MaxValue milliseconds. + public TimeSpan SendTimeout + { + get => _sendTimeout.Value; + init => _sendTimeout = new TerminalSendTimeout(value); + } + /// /// The largest cursor the server accepts: it validates against the JavaScript safe-integer /// range and answers HTTP 400 before the upgrade for anything outside it. The SDK refuses the diff --git a/src/OpenCode.Sdk/PersistentPtys/PersistentPtySession.cs b/src/OpenCode.Sdk/PersistentPtys/PersistentPtySession.cs index a717faea..2c1c582b 100644 --- a/src/OpenCode.Sdk/PersistentPtys/PersistentPtySession.cs +++ b/src/OpenCode.Sdk/PersistentPtys/PersistentPtySession.cs @@ -1,7 +1,7 @@ using System.Globalization; using System.Net.WebSockets; -using System.Runtime.CompilerServices; using OpenCode.Sdk.Internal; +using OpenCode.Sdk.Models; namespace OpenCode.Sdk; @@ -18,8 +18,7 @@ public class PersistentPtySession : IAsyncDisposable private readonly PersistentPtyAttachment? _attachment; private readonly TerminalSocketCore? _core; - private long _cols; - private long _rows; + private PersistentPtyInfoSize? _viewport; internal PersistentPtySession(TerminalSocketCore core, PersistentPtyAttachment attachment) { @@ -28,8 +27,7 @@ internal PersistentPtySession(TerminalSocketCore core, Persi _core = core; _attachment = attachment; - _cols = attachment.Info.Size.Cols; - _rows = attachment.Info.Size.Rows; + _viewport = attachment.Info.Size; } /// @@ -49,18 +47,21 @@ protected PersistentPtySession() /// /// Reads the frames the server sends until it closes the connection normally. The retained /// replay, when any, arrives as output frames bracketed by - /// ; a resize the server reports updates the - /// viewport later writes carry, whoever caused it. One session carries one active enumeration: - /// message reassembly cannot be shared, so a second concurrent enumeration is refused. Reading + /// . A server resize reports rendering size; + /// it does not change the local viewport later writes carry. One session carries one active enumeration: + /// undelivered frames have one consumer, so a second concurrent enumeration is refused. Reading /// after disposal is not an error: unlike , which throws once disposed, - /// the enumeration simply ends empty. + /// the enumeration simply ends empty. Canceling or disposing an enumerator ends only that + /// consumer; another enumeration may continue on the same connection. Unread frames remain + /// queued without a fixed memory limit. Peer closure drains that queue before its outcome; + /// explicit session disposal abandons unread frames and awaits owned I/O cleanup. /// /// The cancellation token ending the read. /// The frames, in the order the server sent them. /// A read enumeration is already active on this session. /// The connection failed, closed abnormally, or carried an unreadable control frame. public virtual IAsyncEnumerable ReadAsync(CancellationToken cancellationToken = default) => - ReadCoreAsync(Core, cancellationToken); + Core.ReadAsync(cancellationToken); /// /// Writes terminal input as one framed binary message carrying the current viewport. The @@ -80,8 +81,8 @@ public virtual Task WriteAsync(ReadOnlyMemory input, CancellationToken can /// /// Resizes the terminal through a control frame and records the viewport later writes carry. /// The server answers with a on the read enumeration. - /// The recorded viewport moves only once the control frame has left: a send that failed - /// resized nothing, so a later write must not carry a size the server never saw. + /// The recorded viewport changes only after the control send succeeds, before the next + /// send begins. Send completion does not acknowledge that the server applied the resize. /// /// The new column count; 1 through 65535. /// The new row count; 1 through 65535. @@ -129,9 +130,22 @@ public virtual async ValueTask DisposeAsync() /// 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, + CancellationToken cancellationToken) => + AttachAsync(socket, ptyId, 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 fixed per-send budget. + /// The token bounding attachment. + /// The attached session. internal static async Task AttachAsync( ITerminalWebSocket socket, string ptyId, + TimeSpan sendTimeout, CancellationToken cancellationToken) { // Ownership stays local until the very end: an attach that never produced a session must @@ -144,7 +158,10 @@ internal static async Task AttachAsync( socket, PersistentPtyFrameDecoder.Instance, PersistentPtyClosePolicy.Instance, - typeof(PersistentPtySession)); + typeof(PersistentPtySession)) + { + SendTimeout = sendTimeout, + }; var attachment = await ReadAttachmentAsync(core, ptyId, cancellationToken).ConfigureAwait(false); var attachedSession = new PersistentPtySession(core, attachment); core = null; @@ -197,39 +214,30 @@ private static async Task ReadAttachmentAsync( private TerminalSocketCore Core => _core ?? throw MockSeam.CreateError("PersistentPtySession", "WebSocket"); - private async IAsyncEnumerable ReadCoreAsync( - TerminalSocketCore core, - [EnumeratorCancellation] CancellationToken cancellationToken) - { - await foreach (var frame in core.ReadAsync(cancellationToken).ConfigureAwait(false)) - { - if (frame is PersistentPtyResizedFrame resized) - { - // The viewport is a wire fact, not a caller preference: whoever resized the - // terminal, the next framed write must carry the size the server now believes. - Volatile.Write(ref _cols, resized.Cols); - Volatile.Write(ref _rows, resized.Rows); - } - - yield return frame; - } - } - - private async Task ResizeCoreAsync(int cols, int rows, CancellationToken cancellationToken) + private Task ResizeCoreAsync(int cols, int rows, CancellationToken cancellationToken) { var core = Core; - var frame = PersistentPtyInputFrame.Encode(PersistentPtyInputFrame.ControlType, cols, rows, []); - await core.SendAsync(new ArraySegment(frame), WebSocketMessageType.Binary, cancellationToken) - .ConfigureAwait(false); - - Volatile.Write(ref _cols, cols); - Volatile.Write(ref _rows, rows); + var viewport = new PersistentPtyInfoSize { Cols = cols, Rows = rows }; + var actions = new TerminalSendActions(OnSent: () => _viewport = viewport); + return core.SendAsync( + () => new ArraySegment(PersistentPtyInputFrame.Encode(PersistentPtyInputFrame.ControlType, cols, rows, [])), + WebSocketMessageType.Binary, actions, cancellationToken); } private Task SendFrameAsync(byte type, ReadOnlyMemory data, CancellationToken cancellationToken) { var core = Core; - var frame = PersistentPtyInputFrame.Encode(type, Volatile.Read(ref _cols), Volatile.Read(ref _rows), data.Span); - return core.SendAsync(new ArraySegment(frame), WebSocketMessageType.Binary, cancellationToken); + // Copy caller-owned input now; fill the viewport only after this send acquires its turn. + byte[]? frame = null; + var actions = new TerminalSendActions(Prepare: () => + { + var viewport = _viewport!; + PersistentPtyInputFrame.WriteViewport(frame!, viewport.Cols, viewport.Rows); + }); + return core.SendAsync(() => + { + frame = PersistentPtyInputFrame.Encode(type, 1, 1, data.Span); + return new ArraySegment(frame); + }, WebSocketMessageType.Binary, actions, cancellationToken); } } diff --git a/src/OpenCode.Sdk/Ptys/PtyClient.cs b/src/OpenCode.Sdk/Ptys/PtyClient.cs index ad397d3e..fb47f618 100644 --- a/src/OpenCode.Sdk/Ptys/PtyClient.cs +++ b/src/OpenCode.Sdk/Ptys/PtyClient.cs @@ -129,7 +129,7 @@ public virtual async Task ConnectAsync(PtyConnectOptions? options = await socket .ConnectAsync(address, ptyId, PtyUpgradeFailurePolicy.Instance, cancellationToken) .ConfigureAwait(false); - return new PtySession(socket); + return new PtySession(socket, options?.SendTimeout); } catch { diff --git a/src/OpenCode.Sdk/Ptys/PtyConnectOptions.cs b/src/OpenCode.Sdk/Ptys/PtyConnectOptions.cs index 79562938..e85424ed 100644 --- a/src/OpenCode.Sdk/Ptys/PtyConnectOptions.cs +++ b/src/OpenCode.Sdk/Ptys/PtyConnectOptions.cs @@ -1,11 +1,26 @@ +using OpenCode.Sdk.Internal; + namespace OpenCode.Sdk; /// -/// Shapes one PTY WebSocket connection. Both members ride the upgrade URL's query, so the scope -/// they resolve must agree with the scope the token door resolved. +/// Shapes one PTY WebSocket connection and its local send budget. The location and cursor ride +/// the upgrade query; the send timeout stays in the SDK. /// public sealed record PtyConnectOptions { + private readonly TerminalSendTimeout _sendTimeout = TerminalSendTimeout.Default; + + /// + /// Gets the fixed total budget for one WebSocket send, including its serialization wait. + /// The default is 30 seconds; this does not limit connection establishment or command execution. + /// + /// The budget is below one millisecond or above int.MaxValue milliseconds. + public TimeSpan SendTimeout + { + get => _sendTimeout.Value; + init => _sendTimeout = new TerminalSendTimeout(value); + } + /// /// The largest cursor the server accepts: it validates against the JavaScript safe-integer /// range and silently ignores anything outside it, which would turn a resume into a full diff --git a/src/OpenCode.Sdk/Ptys/PtySession.cs b/src/OpenCode.Sdk/Ptys/PtySession.cs index 305fd704..e604aced 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) + internal PtySession(ITerminalWebSocket socket, TimeSpan? sendTimeout = null) { ArgumentNullException.ThrowIfNull(socket); @@ -22,7 +22,10 @@ internal PtySession(ITerminalWebSocket socket) socket, PtyFrameDecoder.Instance, PtyClosePolicy.Instance, - typeof(PtySession)); + typeof(PtySession)) + { + SendTimeout = sendTimeout ?? TerminalSocketBounds.DefaultSendTimeout, + }; } /// @@ -36,9 +39,13 @@ protected PtySession() /// /// Reads the frames the server sends until it closes the connection normally. One session - /// carries one active enumeration: message reassembly cannot be shared, so a second + /// carries one active enumeration: undelivered frames have one consumer, so a second /// concurrent enumeration is refused. Reading after disposal is not an error: unlike /// , which throws once disposed, the enumeration simply ends empty. + /// Canceling or disposing an enumerator ends only that consumer; another enumeration may + /// continue on the same connection. Unread frames remain queued without a fixed memory limit. + /// Peer closure drains the queue before its outcome; explicit session disposal abandons + /// unread frames and awaits owned I/O cleanup. /// /// The cancellation token ending the read. /// The frames, in the order the server sent them. @@ -65,8 +72,8 @@ public virtual Task WriteAsync(string input, CancellationToken cancellationToken ArgumentNullException.ThrowIfNull(input); var core = Core; - var payload = new ArraySegment(Encoding.UTF8.GetBytes(input)); - return core.SendAsync(payload, WebSocketMessageType.Text, cancellationToken); + return core.SendAsync(() => new ArraySegment(Encoding.UTF8.GetBytes(input)), + WebSocketMessageType.Text, null, cancellationToken); } /// diff --git a/tests/OpenCode.Sdk.Performance.Tests/Benchmarks/PersistentPtySessionReadBenchmarks.cs b/tests/OpenCode.Sdk.Performance.Tests/Benchmarks/PersistentPtySessionReadBenchmarks.cs index 18ef8d5e..af5dd60b 100644 --- a/tests/OpenCode.Sdk.Performance.Tests/Benchmarks/PersistentPtySessionReadBenchmarks.cs +++ b/tests/OpenCode.Sdk.Performance.Tests/Benchmarks/PersistentPtySessionReadBenchmarks.cs @@ -54,6 +54,7 @@ public class PersistentPtySessionReadBenchmarks : IAsyncDisposable private (WebSocketMessageType Type, byte[] Payload)[] _messages = []; private CannedTerminalWebSocket? _socket; private PersistentPtySession? _session; + private PersistentPtyAttachment? _attachment; public static IEnumerable Fixtures() => Cases.Select(static entry => entry.Fixture); @@ -61,10 +62,8 @@ public class PersistentPtySessionReadBenchmarks : IAsyncDisposable public WireFixture Fixture { get; set; } = null!; /// - /// Builds the one session and canned socket this fixture's benchmark invocations replay - /// against; rewinds it before every read instead of - /// paying for a fresh socket and session per iteration, which would count construction noise - /// as part of the read path this class isolates. The session's attachment is seeded once from + /// Validates the fixture once. Each measured invocation owns a fresh connection because + /// terminal completion is permanent; construction and disposal are included. The attachment is seeded from /// a decoded attached frame rather than hand-built, so the fixture never drifts from /// what the decoder itself considers a valid grant. /// @@ -73,6 +72,7 @@ public async Task SetupAsync() { _messages = MessagesFor(Fixture); _socket = new CannedTerminalWebSocket(_messages); + _attachment = SeedAttachment(); // Ownership stays local until the session has taken the core over — the same // unconditional-dispose-of-a-transferable-local shape PersistentPtySession.AttachAsync @@ -81,7 +81,7 @@ public async Task SetupAsync() try { core = NewCore(_socket); - _session = new PersistentPtySession(core, SeedAttachment()); + _session = new PersistentPtySession(core, _attachment); core = null; } finally @@ -114,20 +114,20 @@ public async Task SetupAsync() throw new InvalidOperationException($"Fixture '{Fixture.Name}' did not decode into complete, non-empty output frames."); } - _socket.Reset(); - if (DecodeFrames() != Fixture.Items) { throw new InvalidOperationException($"Fixture '{Fixture.Name}' did not decode-alone into the expected frame count."); } } - /// The complete read path: canned receive, cross-receive reassembly, decode, and yield. + /// The connection lifetime: construction, receive, reassembly, decode, delivery, and disposal. [Benchmark] - public Task ReadFramesAsync() + public async Task ReadFramesAsync() { - _socket!.Reset(); - return CountFramesAsync(_session!); + using var socket = new CannedTerminalWebSocket(_messages); + await using var core = NewCore(socket); + await using var session = new PersistentPtySession(core, _attachment!); + return await CountFramesAsync(session).ConfigureAwait(false); } /// Decode alone, over the exact messages the complete rung assembles: no socket, no receive loop. diff --git a/tests/OpenCode.Sdk.Performance.Tests/Benchmarks/PtySessionReadBenchmarks.cs b/tests/OpenCode.Sdk.Performance.Tests/Benchmarks/PtySessionReadBenchmarks.cs index 4e839cd1..4fa4e0b1 100644 --- a/tests/OpenCode.Sdk.Performance.Tests/Benchmarks/PtySessionReadBenchmarks.cs +++ b/tests/OpenCode.Sdk.Performance.Tests/Benchmarks/PtySessionReadBenchmarks.cs @@ -45,10 +45,8 @@ public class PtySessionReadBenchmarks : IAsyncDisposable public WireFixture Fixture { get; set; } = null!; /// - /// Builds the one session and canned socket this fixture's benchmark invocations replay - /// against; rewinds it before every read instead of - /// paying for a fresh socket and session per iteration, which would count construction - /// noise as part of the read path this class isolates. + /// Validates the fixture once. Each measured invocation owns a fresh connection because + /// terminal completion is permanent; construction and disposal are included in the cost. /// [GlobalSetup] public async Task SetupAsync() @@ -75,20 +73,19 @@ public async Task SetupAsync() throw new InvalidOperationException($"Fixture '{Fixture.Name}' did not decode into complete, non-empty output frames."); } - _socket.Reset(); - if (DecodeFrames() != Fixture.Items) { throw new InvalidOperationException($"Fixture '{Fixture.Name}' did not decode-alone into the expected frame count."); } } - /// The complete read path: canned receive, cross-receive reassembly, decode, and yield. + /// The connection lifetime: construction, receive, reassembly, decode, delivery, and disposal. [Benchmark] - public Task ReadFramesAsync() + public async Task ReadFramesAsync() { - _socket!.Reset(); - return CountFramesAsync(_session!); + using var socket = new CannedTerminalWebSocket(_messages); + await using var session = new PtySession(socket); + return await CountFramesAsync(session).ConfigureAwait(false); } /// Decode alone, over the exact messages the complete rung assembles: no socket, no receive loop. diff --git a/tests/OpenCode.Sdk.Tests/PersistentPtys/PersistentPtyConnectOptionsTests.cs b/tests/OpenCode.Sdk.Tests/PersistentPtys/PersistentPtyConnectOptionsTests.cs index 36a595c6..aabe8a7c 100644 --- a/tests/OpenCode.Sdk.Tests/PersistentPtys/PersistentPtyConnectOptionsTests.cs +++ b/tests/OpenCode.Sdk.Tests/PersistentPtys/PersistentPtyConnectOptionsTests.cs @@ -2,6 +2,26 @@ namespace OpenCode.Sdk.Tests; public sealed class PersistentPtyConnectOptionsTests { + [Test] + public async Task SendTimeout_Should_Default_To_Thirty_Seconds_And_Accept_A_Custom_Budget() + { + await Assert.That(new PersistentPtyConnectOptions().SendTimeout).IsEqualTo(TimeSpan.FromSeconds(30)); + var options = new PersistentPtyConnectOptions { SendTimeout = TimeSpan.FromSeconds(2) }; + await Assert.That(options.SendTimeout).IsEqualTo(TimeSpan.FromSeconds(2)); + } + + [Test] + [Arguments(-10_000L)] + [Arguments(0L)] + [Arguments(9_999L)] + [Arguments(21_474_836_470_001L)] + public async Task SendTimeout_Should_Refuse_An_Unrepresentable_Finite_Budget(long ticks) + { + _ = Assert.Throws(() => + _ = new PersistentPtyConnectOptions { SendTimeout = TimeSpan.FromTicks(ticks) }); + await Task.CompletedTask; + } + /// The largest cursor the server's JS safe-integer guard accepts. private const long MaximumCursor = 9_007_199_254_740_991; diff --git a/tests/OpenCode.Sdk.Tests/PersistentPtys/PersistentPtyConnectUriBuilderTests.cs b/tests/OpenCode.Sdk.Tests/PersistentPtys/PersistentPtyConnectUriBuilderTests.cs index 8e5931c9..b01771e8 100644 --- a/tests/OpenCode.Sdk.Tests/PersistentPtys/PersistentPtyConnectUriBuilderTests.cs +++ b/tests/OpenCode.Sdk.Tests/PersistentPtys/PersistentPtyConnectUriBuilderTests.cs @@ -16,7 +16,7 @@ await Assert.That(uri.ToString()) } [Test] - public async Task Build_Should_Carry_Every_Connect_Option() + public async Task Build_Should_Carry_Wire_Options_And_Omit_The_Local_Send_Timeout() { var options = new PersistentPtyConnectOptions { @@ -24,6 +24,7 @@ public async Task Build_Should_Carry_Every_Connect_Option() Role = PersistentPtyRole.Observer, AttachmentId = "att_1", Takeover = true, + SendTimeout = TimeSpan.FromSeconds(2), }; var uri = PersistentPtyConnectUriBuilder.Build(Snapshot("http://localhost:4096"), PtyId, options); diff --git a/tests/OpenCode.Sdk.Tests/PersistentPtys/PersistentPtySessionTests.cs b/tests/OpenCode.Sdk.Tests/PersistentPtys/PersistentPtySessionTests.cs index 86f509b6..c295f55c 100644 --- a/tests/OpenCode.Sdk.Tests/PersistentPtys/PersistentPtySessionTests.cs +++ b/tests/OpenCode.Sdk.Tests/PersistentPtys/PersistentPtySessionTests.cs @@ -6,6 +6,71 @@ namespace OpenCode.Sdk.Tests; public sealed class PersistentPtySessionTests { + [Test] + public async Task AttachAsync_Should_Release_The_Receiver_When_Canceled_Before_Attachment() + { + var socket = new ScriptedTerminalWebSocket().Parking(); + 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); +#pragma warning restore CA2025 + OperationCanceledException? failure = null; + try + { + await socket.Parked; + } + finally + { + await cancellation.CancelAsync(); + try + { + var unexpected = await pending; + await unexpected.DisposeAsync(); + } + catch (OperationCanceledException exception) + { + failure = exception; + } + } + + await Assert.That(failure).IsNotNull(); + await Assert.That(socket.DisposeCalls).IsEqualTo(1); + } + + [Test] + public async Task AttachAsync_Should_Detach_The_Connect_Token_After_Returning_The_Session() + { + 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 cancellation.CancelAsync(); + await session.WriteAsync(PersistentPtyFrameData.Output("input")); + socket.ReleaseReceives(); + + var frames = await ReadAllAsync(session); + + await Assert.That(frames.Count).IsEqualTo(1); + await Assert.That(((PersistentPtyOutputFrame)frames[0]).Data.ToArray()) + .IsEquivalentTo(PersistentPtyFrameData.Output("after attach")); + await Assert.That(socket.DisposeCalls).IsEqualTo(0); + } + + [Test] + 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); + using var watchdog = new CancellationTokenSource(TimeSpan.FromSeconds(5)); + + var failure = await Assert.That(async () => await session.ResizeAsync(100, 30, watchdog.Token)) + .Throws(); + + await Assert.That(failure!.InnerException).IsTypeOf(); + } + private const string PtyId = "pty_persistent_7"; private const WebSocketCloseStatus TerminalUnavailable = (WebSocketCloseStatus)4404; @@ -371,19 +436,54 @@ public async Task ReadAsync_Should_Refuse_An_Abnormal_Close() } [Test] - public async Task ReadAsync_Should_Track_The_Viewport_From_A_Resized_Frame() + public async Task ReadAsync_Should_Preserve_The_Local_Viewport_When_The_Server_Reports_A_Resize() { var socket = new ScriptedTerminalWebSocket() .Text(PersistentPtyFrameData.AttachedJson) .Text(PersistentPtyFrameData.ResizedJson) - .Closing(WebSocketCloseStatus.NormalClosure); + .Parking(); await using var session = await PersistentPtySession.AttachAsync(socket, PtyId, CancellationToken.None); - _ = await ReadAllAsync(session); + await using var frames = session.ReadAsync().GetAsyncEnumerator(); + await Assert.That(await frames.MoveNextAsync()).IsTrue(); + var resized = (PersistentPtyResizedFrame)frames.Current; + await Assert.That(resized.Cols).IsEqualTo(120); + await Assert.That(resized.Rows).IsEqualTo(40); await session.WriteAsync(PersistentPtyFrameData.Output("ls\n")); await Assert.That(socket.SentMessages.Single()) - .IsEquivalentTo(PersistentPtyFrameData.Framed(1, 120, 40, PersistentPtyFrameData.Output("ls\n"))); + .IsEquivalentTo(PersistentPtyFrameData.Framed(1, 80, 24, PersistentPtyFrameData.Output("ls\n"))); + } + + [Test] + public async Task WriteAsync_Should_Use_The_Resize_That_Completes_Before_Its_Socket_Send() + { + var socket = new ScriptedTerminalWebSocket() + .Text(PersistentPtyFrameData.AttachedJson) + .Text(PersistentPtyFrameData.ResizedJson) + .Parking() + .GatingSends(); + await using var session = await PersistentPtySession.AttachAsync(socket, PtyId, CancellationToken.None); + var resize = session.ResizeAsync(100, 30); + await socket.SendEntered; + var write = session.WriteAsync(PersistentPtyFrameData.Output("ls\n")); + try + { + await using var frames = session.ReadAsync().GetAsyncEnumerator(); + await Assert.That(await frames.MoveNextAsync()).IsTrue(); + await Assert.That(((PersistentPtyResizedFrame)frames.Current).Cols).IsEqualTo(120); + } + finally + { + socket.ReleaseSends(); + } + + await Task.WhenAll(resize, write); + + await Assert.That(socket.SentMessages[0]).IsEquivalentTo(PersistentPtyFrameData.Framed(0, 100, 30, [])); + await Assert.That(socket.SentMessages[1]) + .IsEquivalentTo(PersistentPtyFrameData.Framed(1, 100, 30, PersistentPtyFrameData.Output("ls\n"))); + await Assert.That(socket.MaxConcurrentSends).IsEqualTo(1); } [Test] @@ -455,7 +555,7 @@ public async Task ResizeAsync_Should_Refuse_A_Zero_Or_Oversized_Dimension() } [Test] - public async Task ResizeAsync_Should_Leave_The_Viewport_Alone_When_The_Control_Frame_Never_Left() + public async Task ResizeAsync_Should_End_The_Connection_When_The_Control_Send_Fails() { var socket = new ScriptedTerminalWebSocket() .Text(PersistentPtyFrameData.AttachedJson) @@ -463,10 +563,25 @@ public async Task ResizeAsync_Should_Leave_The_Viewport_Alone_When_The_Control_F await using var session = await PersistentPtySession.AttachAsync(socket, PtyId, CancellationToken.None); _ = await Assert.That(async () => await session.ResizeAsync(100, 30)).Throws(); + _ = await Assert.That(async () => await session.WriteAsync(PersistentPtyFrameData.Output("ls\n"))) + .Throws(); + await Assert.That(socket.SentMessages).IsEmpty(); + await Assert.That(socket.DisposeCalls).IsEqualTo(1); + } + + [Test] + 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); + using var cancellation = new CancellationTokenSource(); + await cancellation.CancelAsync(); + + _ = await Assert.That(async () => await session.ResizeAsync(100, 30, cancellation.Token)) + .Throws(); await session.WriteAsync(PersistentPtyFrameData.Output("ls\n")); - // The write carries the attachment's own 80x24, not the size the failed resize asked for: - // a control frame that never left resized nothing on the server either. + // No control entered the socket, so the still-live attachment retains its original size. await Assert.That(socket.SentMessages.Single()) .IsEquivalentTo(PersistentPtyFrameData.Framed(1, 80, 24, PersistentPtyFrameData.Output("ls\n"))); } diff --git a/tests/OpenCode.Sdk.Tests/PtyConnectUriBuilderTests.cs b/tests/OpenCode.Sdk.Tests/PtyConnectUriBuilderTests.cs index 6f993cb1..9480b524 100644 --- a/tests/OpenCode.Sdk.Tests/PtyConnectUriBuilderTests.cs +++ b/tests/OpenCode.Sdk.Tests/PtyConnectUriBuilderTests.cs @@ -73,7 +73,7 @@ public async Task Build_Should_Carry_The_Live_Only_Cursor() public async Task Build_Should_Carry_A_Resume_Cursor_Beside_The_Location() { var connection = Snapshot("http://localhost:4096", new LocationSelector { Directory = "/repo" }); - var options = new PtyConnectOptions { Cursor = 4096 }; + var options = new PtyConnectOptions { Cursor = 4096, SendTimeout = TimeSpan.FromSeconds(2) }; var uri = PtyConnectUriBuilder.Build(connection, PtyId, options); diff --git a/tests/OpenCode.Sdk.Tests/Ptys/PtyConnectOptionsTests.cs b/tests/OpenCode.Sdk.Tests/Ptys/PtyConnectOptionsTests.cs index 9ee5387a..138f4355 100644 --- a/tests/OpenCode.Sdk.Tests/Ptys/PtyConnectOptionsTests.cs +++ b/tests/OpenCode.Sdk.Tests/Ptys/PtyConnectOptionsTests.cs @@ -2,6 +2,26 @@ namespace OpenCode.Sdk.Tests; public sealed class PtyConnectOptionsTests { + [Test] + public async Task SendTimeout_Should_Default_To_Thirty_Seconds_And_Accept_A_Custom_Budget() + { + await Assert.That(new PtyConnectOptions().SendTimeout).IsEqualTo(TimeSpan.FromSeconds(30)); + var options = new PtyConnectOptions { SendTimeout = TimeSpan.FromSeconds(2) }; + await Assert.That(options.SendTimeout).IsEqualTo(TimeSpan.FromSeconds(2)); + } + + [Test] + [Arguments(-10_000L)] + [Arguments(0L)] + [Arguments(9_999L)] + [Arguments(21_474_836_470_001L)] + public async Task SendTimeout_Should_Refuse_An_Unrepresentable_Finite_Budget(long ticks) + { + _ = Assert.Throws(() => + _ = new PtyConnectOptions { SendTimeout = TimeSpan.FromTicks(ticks) }); + await Task.CompletedTask; + } + /// The largest cursor the server's JS safe-integer guard accepts. private const long MaximumCursor = 9_007_199_254_740_991; diff --git a/tests/OpenCode.Sdk.Tests/Ptys/PtySessionLiveTests.cs b/tests/OpenCode.Sdk.Tests/Ptys/PtySessionLiveTests.cs index 37abc0a8..80d682c9 100644 --- a/tests/OpenCode.Sdk.Tests/Ptys/PtySessionLiveTests.cs +++ b/tests/OpenCode.Sdk.Tests/Ptys/PtySessionLiveTests.cs @@ -169,10 +169,57 @@ private async Task AssertInitialExecutionAsync( _scenario.Diagnostics.Enter("initial input acknowledgement", transcript); await session.WriteAsync("RUN " + nonce + "\r", cancellationToken); await transcript.ReadUntilRecordAsync(session, acknowledgement, cancellationToken); + await AssertReadCancellationReuseAsync(session, cancellationToken); await session.DisposeAsync(); return acknowledgement; } + private async Task AssertReadCancellationReuseAsync(PtySession session, CancellationToken cancellationToken) + { + _scenario.Diagnostics.Enter("cancel read"); + var canceled = false; + while (!canceled) + { + cancellationToken.ThrowIfCancellationRequested(); + using var stopReading = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken); + await using var reader = session.ReadAsync(stopReading.Token).GetAsyncEnumerator(stopReading.Token); + while (true) + { + var move = reader.MoveNextAsync(); + var pending = move.AsTask(); + if (pending.IsCompleted) + { + await Assert.That(await pending).IsTrue(); + continue; + } + + // Drain any trailing terminal output first. Cancellation is exercised only + // after a move actually suspends, never on an already-selected frame. + await stopReading.CancelAsync(); + try + { + await Assert.That(await pending).IsTrue(); + } + catch (OperationCanceledException) when (stopReading.IsCancellationRequested) + { + canceled = true; + } + + // Output selected between the incomplete-task observation and cancellation + // may win. Retry with a fresh token; only cancellation of this pending move + // supplies the proof, never the next move on an already-canceled token. + break; + } + } + + var nonce = Guid.NewGuid().ToString("N"); + var transcript = new PtyLiveTranscript(); + _scenario.Diagnostics.Enter("execute after read cancellation", transcript); + await session.WriteAsync("RUN " + nonce + "\r", cancellationToken); + await transcript.ReadUntilRecordAsync(session, "ACK:" + nonce, cancellationToken); + Console.WriteLine("pty-live: read-canceled same-connection-execution=" + nonce); + } + private async Task AssertReplayAndResumeAsync( PtyClient terminal, string acknowledgementA, diff --git a/tests/OpenCode.Sdk.Tests/Ptys/PtySessionTests.cs b/tests/OpenCode.Sdk.Tests/Ptys/PtySessionTests.cs index 06fa3e94..8aee42a9 100644 --- a/tests/OpenCode.Sdk.Tests/Ptys/PtySessionTests.cs +++ b/tests/OpenCode.Sdk.Tests/Ptys/PtySessionTests.cs @@ -6,6 +6,19 @@ namespace OpenCode.Sdk.Tests; public sealed class PtySessionTests { + [Test] + 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)); + using var watchdog = new CancellationTokenSource(TimeSpan.FromSeconds(5)); + + var failure = await Assert.That(async () => await session.WriteAsync("input", watchdog.Token)) + .Throws(); + + await Assert.That(failure!.InnerException).IsTypeOf(); + } + private const string SessionNotFoundReason = "session exited"; private const WebSocketCloseStatus SessionNotFound = (WebSocketCloseStatus)4404; @@ -188,19 +201,22 @@ public async Task ReadAsync_Should_Refuse_A_Second_Concurrent_Enumeration() } [Test] - public async Task ReadAsync_Should_Allow_A_Second_Enumeration_After_The_First_Ended() + public async Task ReadAsync_Should_Allow_A_Second_Enumeration_After_The_First_Breaks() { var socket = new ScriptedTerminalWebSocket() .Text("first") - .Closing(WebSocketCloseStatus.NormalClosure) .Text("second") .Closing(WebSocketCloseStatus.NormalClosure); await using var session = new PtySession(socket); - var first = await ReadAllAsync(session); + await using (var first = session.ReadAsync().GetAsyncEnumerator()) + { + await Assert.That(await first.MoveNextAsync()).IsTrue(); + await Assert.That(((PtyOutputFrame)first.Current).Text).IsEqualTo("first"); + } + var second = await ReadAllAsync(session); - await Assert.That(((PtyOutputFrame)first.Single()).Text).IsEqualTo("first"); await Assert.That(((PtyOutputFrame)second.Single()).Text).IsEqualTo("second"); } diff --git a/tests/OpenCode.Sdk.Tests/Snapshots/PublicApi.verified.txt b/tests/OpenCode.Sdk.Tests/Snapshots/PublicApi.verified.txt index 3c65bad3..103d44a0 100644 --- a/tests/OpenCode.Sdk.Tests/Snapshots/PublicApi.verified.txt +++ b/tests/OpenCode.Sdk.Tests/Snapshots/PublicApi.verified.txt @@ -992,6 +992,7 @@ public string? AttachmentId { get; init; } public long? Cursor { get; init; } public OpenCode.Sdk.PersistentPtyRole Role { get; init; } + public System.TimeSpan SendTimeout { get; init; } public bool Takeover { get; init; } } public sealed class PersistentPtyConnectTokenPostResponse : OpenCode.Sdk.OpenCodeResponse, System.IEquatable @@ -1248,6 +1249,7 @@ public PtyConnectOptions() { } public long? Cursor { get; init; } public OpenCode.Sdk.LocationSelector? Location { get; init; } + public System.TimeSpan SendTimeout { get; init; } } public sealed class PtyConnectTokenPostRequest : System.IEquatable { diff --git a/tests/OpenCode.Sdk.Tests/Support/OwnedEventReader.cs b/tests/OpenCode.Sdk.Tests/Support/OwnedEventReader.cs index c2f108de..9a7da15a 100644 --- a/tests/OpenCode.Sdk.Tests/Support/OwnedEventReader.cs +++ b/tests/OpenCode.Sdk.Tests/Support/OwnedEventReader.cs @@ -46,7 +46,8 @@ public void Own(Task reader) public async Task CompleteAsync(Exception? primaryFailure) { _teardownCancellation = !_window.IsCancellationRequested; - _cancellation = Task.Run(_window.CancelAsync, CancellationToken.None); + // Already on a worker: downlevel CancelAsync would queue another worker and spin waiting for it. + _cancellation = Task.Run(_window.Cancel, CancellationToken.None); _cleanup.Own("event cancellation", _cancellation); _cleanup.Own("event reader", _reader, IsTeardownCancellation); try diff --git a/tests/OpenCode.Sdk.Tests/Support/OwnedEventReaderTests.cs b/tests/OpenCode.Sdk.Tests/Support/OwnedEventReaderTests.cs index 80aa03b7..55e5b563 100644 --- a/tests/OpenCode.Sdk.Tests/Support/OwnedEventReaderTests.cs +++ b/tests/OpenCode.Sdk.Tests/Support/OwnedEventReaderTests.cs @@ -6,7 +6,7 @@ public sealed class OwnedEventReaderTests [Test] public async Task CompleteAsync_Should_Preserve_Primary_Reader_And_Later_Session_Failures() { - var owner = new OwnedEventReader(TimeSpan.FromMinutes(1), TimeSpan.FromSeconds(1), CancellationToken.None, new OperationDeadlineScenario().Deadline); + var owner = new OwnedEventReader(Timeout.InfiniteTimeSpan, TimeSpan.FromSeconds(1), CancellationToken.None, new OperationDeadlineScenario().Deadline); var primary = new InvalidOperationException("prompt failed"); var readerFailure = new InvalidOperationException("reader failed"); var removalFailure = new InvalidOperationException("removal failed"); @@ -25,7 +25,7 @@ public async Task CompleteAsync_Should_Preserve_Primary_Reader_And_Later_Session [Test] public async Task CompleteAsync_Should_Observe_Reader_After_Cancellation_Callback_Fails() { - var owner = new OwnedEventReader(TimeSpan.FromMinutes(1), TimeSpan.FromSeconds(1), CancellationToken.None, new OperationDeadlineScenario().Deadline); + var owner = new OwnedEventReader(Timeout.InfiniteTimeSpan, TimeSpan.FromSeconds(1), CancellationToken.None, new OperationDeadlineScenario().Deadline); var callbackFailure = new InvalidOperationException("cancellation callback failed"); var readerFailure = new InvalidOperationException("reader also failed"); var primary = new InvalidOperationException("prompt failed"); @@ -49,7 +49,7 @@ public async Task CompleteAsync_Should_Bound_Stalls_Reach_Removal_And_Attach_Lat var scenario = new OperationDeadlineScenario(); var cancellationDeadline = scenario.Hold("event cancellation"); var readerDeadline = scenario.Hold("event reader"); - var owner = new OwnedEventReader(TimeSpan.FromMinutes(1), TimeSpan.FromMilliseconds(50), CancellationToken.None, scenario.Deadline); + var owner = new OwnedEventReader(Timeout.InfiniteTimeSpan, TimeSpan.FromMilliseconds(50), CancellationToken.None, scenario.Deadline); var callbackFailure = new InvalidOperationException("late callback failed"); var readerFailure = new InvalidOperationException("late reader failed"); var primary = new InvalidOperationException("prompt failed"); @@ -110,7 +110,7 @@ public async Task CompleteAsync_Should_Bound_Stalls_Reach_Removal_And_Attach_Lat [Test] public async Task CompleteAsync_Should_Accept_Only_Its_Own_Cooperative_Reader_Cancellation() { - var owner = new OwnedEventReader(TimeSpan.FromMinutes(1), TimeSpan.FromSeconds(1), CancellationToken.None, new OperationDeadlineScenario().Deadline); + var owner = new OwnedEventReader(Timeout.InfiniteTimeSpan, TimeSpan.FromSeconds(1), CancellationToken.None, new OperationDeadlineScenario().Deadline); var cancellation = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); var token = owner.Token; using var registration = token.Register(() => _ = cancellation.TrySetCanceled(token)); @@ -137,7 +137,7 @@ public async Task CompleteAsync_Should_Accept_Only_Its_Own_Cooperative_Reader_Ca public async Task CompleteAsync_Should_Preserve_Caller_Cancellation_Identity() { using var caller = new CancellationTokenSource(); - var owner = new OwnedEventReader(TimeSpan.FromMinutes(1), TimeSpan.FromSeconds(1), caller.Token, new OperationDeadlineScenario().Deadline); + var owner = new OwnedEventReader(Timeout.InfiniteTimeSpan, TimeSpan.FromSeconds(1), caller.Token, new OperationDeadlineScenario().Deadline); await caller.CancelAsync(); var primary = new OperationCanceledException(caller.Token); owner.Own(Task.FromException(primary)); @@ -153,7 +153,7 @@ public async Task CompleteAsync_Should_Preserve_Caller_Cancellation_Identity() [Test] public async Task CompleteAsync_Should_Report_Unexpected_Reader_Cancellation() { - var owner = new OwnedEventReader(TimeSpan.FromMinutes(1), TimeSpan.FromSeconds(1), CancellationToken.None, new OperationDeadlineScenario().Deadline); + var owner = new OwnedEventReader(Timeout.InfiniteTimeSpan, TimeSpan.FromSeconds(1), CancellationToken.None, new OperationDeadlineScenario().Deadline); var unexpected = new OperationCanceledException("unrelated cancellation"); owner.Own(Task.FromException(unexpected)); var primary = new InvalidOperationException("prompt failed"); @@ -172,7 +172,7 @@ public async Task CompleteAsync_Should_Observe_Late_Cooperative_Cancellation_Aft var pending = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); var scenario = new OperationDeadlineScenario(); var deadline = scenario.Hold("event reader"); - var owner = new OwnedEventReader(TimeSpan.FromMinutes(1), TimeSpan.FromMilliseconds(50), CancellationToken.None, scenario.Deadline); + var owner = new OwnedEventReader(Timeout.InfiniteTimeSpan, TimeSpan.FromMilliseconds(50), CancellationToken.None, scenario.Deadline); owner.Own(pending.Task); var token = owner.Token; var completing = owner.CompleteAsync(null); diff --git a/tests/OpenCode.Sdk.Tests/Support/ScriptedTerminalReceive.cs b/tests/OpenCode.Sdk.Tests/Support/ScriptedTerminalReceive.cs index 90c22206..5a29d0e9 100644 --- a/tests/OpenCode.Sdk.Tests/Support/ScriptedTerminalReceive.cs +++ b/tests/OpenCode.Sdk.Tests/Support/ScriptedTerminalReceive.cs @@ -25,4 +25,7 @@ internal sealed record ScriptedTerminalReceive /// Gets whether the step parks until the socket is disposed or the read is canceled. public bool Parks { get; init; } + + /// Gets whether the step waits for the test to release the receive sequence. + public bool Pauses { get; init; } } diff --git a/tests/OpenCode.Sdk.Tests/Support/ScriptedTerminalWebSocket.cs b/tests/OpenCode.Sdk.Tests/Support/ScriptedTerminalWebSocket.cs index a6cdf376..90f60989 100644 --- a/tests/OpenCode.Sdk.Tests/Support/ScriptedTerminalWebSocket.cs +++ b/tests/OpenCode.Sdk.Tests/Support/ScriptedTerminalWebSocket.cs @@ -13,6 +13,8 @@ internal sealed class ScriptedTerminalWebSocket : ITerminalWebSocket { private readonly TaskCompletionSource _disposal = new(TaskCreationOptions.RunContinuationsAsynchronously); private readonly TaskCompletionSource _parked = new(TaskCreationOptions.RunContinuationsAsynchronously); + private readonly TaskCompletionSource _paused = new(TaskCreationOptions.RunContinuationsAsynchronously); + private readonly TaskCompletionSource _resume = new(TaskCreationOptions.RunContinuationsAsynchronously); private readonly Queue _receives = new(); private readonly List _sent = []; private readonly List _sentTypes = []; @@ -40,6 +42,9 @@ internal sealed class ScriptedTerminalWebSocket : ITerminalWebSocket /// Gets a task that completes once a scripted park step is reached. public Task Parked => _parked.Task; + /// Gets a signal that physical receive reached the scripted pause. + public Task Paused => _paused.Task; + /// Gets a task that completes once the first send reaches the socket. public Task SendEntered => _sendEntered.Task; @@ -113,6 +118,20 @@ public ScriptedTerminalWebSocket Parking() return this; } + /// Pauses physical receive until is called. + public ScriptedTerminalWebSocket Pausing() + { + _receives.Enqueue(new ScriptedTerminalReceive { Pauses = true }); + return this; + } + + /// Scripts a text fragment whose message continues in a later step. + public ScriptedTerminalWebSocket TextFragment(string text) => + Bytes(WebSocketMessageType.Text, Encoding.UTF8.GetBytes(text), endOfMessage: false); + + /// Releases the scripted receive pause. + public void ReleaseReceives() => _resume.TrySetResult(true); + /// Holds every send inside the socket until runs. public ScriptedTerminalWebSocket GatingSends() { @@ -149,10 +168,16 @@ public Task ReceiveAsync(ArraySegment buffer, Cancellati { if (_receives.Count is 0) { - throw new InvalidOperationException("The scripted terminal WebSocket ran out of receive steps."); + // An open network socket waits for the next message even with no public reader. + return ParkAsync(cancellationToken); } var step = _receives.Dequeue(); + if (step.Pauses) + { + return ResumeReceiveAsync(buffer, cancellationToken); + } + if (step.Parks) { return ParkAsync(cancellationToken); @@ -239,6 +264,7 @@ public void Dispose() DisposeCalls++; _ = _disposal.TrySetResult(true); ReleaseSends(); + ReleaseReceives(); } private ScriptedTerminalWebSocket Bytes(WebSocketMessageType messageType, byte[] payload, bool endOfMessage) @@ -264,4 +290,11 @@ private async Task ParkAsync(CancellationToken cancellationTok cancellationToken.ThrowIfCancellationRequested(); throw new ObjectDisposedException(nameof(ScriptedTerminalWebSocket)); } + + private async Task ResumeReceiveAsync(ArraySegment buffer, CancellationToken cancellationToken) + { + _ = _paused.TrySetResult(true); + _ = await _resume.Task.WaitAsync(cancellationToken); + return await ReceiveAsync(buffer, cancellationToken); + } } diff --git a/tests/OpenCode.Sdk.Tests/TerminalSocketCoreTests.cs b/tests/OpenCode.Sdk.Tests/TerminalSocketCoreTests.cs index 4e695467..df2b7fc8 100644 --- a/tests/OpenCode.Sdk.Tests/TerminalSocketCoreTests.cs +++ b/tests/OpenCode.Sdk.Tests/TerminalSocketCoreTests.cs @@ -1,5 +1,7 @@ using System.Net.WebSockets; +using NSubstitute; using OpenCode.Sdk.Internal; +using OpenCode.Sdk.Internal.Abstractions; using OpenCode.Sdk.Tests.Support; namespace OpenCode.Sdk.Tests; @@ -12,6 +14,290 @@ namespace OpenCode.Sdk.Tests; /// public sealed class TerminalSocketCoreTests { + [Test] + [Arguments(WebSocketCloseStatus.NormalClosure, "connection has ended")] + [Arguments(WebSocketCloseStatus.InternalServerError, "1011")] + public async Task SendAsync_Should_Wake_A_Queued_Call_With_The_Connections_Terminal_Failure( + WebSocketCloseStatus status, string expectedMessage) + { + using var socket = new ScriptedTerminalWebSocket().Pausing() + .Closing(status).GatingSends(); + await using var core = new TerminalSocketCore( + socket, PtyFrameDecoder.Instance, PtyClosePolicy.Instance, typeof(PtySession)); + 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); + socket.ReleaseReceives(); + OpenCodeTransportException? failure = null; + try + { + await queued; + } + catch (OpenCodeTransportException exception) + { + failure = exception; + } + finally + { + socket.ReleaseSends(); + await first; + } + + await Assert.That(failure).IsNotNull(); + await Assert.That(failure!.Message).Contains(expectedMessage); + await Assert.That(socket.SentMessages.Count).IsEqualTo(1); + } + + [Test] + public async Task SendAsync_Should_Preserve_The_First_Terminal_Outcome_After_A_Later_Send_Cancellation() + { + using var socket = new ScriptedTerminalWebSocket().Pausing() + .Closing(WebSocketCloseStatus.InternalServerError).GatingSends(); + await using var core = new TerminalSocketCore( + socket, PtyFrameDecoder.Instance, PtyClosePolicy.Instance, typeof(PtySession)); + using var cancellation = new CancellationTokenSource(); + var pending = core.SendAsync(new ArraySegment([0x61]), WebSocketMessageType.Text, cancellation.Token); + await socket.SendEntered; + socket.ReleaseReceives(); + await using var reader = core.ReadAsync(CancellationToken.None).GetAsyncEnumerator(); + var firstFailure = await Assert.That(async () => _ = await reader.MoveNextAsync()).Throws(); + await cancellation.CancelAsync(); + OperationCanceledException? canceled = null; + try + { + await pending; + } + catch (OperationCanceledException exception) + { + canceled = exception; + } + + await Assert.That(canceled).IsNotNull(); + var laterFailure = await Assert.That(async () => await core.SendAsync( + new ArraySegment([0x62]), WebSocketMessageType.Text, CancellationToken.None)).Throws(); + await Assert.That(laterFailure).IsSameReferenceAs(firstFailure); + } + + [Test] + public async Task SendAsync_Should_End_The_Attachment_After_Physical_Caller_Cancellation() + { + using var socket = new ScriptedTerminalWebSocket().GatingSends(); + await using var core = new TerminalSocketCore( + socket, PtyFrameDecoder.Instance, PtyClosePolicy.Instance, typeof(PtySession)); + using var cancellation = new CancellationTokenSource(); + var pending = core.SendAsync(new ArraySegment([0x61]), WebSocketMessageType.Text, cancellation.Token); + await socket.SendEntered; + await cancellation.CancelAsync(); + OperationCanceledException? failure = null; + try + { + await pending; + } + catch (OperationCanceledException exception) + { + failure = exception; + } + + await Assert.That(failure!.CancellationToken).IsEqualTo(cancellation.Token); + _ = await Assert.That(async () => await core.SendAsync( + new ArraySegment([0x62]), WebSocketMessageType.Text, CancellationToken.None)).Throws(); + await Assert.That(socket.SentMessages.Count).IsEqualTo(1); + await Assert.That(socket.DisposeCalls).IsEqualTo(1); + } + + [Test] + public async Task ReadAsync_Should_Preserve_A_Fragmented_Message_Across_Consumer_Cancellation() + { + 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)); + using var cancellation = new CancellationTokenSource(); + await using (var reader = core.ReadAsync(cancellation.Token).GetAsyncEnumerator()) + { + var pending = reader.MoveNextAsync(); + await socket.Paused; + await cancellation.CancelAsync(); + _ = await Assert.That(async () => _ = await pending).Throws(); + } + + await core.SendAsync(new ArraySegment([0x61]), WebSocketMessageType.Text, CancellationToken.None); + socket.ReleaseReceives(); + await using var resumed = core.ReadAsync(CancellationToken.None).GetAsyncEnumerator(); + await Assert.That(await resumed.MoveNextAsync()).IsTrue(); + await Assert.That(((PtyOutputFrame)resumed.Current).Text).IsEqualTo("first second"); + await Assert.That(await resumed.MoveNextAsync()).IsFalse(); + } + + [Test] + public async Task ReadAsync_Should_Leave_Queued_Data_And_The_Terminal_Error_When_Already_Canceled() + { + using var socket = new ScriptedTerminalWebSocket().Text("prefix").Closing(WebSocketCloseStatus.InternalServerError); + await using var core = new TerminalSocketCore( + socket, PtyFrameDecoder.Instance, PtyClosePolicy.Instance, typeof(PtySession)); + using var cancellation = new CancellationTokenSource(); + await cancellation.CancelAsync(); + await using (var reader = core.ReadAsync(cancellation.Token).GetAsyncEnumerator()) + { + _ = await Assert.That(async () => _ = await reader.MoveNextAsync()).Throws(); + } + + await using var resumed = core.ReadAsync(CancellationToken.None).GetAsyncEnumerator(); + await Assert.That(await resumed.MoveNextAsync()).IsTrue(); + await Assert.That(((PtyOutputFrame)resumed.Current).Text).IsEqualTo("prefix"); + _ = await Assert.That(async () => _ = await resumed.MoveNextAsync()).Throws(); + } + + [Test] + 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)); + await using (var first = core.ReadAsync(CancellationToken.None).GetAsyncEnumerator()) + { + await Assert.That(await first.MoveNextAsync()).IsTrue(); + await Assert.That(((PtyOutputFrame)first.Current).Text).IsEqualTo("prefix"); + _ = await Assert.That(async () => _ = await first.MoveNextAsync()).Throws(); + } + + await using var second = core.ReadAsync(CancellationToken.None).GetAsyncEnumerator(); + var failure = await Assert.That(async () => _ = await second.MoveNextAsync()).Throws(); + await Assert.That(failure!.Message).Contains("1011"); + } + + [Test] + public async Task ReadAsync_Should_Keep_Normal_Completion_After_The_Queued_Prefix() + { + using var socket = new ScriptedTerminalWebSocket().Text("prefix").Closing(WebSocketCloseStatus.NormalClosure); + await using var core = new TerminalSocketCore( + socket, PtyFrameDecoder.Instance, PtyClosePolicy.Instance, typeof(PtySession)); + await using (var first = core.ReadAsync(CancellationToken.None).GetAsyncEnumerator()) + { + await Assert.That(await first.MoveNextAsync()).IsTrue(); + await Assert.That(await first.MoveNextAsync()).IsFalse(); + } + + await using var second = core.ReadAsync(CancellationToken.None).GetAsyncEnumerator(); + await Assert.That(await second.MoveNextAsync()).IsFalse(); + } + + [Test] + 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)); + await using var first = core.ReadAsync(CancellationToken.None).GetAsyncEnumerator(); + await Assert.That(await first.MoveNextAsync()).IsTrue(); + + await core.DisposeAsync(); + + await using var afterDisposal = core.ReadAsync(CancellationToken.None).GetAsyncEnumerator(); + await Assert.That(await afterDisposal.MoveNextAsync()).IsFalse(); + await Assert.That(socket.DisposeCalls).IsEqualTo(1); + } + + [Test] + 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)); + var send = core.SendAsync(new ArraySegment([0x61]), WebSocketMessageType.Text, CancellationToken.None); + await socket.SendEntered; + var firstDisposal = core.DisposeAsync(); + var secondDisposal = core.DisposeAsync(); + var first = firstDisposal.AsTask(); + var second = secondDisposal.AsTask(); + try + { + await Assert.That(second.IsCompleted).IsFalse(); + } + finally + { + socket.ReleaseSends(); + } + + await Task.WhenAll(first, second, send); + await Assert.That(socket.DisposeCalls).IsEqualTo(1); + } + + [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); + using var socket = new ScriptedTerminalWebSocket().GatingSends(); + await using var core = new TerminalSocketCore( + socket, PtyFrameDecoder.Instance, PtyClosePolicy.Instance, typeof(PtySession)) + { + DeadlineFactory = timers, + }; + 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); + await queuedDeadline.CancelAsync(); + socket.ReleaseSends(); + await first; + + OpenCodeTransportException? failure = null; + try + { + await queued; + } + catch (OpenCodeTransportException exception) + { + failure = exception; + } + + await Assert.That(failure).IsNotNull(); + await Assert.That(failure!.InnerException).IsTypeOf(); + await Assert.That(socket.SentMessages.Count).IsEqualTo(1); + await Assert.That(socket.DisposeCalls).IsEqualTo(0); + await core.SendAsync(new ArraySegment([0x63]), WebSocketMessageType.Text, CancellationToken.None); + await Assert.That(socket.SentMessages.Count).IsEqualTo(2); + } + + [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); + using var socket = new ScriptedTerminalWebSocket().GatingSends(); + await using var core = new TerminalSocketCore( + socket, PtyFrameDecoder.Instance, PtyClosePolicy.Instance, typeof(PtySession)) + { + DeadlineFactory = timers, + }; + var send = core.SendAsync(new ArraySegment([0x61]), WebSocketMessageType.Text, CancellationToken.None); + await socket.SendEntered; + await deadline.CancelAsync(); + socket.ReleaseSends(); + + OpenCodeTransportException? failure = null; + try + { + await send; + } + catch (OpenCodeTransportException exception) + { + failure = exception; + } + + await Assert.That(failure).IsNotNull(); + await Assert.That(failure!.InnerException).IsTypeOf(); + _ = await Assert.That(async () => await core.SendAsync( + new ArraySegment([0x62]), WebSocketMessageType.Text, CancellationToken.None)) + .Throws(); + await Assert.That(socket.SentMessages.Count).IsEqualTo(1); + await Assert.That(socket.DisposeCalls).IsEqualTo(1); + } + [Test] public async Task Constructor_Should_Refuse_A_Null_Socket() { diff --git a/tests/Shared/Ownership/OwnedCleanup.cs b/tests/Shared/Ownership/OwnedCleanup.cs index 5f34ba5b..ca3697c8 100644 --- a/tests/Shared/Ownership/OwnedCleanup.cs +++ b/tests/Shared/Ownership/OwnedCleanup.cs @@ -68,7 +68,8 @@ private async Task CaptureFailureAsync( return; } - var cancellation = Task.Run(budget.CancelAsync, CancellationToken.None); + // The owned worker captures callback faults without a nested downlevel CancelAsync work item. + var cancellation = Task.Run(budget.Cancel, CancellationToken.None); await ObserveAsync(operation.Name + " cancellation", cancellation, failures); DisposeBudgetAfterCompletion(budget, pending, cancellation); budget = null;