Stop a flooding remote growing the process, and let an ended session go

The terminal's memory was bounded by an argument that is not true of the SSH
library underneath it, and the whole of CreditWindow's remark was that argument:
the pump stops reading when the renderer falls behind, SSH's own receive window
closes, the remote sshd blocks on write, and the backpressure reaches the process
producing the output. The middle step does not happen. Renci.SshNet's
Channels.Channel.OnData calls its window adjustment before it raises DataReceived
and re-opens the channel window as soon as it drops below one packet, and
ShellStream's own handler copies the bytes into a System.Net.ArrayBuffer without
blocking and without a ceiling — a buffer that doubles on demand and is not given
back until the stream is disposed. So a reader that stopped reading slowed the
remote down by nothing at all. It only chose whose buffer the flood landed in,
and it chose the one nothing bounds.

Measured against the container this repository already runs sshd in: a reader
that reads nothing while `yes` runs leaves ShellStream holding 8.9 MB after one
second, 26.8 MB after three and 44.3 MB after five, its capacity stepping 16 → 32
→ 64 MB. That is the report of "a massive memory leak on Windows"; it is not a
Windows fault, it is simply where long sessions with heavy output live.

The fix is to read continuously, which is what keeps SSH.NET's buffer empty, and
to put the bound where it can actually be enforced. OutputBacklog is that bound:
a byte-capped queue of unsent output, 4 MiB by default, which drops its oldest
bytes when full and counts what it discarded. Oldest rather than newest, because
the tail of a flood is the prompt somebody is waiting for.

The two bounds are now separate and each is enforced where it means something.
The backlog bounds what is waiting to be sent. The credit window goes on bounding
what has been sent and not yet drawn — the queue inside the WebView that nothing
on this side can measure — and is reserved against in the flush loop rather than
the read loop. Reading is part of neither, because reading was never what the
remote was waiting on.

Dropping is announced rather than silent. A terminal quietly showing a different
transcript from the one the remote sent is worse than one that says it could not
keep up, because the second can be believed. The line goes out under its own
opcode, TerminalServerOpcode.OutputDropped, rather than spliced into the output
stream: what this application says about the transcript is not part of the
transcript, and keeping them apart is also what makes the arithmetic hold — every
byte read is either in an Output payload or counted in BytesDropped. It opens
with an SGR reset, because the cut lands at an arbitrary offset and whatever
escape sequence the remote was in the middle of has been left half-said.

Frames are now capped at 64 KiB and taken straight into the frame buffer. The cap
is the large-object-heap threshold rather than a guess — a quarter-megabyte array
allocated sixty times a second is collected only by a generation 2 pass — and the
direct take removes an allocation and a copy of every byte the terminal shows.

The second half is the one that made the first half's cost persist. Nothing gave
a session back when the remote hung up: TerminalWorkspace kept the entry, and
with it an undisposed ISshConnection — an authenticated socket, its cipher state,
and a keep-alive putting a packet on the wire every thirty seconds — plus the
pump and SSH.NET's read buffer at whatever size the session's busiest moment had
grown it to, for as long as the tab stayed open. A tab somebody typed `exit` into
an hour ago was indistinguishable from a live one in everything except what it
drew. ReleaseAsync disposes both, from AnnounceEndedAsync and after the
announcement: the tab is told first, because tearing down an SSH connection in
front of that would be latency on the one thing the user is watching for. The
dictionary entry stays, because LiveSessionCount, IsSessionLive and the tab's own
ENDED and RECONNECT all read it. SendInputAsync therefore asks about liveness
rather than presence, since the shell channel is gone the moment the remote is.

Two faults introduced on the way here and fixed, both worth naming because
neither would have been visible from the outside:

The read loop left through a condition on its cancellation token, and RunAsync
reads a read loop that returns as the remote having closed the channel. A tab the
user closed would have reported itself as a shell that exited on its own. It
throws instead.

With the reader finished and a renderer that had stopped acknowledging, the flush
loop span: output to send, no credit to send it with, and a wait for more output
that returned at once because there was some. RunAsync never completed, so the tab
was never told and the connection was never released — a session invisible and
permanent. The loop now leaves for the drain as soon as the backlog is complete,
and RunAsync resets the credit window alongside completing it so a flusher already
parked on an acknowledgement that will never come is released.

