diff --git a/docs/Sessions.md b/docs/Sessions.md
index b945fd0..0ac66a0 100644
--- a/docs/Sessions.md
+++ b/docs/Sessions.md
@@ -247,6 +247,12 @@ Mechanically: when the last external subscriber detaches and `DetachGraceSeconds
`DetachGraceSeconds` controls retention and expiry only; the reconnect/replay path that re-attaches a dropped client to a retained session is described in [Reconnect and replay](#reconnect-and-replay).
+#### Worker exit outside a gateway close
+
+`IWorkerClient` has no state-change callback, so a worker that reached `Closed` or `Faulted` *without the gateway closing it* — the client sent the `ShutdownWorker` control command and the worker exited cleanly, the worker crashed, its pipe broke, a heartbeat timed out — used to leave the session reporting `Ready` until its lease expired (a live finding: `WorkerClient` was `Closed` on the shutdown ack, the session still `Ready`, even with an event stream attached). The session now observes it from its own worker-event pump source (`MapWorkerEventsAsync`): the single distributor pump is the session's one always-on view of the worker (it runs from `MarkReady` via the dashboard mirror), and `WorkerClient` enters its terminal state *before* it completes the event channel, so when the enumeration ends the terminal state is visible. `ObserveWorkerTerminalState` then faults a `Ready` session whose worker is `Closed`/`Faulted`.
+
+The choice is **`Faulted`, not `Closed`**: `Closed` means the gateway closed the session, and a `Closed` session left in the registry would linger until its lease expired with nothing reaping it. A faulted session is permanently unusable (every command fails the readiness check, as the worker is gone), stays observable via `GetSessionStatus` with `FinalFault` naming the worker state, and is reaped by the faulted-session sweep below after `FaultedGraceSeconds` — which is where `SessionClosed` is accounted, exactly once, through the ordinary `CloseSessionCoreAsync` path (`WorkerClient.ShutdownAsync` on a `Closed` client returns immediately, so no kill fallback or warning). Only a `Ready` session follows its worker: during a gateway-initiated close the worker closing is the expected outcome, not a fault.
+
#### Faulted-session reaping
A session that faults (for example, a slow single-subscriber client that overflows its event queue under the `FailFast` backpressure policy) is left permanently unusable — every command fails `EvaluateReadyUnderLock` — but its worker keeps running and it still pins one of `MxGateway:Sessions:MaxSessions` slots. Rather than waiting up to `DefaultLeaseSeconds` for the lease to expire, `SessionManager.CloseExpiredLeasesAsync` also checks `IsFaultedReapable(now)` and reaps a faulted session through the same `TryBeginCloseIfExpired` → `CloseSessionCoreAsync` teardown, with the distinct `FaultedReason` (`"faulted-reaped"`). `MarkFaulted` stamps a fault timestamp from the session's `TimeProvider`; `MxGateway:Sessions:FaultedGraceSeconds` (default `0`) bounds how long the faulted session is retained before the next sweep reaps it — `0` reaps it on the next sweep, a positive value keeps it observable via `GetSessionStatus` for that window first. Sweep precedence when several conditions fire at once is lease-expiry, then faulted, then detach-grace, then unattached-grace.
diff --git a/src/ZB.MOM.WW.MxGateway.Server/Sessions/GatewaySession.cs b/src/ZB.MOM.WW.MxGateway.Server/Sessions/GatewaySession.cs
index 8cdb36f..5d57f0c 100644
--- a/src/ZB.MOM.WW.MxGateway.Server/Sessions/GatewaySession.cs
+++ b/src/ZB.MOM.WW.MxGateway.Server/Sessions/GatewaySession.cs
@@ -1004,12 +1004,54 @@ public sealed class GatewaySession
IWorkerClient workerClient = await GetReadyWorkerClientAsync(cancellationToken).ConfigureAwait(false);
TouchClientActivity(_eventStreaming.TimeProvider.GetUtcNow());
- await foreach (WorkerEvent workerEvent in workerClient
- .ReadEventsAsync(cancellationToken)
- .WithCancellation(cancellationToken)
- .ConfigureAwait(false))
+ try
{
- yield return mapper.MapEvent(workerEvent);
+ await foreach (WorkerEvent workerEvent in workerClient
+ .ReadEventsAsync(cancellationToken)
+ .WithCancellation(cancellationToken)
+ .ConfigureAwait(false))
+ {
+ yield return mapper.MapEvent(workerEvent);
+ }
+ }
+ finally
+ {
+ // The worker event stream only ends when the worker client has reached a terminal
+ // state (it enters Closed/Faulted BEFORE completing the channel) or the pump is being
+ // torn down. This is the session's one always-on view of the worker, so it is where
+ // a worker that went away on its own is noticed.
+ ObserveWorkerTerminalState();
+ }
+ }
+
+ // IWorkerClient has no state-change callback, so a worker that reached Closed/Faulted
+ // without the gateway closing it — the client sent ShutdownWorker and the worker exited,
+ // the worker crashed, its pipe broke, a heartbeat timed out — used to leave the session
+ // reporting Ready until its lease expired (live finding). The session follows the worker
+ // into Faulted: it is permanently unusable (every command fails the readiness check), stays
+ // observable via GetSessionStatus with the reason, and the faulted-session sweep reaps it
+ // after FaultedGraceSeconds, accounting SessionClosed exactly once there. Not Closed: Closed
+ // means the gateway closed it, and a Closed session still in the registry would linger until
+ // its lease expired with nothing reaping it. Only a Ready session follows — during a
+ // gateway-initiated close the worker closing is the expected outcome, not a fault.
+ private void ObserveWorkerTerminalState()
+ {
+ lock (_syncRoot)
+ {
+ if (_state is not SessionState.Ready || _workerClient is null)
+ {
+ return;
+ }
+
+ WorkerClientState workerState = _workerClient.State;
+ if (workerState is not (WorkerClientState.Closed or WorkerClientState.Faulted))
+ {
+ return;
+ }
+
+ MarkFaulted(
+ $"Worker for session {SessionId} reached {workerState} outside a gateway-initiated close "
+ + "(worker exited or faulted).");
}
}
diff --git a/src/ZB.MOM.WW.MxGateway.Tests/Gateway/Sessions/GatewaySessionWorkerExitTests.cs b/src/ZB.MOM.WW.MxGateway.Tests/Gateway/Sessions/GatewaySessionWorkerExitTests.cs
new file mode 100644
index 0000000..2a17c8b
--- /dev/null
+++ b/src/ZB.MOM.WW.MxGateway.Tests/Gateway/Sessions/GatewaySessionWorkerExitTests.cs
@@ -0,0 +1,151 @@
+using Microsoft.Extensions.Options;
+using ZB.MOM.WW.MxGateway.Contracts.Proto;
+using ZB.MOM.WW.MxGateway.Server.Configuration;
+using ZB.MOM.WW.MxGateway.Server.Metrics;
+using ZB.MOM.WW.MxGateway.Server.Sessions;
+using ZB.MOM.WW.MxGateway.Server.Workers;
+using ZB.MOM.WW.MxGateway.Tests.Gateway.Workers.Fakes;
+
+namespace ZB.MOM.WW.MxGateway.Tests.Gateway.Sessions;
+
+///
+/// A worker that goes away on its own — the client sent the ShutdownWorker control
+/// command and the worker exited cleanly, or the worker crashed / its pipe broke — must be
+/// noticed by its . The real reaches
+/// Closed / Faulted, but nothing used to tell the session, which kept reporting
+/// Ready (live windev finding) until its lease expired.
+///
+public sealed class GatewaySessionWorkerExitTests
+{
+ private static readonly TimeSpan TestTimeout = TimeSpan.FromSeconds(5);
+
+ ///
+ /// After the client's ShutdownWorker command the fake worker replies OK, sends the
+ /// shutdown ack and the worker client closes. The session must fault (the worker is gone
+ /// outside a gateway-initiated close), stay observable with a fault reason, and then be
+ /// reaped by the ordinary faulted-session sweep — which is where the session is accounted
+ /// closed, exactly once.
+ ///
+ /// A task that represents the asynchronous operation.
+ [Fact]
+ public async Task ClientShutdownWorker_WorkerExits_SessionFaultsAndSweepReapsItOnce()
+ {
+ await using FakeWorkerHarness fakeWorker = await FakeWorkerHarness.CreateConnectedPairAsync();
+ using GatewayMetrics metrics = new();
+ WorkerClient client = fakeWorker.CreateClient(metrics: metrics);
+ await StartClientAsync(fakeWorker, client);
+ SessionManager manager = CreateManager(client, metrics);
+ GatewaySession session = await OpenSessionAsync(manager);
+ using IEventSubscriberLease stream = session.AttachEventSubscriber(maxSubscribers: 1);
+
+ Task invokeTask = manager.InvokeAsync(session.SessionId, CreateShutdownWorkerCommand(), CancellationToken.None);
+ await fakeWorker.RespondToControlCommandAsync().WaitAsync(TestTimeout);
+ WorkerCommandReply reply = await invokeTask.WaitAsync(TestTimeout);
+ Assert.Equal(ProtocolStatusCode.Ok, reply.Reply.ProtocolStatus.Code);
+
+ await WaitUntilAsync(() => session.State == SessionState.Faulted);
+
+ Assert.Equal(WorkerClientState.Closed, client.State);
+ Assert.Contains("worker", session.FinalFault, StringComparison.OrdinalIgnoreCase);
+ Assert.True(manager.TryGetSession(session.SessionId, out _));
+ Assert.Equal(1, metrics.GetSnapshot().OpenSessions);
+ Assert.Equal(0, metrics.GetSnapshot().SessionsClosed);
+
+ int closedCount = await manager.CloseExpiredLeasesAsync(DateTimeOffset.UtcNow, CancellationToken.None);
+
+ Assert.Equal(1, closedCount);
+ Assert.Equal(SessionState.Closed, session.State);
+ Assert.False(manager.TryGetSession(session.SessionId, out _));
+ GatewayMetricsSnapshot snapshot = metrics.GetSnapshot();
+ Assert.Equal(1, snapshot.SessionsClosed);
+ Assert.Equal(0, snapshot.OpenSessions);
+ }
+
+ ///
+ /// The other way a worker goes away: its pipe breaks (crash). The worker client faults, and
+ /// the session must follow it to Faulted rather than stay Ready.
+ ///
+ /// A task that represents the asynchronous operation.
+ [Fact]
+ public async Task WorkerPipeBreaks_SessionFaults()
+ {
+ await using FakeWorkerHarness fakeWorker = await FakeWorkerHarness.CreateConnectedPairAsync();
+ using GatewayMetrics metrics = new();
+ WorkerClient client = fakeWorker.CreateClient(metrics: metrics);
+ await StartClientAsync(fakeWorker, client);
+ SessionManager manager = CreateManager(client, metrics);
+ GatewaySession session = await OpenSessionAsync(manager);
+ using IEventSubscriberLease stream = session.AttachEventSubscriber(maxSubscribers: 1);
+
+ await fakeWorker.DisposeWorkerSideAsync();
+
+ await WaitUntilAsync(() => session.State == SessionState.Faulted);
+ Assert.Equal(WorkerClientState.Faulted, client.State);
+
+ await manager.CloseExpiredLeasesAsync(DateTimeOffset.UtcNow, CancellationToken.None);
+ Assert.False(manager.TryGetSession(session.SessionId, out _));
+ Assert.Equal(1, metrics.GetSnapshot().SessionsClosed);
+ Assert.Equal(0, metrics.GetSnapshot().OpenSessions);
+ }
+
+ private static async Task StartClientAsync(FakeWorkerHarness fakeWorker, WorkerClient client)
+ {
+ Task startTask = client.StartAsync(CancellationToken.None);
+ await fakeWorker.CompleteStartupAsync();
+ await startTask.WaitAsync(TestTimeout);
+ Assert.Equal(WorkerClientState.Ready, client.State);
+ }
+
+ private static SessionManager CreateManager(IWorkerClient client, GatewayMetrics metrics)
+ {
+ return new SessionManager(
+ new SessionRegistry(),
+ new SingleClientFactory(client),
+ Options.Create(new GatewayOptions()),
+ metrics);
+ }
+
+ private static Task OpenSessionAsync(SessionManager manager)
+ {
+ return manager.OpenSessionAsync(
+ new SessionOpenRequest(RequestedBackend: null, ClientSessionName: "worker-exit", ClientCorrelationId: null, CommandTimeout: null),
+ "client-1",
+ ownerKeyId: null,
+ CancellationToken.None);
+ }
+
+ private static WorkerCommand CreateShutdownWorkerCommand()
+ {
+ return new WorkerCommand
+ {
+ Command = new MxCommand
+ {
+ Kind = MxCommandKind.ShutdownWorker,
+ ShutdownWorker = new ShutdownWorkerCommand(),
+ },
+ };
+ }
+
+ private static async Task WaitUntilAsync(Func condition)
+ {
+ DateTime deadline = DateTime.UtcNow + TestTimeout;
+ while (!condition())
+ {
+ if (DateTime.UtcNow > deadline)
+ {
+ Assert.Fail("Condition was not met within the test timeout.");
+ }
+
+ await Task.Delay(10);
+ }
+ }
+
+ private sealed class SingleClientFactory(IWorkerClient client) : ISessionWorkerClientFactory
+ {
+ ///
+ public Task CreateAsync(GatewaySession session, CancellationToken cancellationToken)
+ {
+ return Task.FromResult(client);
+ }
+ }
+}