Four tests asserted the old contract and now assert its reverse, which is the
point rather than a relaxation — WhenCreditRunsOut_ThePumpStopsReading is now
WhenCreditRunsOut_ThePumpKeepsReadingAndTheBacklogHoldsTheLine, and has to prove
both halves at once. The throughput harness gained a counting transport, because
a consumer that copies every frame into a 32 MB list was the slowest thing in the
pipeline and a harness whose consumer cannot keep up measures itself. The fake
shell gained a pace, so "a renderer that keeps up loses nothing" can be asked of a
producer a network could actually deliver rather than of one running at memory
bandwidth.

The test that would have caught this in the first place is
ARunawayRemote_KeepsBeingReadAndStaysInsideTheBacklog, and it needs the real
SSH.NET and the real sshd to say anything: a fake shell hands over bytes when it
is asked, and a real channel hands them over when they arrive. Under the old code
its BytesRead would have stopped at exactly WindowBytes.

121 terminal tests pass over three consecutive runs with no flakes, 105 against
the container, and the app, session, layout and remaining suites are unchanged and
green. Manual checks 1.10a and 1.10b cover the two symptoms a person can see: a
working set that plateaus under `yes`, and a connection that goes when the shell
does rather than when the tab does.

tests/DodoSSH.Client.Terminal.Tests/FakeShellSession.cs

@@ -9,6 +9,7 @@ internal sealed class FakeShellSession : ISshShellSession
     private readonly Lock gate = new();
 
     private readonly bool blockReads;
+    private readonly TimeSpan? pacePerRead;
 
     private long remaining;
     private byte pattern;
@@ -26,10 +27,26 @@ internal sealed class FakeShellSession : ISshShellSession
     /// read loop racing the test for control of the pump's credit window — see the reattach tests in
     /// <c>TerminalWorkspaceTests</c>.
     /// </param>
-    internal FakeShellSession(long bytesToProduce = 0, bool blockReads = false)
+    /// <param name="pacePerRead">
+    /// How long each read takes, or null for one that returns as fast as the CPU can fill a buffer.
+    /// </param>
+    /// <remarks>
+    /// <para>
+    /// The default producer emits at memory bandwidth, which is faster than any network and faster than
+    /// the pump can frame and send — useful for the throughput harness, and useless for asking whether a
+    /// terminal that <em>can</em> keep up loses anything, because nothing can keep up with it. A pace makes
+    /// the fake a link instead of a memcpy: 8 KiB every millisecond is 8 MB/s, which is a fast remote
+    /// rather than an impossible one.
+    /// </para>
+    /// </remarks>
+    internal FakeShellSession(
+        long bytesToProduce = 0,
+        bool blockReads = false,
+        TimeSpan? pacePerRead = null)
     {
         remaining = bytesToProduce;
         this.blockReads = blockReads;
+        this.pacePerRead = pacePerRead;
     }
 
     /// <inheritdoc />
@@ -81,6 +98,11 @@ internal sealed class FakeShellSession : ISshShellSession
             await Task.Delay(Timeout.InfiniteTimeSpan, cancellationToken).ConfigureAwait(false);
         }
 
+        if (pacePerRead is { } pace)
+        {
+            await Task.Delay(pace, cancellationToken).ConfigureAwait(false);
+        }
+
         await Task.Yield();
         cancellationToken.ThrowIfCancellationRequested();
 
@@ -212,6 +234,40 @@ internal sealed class FakeConnection(SshConnectionRequest request, long bytesPer
     }
 }
 
+/// <summary>
+/// Counts output without keeping it, and acknowledges it as a renderer that is keeping up would.
+/// </summary>
+/// <remarks>
+/// <see cref="RecordingTransport"/>'s cheap sibling, and the throughput harness needs it to be the one
+/// measuring rather than the one measured: recording copies every frame into a list that grows to the size
+/// of the whole run, which at the rates that harness drives makes the <em>test double</em> the slowest thing
+/// in the pipeline. A harness whose consumer cannot keep up is measuring itself.
+/// </remarks>
+internal sealed class CountingTransport : ITerminalTransport
+{
+    private long outputBytes;
+
+    /// <summary>Set to acknowledge every output frame immediately, as a keeping-up renderer would.</summary>
+    internal TerminalSessionPump? AutoAcknowledge { get; set; }
+
+    /// <summary>Payload bytes of every output frame sent so far.</summary>
+    internal long OutputBytes => Interlocked.Read(ref outputBytes);
+
+    /// <inheritdoc />
+    public ValueTask SendAsync(ReadOnlyMemory<byte> frame, CancellationToken cancellationToken)
+    {
+        if (TerminalFrame.TryRead(frame.Span, out var opcode, out _, out var payload)
+            && opcode == (byte)TerminalServerOpcode.Output)
+        {
+            _ = Interlocked.Add(ref outputBytes, payload.Length);
+
+            AutoAcknowledge?.Acknowledge((uint)payload.Length);
+        }
+
+        return ValueTask.CompletedTask;
+    }
+}
+
 /// <summary>Records frames, and can acknowledge them to keep credit flowing.</summary>
 internal sealed class RecordingTransport : ITerminalTransport
 {
@@ -270,6 +326,23 @@ internal sealed class RecordingTransport : ITerminalTransport
         return [.. output];
     }
 
+    /// <summary>Concatenated payloads of every frame of one opcode.</summary>
+    internal byte[] PayloadsOf(TerminalServerOpcode opcode)
+    {
+        var payloads = new List<byte>();
+
+        foreach (var frame in Frames)
+        {
+            if (TerminalFrame.TryRead(frame, out var actual, out _, out var payload)
+                && actual == (byte)opcode)
+            {
+                payloads.AddRange(payload);
+            }
+        }
+
+        return [.. payloads];
+    }
+
     /// <summary>Counts frames of one opcode.</summary>
     internal int CountOf(TerminalServerOpcode opcode) =>
         Frames.Count(frame =>

tests/DodoSSH.Client.Terminal.Tests/TerminalSessionPumpTests.cs

@@ -1,5 +1,6 @@
 using System.Diagnostics;
 using System.Globalization;
+using System.Text;
 using DodoSSH.Client.Ssh;
 
 namespace DodoSSH.Client.Terminal.Tests;
@@ -84,13 +85,22 @@ public sealed class TerminalSessionPumpTests
         await run;
     }
 
+    /// <summary>
+    /// A renderer that has stopped acknowledging does not stop the pump reading, and does not grow it.
+    /// </summary>
+    /// <remarks>
+    /// ◆ THIS TEST REPLACES ONE THAT ASSERTED THE OPPOSITE, and the reversal is the point rather than a
+    /// relaxation. The old contract was that the pump reads exactly one credit window and then stops,
+    /// which was believed to close SSH's receive window and block the remote. It does not — see
+    /// <see cref="OutputBacklog"/> — so all the old behaviour bought was an unbounded buffer inside
+    /// SSH.NET instead of a bounded one here. What has to be true now is both halves at once: reading
+    /// continues, so nothing piles up underneath, <em>and</em> what is held stays inside the backlog.
+    /// </remarks>
     [Fact]
-    public async Task WhenCreditRunsOut_ThePumpStopsReading()
+    public async Task WhenCreditRunsOut_ThePumpKeepsReadingAndTheBacklogHoldsTheLine()
     {
-        // The property the whole design rests on. With no acknowledgement the pump must read exactly
-        // the window and then stop, so nothing accumulates between the socket and the screen. Left
-        // unbounded, a remote running `yes` grows the client's memory until it dies.
         const int Window = 4 * 1024;
+        const int Backlog = 64 * 1024;
 
         await using var session = new FakeShellSession(bytesToProduce: long.MaxValue);
         var transport = new RecordingTransport();
@@ -104,19 +114,32 @@ public sealed class TerminalSessionPumpTests
                 ReadBufferBytes = 1024,
                 FlushInterval = TimeSpan.FromMilliseconds(5),
                 WindowBytes = Window,
+                BacklogBytes = Backlog,
             });
 
         var run = pump.RunAsync(TestContext.Current.CancellationToken);
 
         await Task.Delay(TimeSpan.FromMilliseconds(300), TestContext.Current.CancellationToken);
 
-        pump.BytesRead.ShouldBe(
-            Window,
-            "The pump read past its credit window, so output accumulates without bound.");
+        var readByNow = pump.BytesRead;
 
-        // Still stalled a moment later, rather than merely slow.
+        readByNow.ShouldBeGreaterThan(
+            Window,
+            "The pump stopped reading, which leaves the flood accumulating inside SSH.NET instead.");
+
+        pump.BytesPending.ShouldBeLessThanOrEqualTo(
+            Backlog,
+            "The backlog grew past its capacity, which is the unbounded growth it exists to prevent.");
+
+        pump.BytesDropped.ShouldBeGreaterThan(
+            0,
+            "Nothing was dropped, so something is still holding what the renderer never took.");
+
+        // Still reading a moment later, rather than having quietly stalled once the backlog filled.
         await Task.Delay(TimeSpan.FromMilliseconds(200), TestContext.Current.CancellationToken);
-        pump.BytesRead.ShouldBe(Window);
+
+        pump.BytesRead.ShouldBeGreaterThan(readByNow);
+        pump.BytesPending.ShouldBeLessThanOrEqualTo(Backlog);
 
         pump.Credits.Available.ShouldBe(0);
 
@@ -125,10 +148,13 @@ public sealed class TerminalSessionPumpTests
     }
 
     [Fact]
-    public async Task AcknowledgingCredit_LetsThePumpResume()
+    public async Task AcknowledgingCredit_LetsMoreOutputThrough()
     {
         const int Window = 4 * 1024;
 
+        // Endless, so the backlog never completes while this is being measured: once it does, the tail
+        // goes out regardless of credit and there is nothing for the window to hold back. Drops along the
+        // way change which bytes are sent, never how many — and the notice they raise is its own opcode.
         await using var session = new FakeShellSession(bytesToProduce: long.MaxValue);
         var transport = new RecordingTransport();
 
@@ -145,13 +171,16 @@ public sealed class TerminalSessionPumpTests
         var run = pump.RunAsync(TestContext.Current.CancellationToken);
 
         await Task.Delay(TimeSpan.FromMilliseconds(200), TestContext.Current.CancellationToken);
-        pump.BytesRead.ShouldBe(Window);
+
+        // One window's worth has gone out and stopped there, because nothing has been rendered.
+        var sentBeforeAcknowledgement = transport.OutputBytes().Length;
+        sentBeforeAcknowledgement.ShouldBe(Window);
 
         pump.Acknowledge(2048);
 
         await Task.Delay(TimeSpan.FromMilliseconds(200), TestContext.Current.CancellationToken);
 
-        pump.BytesRead.ShouldBe(Window + 2048);
+        transport.OutputBytes().Length.ShouldBe(sentBeforeAcknowledgement + 2048);
 
         await pump.DisposeAsync();
         await run;
@@ -161,9 +190,10 @@ public sealed class TerminalSessionPumpTests
     public async Task AnOverLargeAcknowledgement_DoesNotWidenTheWindow()
     {
         // The acknowledgement comes from JavaScript. A page claiming to have rendered a gigabyte must
-        // not be able to talk the host into an unbounded read.
+        // not be able to talk the host into filling its own buffer without limit.
         const int Window = 4 * 1024;
 
+        // Endless for the reason the test above gives: a backlog that completes stops the window mattering.
         await using var session = new FakeShellSession(bytesToProduce: long.MaxValue);
         var transport = new RecordingTransport();
 
@@ -185,7 +215,8 @@ public sealed class TerminalSessionPumpTests
         await Task.Delay(TimeSpan.FromMilliseconds(200), TestContext.Current.CancellationToken);
 
         // One further window's worth at most, never everything the endless producer could offer.
-        pump.BytesRead.ShouldBeLessThanOrEqualTo(Window * 2);
+        transport.OutputBytes().Length.ShouldBeLessThanOrEqualTo(Window * 2);
+        pump.Credits.Outstanding.ShouldBeLessThanOrEqualTo(Window);
 
         await pump.DisposeAsync();
         await run;
@@ -344,7 +375,11 @@ public sealed class TerminalSessionPumpTests
         const int Window = CreditWindow.DefaultWindowBytes;
 
         await using var session = new FakeShellSession(bytesToProduce: Target);
-        var transport = new RecordingTransport();
+
+        // Counting rather than recording, for the reason CountingTransport carries: a consumer that
+        // copies every frame into a 32 MB list is the slowest thing in this pipeline, and a harness
+        // whose consumer cannot keep up measures the harness.
+        var transport = new CountingTransport();
 
         await using var pump = CreatePump(
             session,
@@ -369,7 +404,15 @@ public sealed class TerminalSessionPumpTests
         var peakOutstanding = await sampler;
 
         pump.BytesRead.ShouldBe(Target);
-        transport.OutputBytes().Length.ShouldBe((int)Target);
+
+        // ◆ Every byte read either reached the renderer or was counted as dropped, and none of it went
+        // anywhere else. That is the invariant worth asserting here rather than "nothing was dropped":
+        // this producer emits at memory bandwidth with no network under it, which is faster than any
+        // real remote and faster than the pump can frame and send, so drops are the honest outcome and
+        // the harness would be lying to forbid them. What must never happen is a byte that is neither
+        // delivered nor accounted for — see NothingIsDropped_WhenTheRendererKeepsUpWithTheRemote for
+        // the case where the producer is one a network could actually deliver.
+        (transport.OutputBytes + pump.BytesDropped).ShouldBe(Target);
 
         var megabytesPerSecond = Target / 1024d / 1024d / stopwatch.Elapsed.TotalSeconds;
 
@@ -387,6 +430,142 @@ public sealed class TerminalSessionPumpTests
                 $"Peak in-flight was {peakOutstanding} bytes against a {Window}-byte window."));
     }
 
+    /// <summary>
+    /// A renderer that keeps up loses nothing, however much the remote sends.
+    /// </summary>
+    /// <remarks>
+    /// The other half of the trade the backlog makes, and the half a user actually lives in: dropping is
+    /// for a terminal that cannot keep up, and a terminal that can must show every byte. The producer here
+    /// is sized for a real link rather than for memory bandwidth — several times the backlog, delivered in
+    /// pieces a network could carry — so a drop would mean the pump losing output it had room for.
+    /// </remarks>
+    [Fact]
+    public async Task NothingIsDropped_WhenTheRendererKeepsUpWithTheRemote()
+    {
+        const int Backlog = 256 * 1024;
+        const long Target = 512L * 1024;
+
+        // 8 KiB a millisecond — 8 MB/s, a fast remote rather than an impossible one, and thirty-two
+        // times the backlog's capacity per second so the pump has no room to be merely lucky.
+        await using var session = new FakeShellSession(
+            bytesToProduce: Target,
+            pacePerRead: TimeSpan.FromMilliseconds(1));
+
+        var transport = new CountingTransport();
+
+        await using var pump = CreatePump(
+            session,
+            transport,
+            new TerminalPumpOptions
+            {
+                ReadBufferBytes = 8 * 1024,
+                FlushInterval = TimeSpan.FromMilliseconds(1),
+                WindowBytes = CreditWindow.DefaultWindowBytes,
+                BacklogBytes = Backlog,
+            });
+
+        transport.AutoAcknowledge = pump;
+
+        await pump.RunAsync(TestContext.Current.CancellationToken);
+
+        pump.BytesDropped.ShouldBe(0);
+        transport.OutputBytes.ShouldBe(Target);
+    }
+
+    /// <summary>
+    /// A shell that ends against a renderer which stopped acknowledging still finishes.
+    /// </summary>
+    /// <remarks>
+    /// ◆ THE HANG THIS DESIGN MAKES POSSIBLE, AND THE TEST THAT SAYS IT DOES NOT HAPPEN. The flush loop
+    /// waits on credit when the renderer is behind, and completing the backlog is a different signal from
+    /// returning credit — so a run whose reader finished while the flusher was parked on an
+    /// acknowledgement that will never come would wait for ever. Nothing above would notice: the shell has
+    /// ended, but the tab is told by <c>SessionEnded</c>, which fires off the run task completing, and the
+    /// connection is given back at the same moment. A session in that state is invisible and permanent.
+    /// </remarks>
+    [Fact]
+    public async Task WhenTheRemoteCloses_TheRunEndsEvenIfTheRendererNeverAcknowledged()
+    {
+        const int Window = 4 * 1024;
+
+        // More than the window, so the flusher has certainly run out of credit and parked by the time the
+        // producer reaches the end of what it has to give.
+        await using var session = new FakeShellSession(bytesToProduce: Window * 4);
+
+        // No AutoAcknowledge, ever: this renderer's process died holding the socket.
+        var transport = new RecordingTransport();
+
+        await using var pump = CreatePump(
+            session,
+            transport,
+            new TerminalPumpOptions
+            {
+                ReadBufferBytes = 1024,
+                FlushInterval = TimeSpan.FromMilliseconds(5),
+                WindowBytes = Window,
+            });
+
+        var run = pump.RunAsync(TestContext.Current.CancellationToken);
+
+        await Should.NotThrowAsync(() => run.WaitAsync(TimeSpan.FromSeconds(10), TestContext.Current.CancellationToken));
+
+        pump.Outcome.End.ShouldBe(SshSessionEnd.RemoteClosed);
+
+        // And the tail went out rather than being abandoned with the flusher: the last thing a remote
+        // writes is usually why it ended.
+        transport.OutputBytes().Length.ShouldBe(Window * 4);
+        transport.CountOf(TerminalServerOpcode.SessionClosed).ShouldBe(1);
+    }
+
+    /// <summary>
+    /// A drop is announced in the pane rather than leaving a silent gap.
+    /// </summary>
+    /// <remarks>
+    /// The thing that makes dropping defensible at all. A terminal quietly showing a different transcript
+    /// from the one the remote sent is worse than one that says it could not keep up, because the second
+    /// can be believed. Its own opcode, so the sentence is never counted as something the remote said —
+    /// see <see cref="TerminalServerOpcode.OutputDropped"/>.
+    /// </remarks>
+    [Fact]
+    public async Task WhenOutputIsDropped_TheUserIsToldInThePane()
+    {
+        const int Window = 4 * 1024;
+        const int Backlog = 16 * 1024;
+
+        await using var session = new FakeShellSession(bytesToProduce: long.MaxValue);
+        var transport = new RecordingTransport();
+
+        await using var pump = CreatePump(
+            session,
+            transport,
+            new TerminalPumpOptions
+            {
+                ReadBufferBytes = 1024,
+                FlushInterval = TimeSpan.FromMilliseconds(5),
+                WindowBytes = Window,
+                BacklogBytes = Backlog,
+            });
+
+        var run = pump.RunAsync(TestContext.Current.CancellationToken);
+
+        await Task.Delay(TimeSpan.FromMilliseconds(200), TestContext.Current.CancellationToken);
+
+        // Enough credit for one more frame, which is what carries the notice out.
+        pump.Acknowledge((uint)Window);
+
+        await Task.Delay(TimeSpan.FromMilliseconds(200), TestContext.Current.CancellationToken);
+
+        transport.CountOf(TerminalServerOpcode.OutputDropped).ShouldBeGreaterThan(0);
+
+        var notice = Encoding.UTF8.GetString(transport.PayloadsOf(TerminalServerOpcode.OutputDropped));
+
+        notice.ShouldContain("of output was dropped");
+        notice.ShouldStartWith("\u001b[0m");
+
+        await pump.DisposeAsync();
+        await run;
+    }
+
     /// <summary>Watches in-flight bytes for the duration of a run, returning the highest seen.</summary>
     private static async Task<int> SamplePeakOutstandingAsync(
         TerminalSessionPump pump,

tests/DodoSSH.Client.Terminal.Tests/TerminalWorkspaceTests.cs

@@ -83,9 +83,9 @@ public sealed class TerminalWorkspaceTests
     }
 
     /// <remarks>
-    /// The case a naive <c>sessions.Count</c> gets wrong. Nothing removes the entry when the remote
-    /// closes the channel by itself — the session is still in the dictionary, holding a connection that
-    /// is finished — so counting entries would report a shell that exited as still running. A user
+    /// The case a naive <c>sessions.Count</c> gets wrong. Nothing removes the entry when the remote closes
+    /// the channel by itself — what it held is given back, but the entry stays so the tab can go on saying
+    /// the session ended — so counting entries would report a shell that exited as still running. A user
     /// deciding whether it is safe to walk away is the person that lie is told to.
     /// </remarks>
     [Fact]
@@ -645,6 +645,86 @@ public sealed class TerminalWorkspaceTests
             await workspace.SendInputAsync(sessionId, "x"u8.ToArray(), TestContext.Current.CancellationToken));
     }
 
+    /// <summary>
+    /// A shell that ended on its own gives its connection back at once, without waiting for the tab.
+    /// </summary>
+    /// <remarks>
+    /// ◆ THE HALF THAT WAS MISSING. Closing a tab disposed the connection and always had; a remote hanging
+    /// up did not, so a session that ended went on holding an authenticated socket, a keep-alive putting a
+    /// packet on the wire every thirty seconds, and SSH.NET's read buffer at whatever size the session's
+    /// busiest moment had grown it to — for as long as the user left the tab open, which for a tab showing
+    /// the last thing a job said is indefinitely.
+    /// </remarks>
+    [Fact]
+    public async Task ASessionWhoseRemoteHasExited_GivesBackItsConnectionWithoutWaitingForTheTab()
+    {
+        // A shell with no output to give: its first read returns 0, which is the remote closing the
+        // channel, so the pump finishes with nobody having asked it to.
+        var connections = new FakeConnectionFactory(bytesPerShell: 0);
+
+        await using var workspace = CreateWorkspace(connections);
+
+        var sessionId = await workspace.OpenSessionAsync(
+            Request(), TerminalSize.Default, progress: null, TestContext.Current.CancellationToken);
+
+        await WaitUntilAsync(() => connections.Connections.TrueForAll(connection => connection.IsDisposed));
+
+        var connection = connections.Connections.ShouldHaveSingleItem();
+
+        connection.IsConnected.ShouldBeFalse();
+        connection.Shell.ShouldNotBeNull().IsOpen.ShouldBeFalse();
+
+        // The entry itself stays, and has to: the tab is still on screen, still says it ended, and still
+        // offers a reconnect — all of which read this workspace rather than the tab strip's own memory.
+        workspace.CreditsFor(sessionId).ShouldNotBeNull();
+        workspace.IsSessionLive(sessionId).ShouldBeFalse();
+    }
+
+    /// <remarks>
+    /// The consequence of the release above, and the reason <c>SendInputAsync</c> asks about liveness
+    /// rather than presence: the shell channel is disposed the moment the remote hangs up, so an accessory
+    /// key pressed at a tab that ended reaches a disposed stream. It has to be as ordinary as pressing one
+    /// at a tab that has closed, which is to say nothing happens.
+    /// </remarks>
+    [Fact]
+    public async Task SendingInputToASessionWhoseRemoteHasExited_IsIgnored()
+    {
+        var connections = new FakeConnectionFactory(bytesPerShell: 0);
+
+        await using var workspace = CreateWorkspace(connections);
+
+        var sessionId = await workspace.OpenSessionAsync(
+            Request(), TerminalSize.Default, progress: null, TestContext.Current.CancellationToken);
+
+        await WaitUntilAsync(() => !workspace.IsSessionLive(sessionId));
+
+        await Should.NotThrowAsync(async () =>
+            await workspace.SendInputAsync(sessionId, "x"u8.ToArray(), TestContext.Current.CancellationToken));
+    }
+
+    /// <remarks>
+    /// Closing a tab over a session that already released itself must still work, because that is the
+    /// ordinary way a dead tab goes away. Everything the close does is idempotent by construction — the
+    /// entry is removed under the lock, and both disposals are guarded — and this is the test that says so
+    /// rather than leaving it to be inferred from three separate classes.
+    /// </remarks>
+    [Fact]
+    public async Task ClosingATabOverASessionThatAlreadyEnded_IsStillSafe()
+    {
+        var connections = new FakeConnectionFactory(bytesPerShell: 0);
+
+        await using var workspace = CreateWorkspace(connections);
+
+        var sessionId = await workspace.OpenSessionAsync(
+            Request(), TerminalSize.Default, progress: null, TestContext.Current.CancellationToken);
+
+        await WaitUntilAsync(() => !workspace.IsSessionLive(sessionId));
+
+        await Should.NotThrowAsync(() => workspace.CloseSessionAsync(sessionId));
+
+        workspace.CreditsFor(sessionId).ShouldBeNull();
+    }
+
     /// <param name="connections">
     /// How connections are made. The renderer-gate tests never reach it — they stop at the gate — but they
     /// take a fake anyway, because reaching a real host from here would make this a network test.