using System.Collections;
using System.Collections.Concurrent;
using System.Diagnostics;
using System.Diagnostics.Tracing;
using System.Globalization;
using System.Net;
using System.Net.Sockets;
using System.Reflection;
using System.Runtime.ExceptionServices;
using System.Text;
using System.Text.Json;
using Opc.Ua;
using Opc.Ua.Client;
using Opc.Ua.Configuration;
using Opc.Ua.Server;
using Shouldly;
using Xunit;
using ZB.MOM.WW.OtOpcUa.Commons.OpcUa;
using ZB.MOM.WW.OtOpcUa.Configuration.Enums;
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
using ZB.MOM.WW.OtOpcUa.Core.Authorization;
using ClientSession = Opc.Ua.Client.ISession;
using Subscription = Opc.Ua.Client.Subscription;
using MonitoredItem = Opc.Ua.Client.MonitoredItem;
using ServerSession = Opc.Ua.Server.ISession;

namespace ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests;

/// <summary>
/// Opt-in Task 12 lifecycle evidence. Diagnostic mode is deliberately separate from the two-hour gate.
/// Run only this test in an otherwise idle test process; RSS/FD counts describe the whole process.
/// No key, certificate, store path, token, or OPC UA payload is written to the JSONL evidence.
/// </summary>
[Trait("Category", "E2E")]
public sealed partial class Sdk2LifecycleSoakTests
{
    private const string Scalar = ServerEventRig.RawDeviceFolder + "/SoakScalar";
    private const string ArrayTag = ServerEventRig.RawDeviceFolder + "/SoakArray";
    private const string UnsScalar = "SoakEquipment/Scalar";
    private static readonly DateTime Epoch = new(2026, 10, 1, 0, 0, 0, DateTimeKind.Utc);
    public static bool Enabled => Environment.GetEnvironmentVariable("OTOPCUA_SDK2_SOAK") == "1";

    [Fact(SkipUnless = nameof(Enabled), Skip = "Opt in with OTOPCUA_SDK2_SOAK=1 and an explicit soak mode.")]
    public async Task Repeated_lifecycles_return_resources_to_a_steady_baseline()
    {
        var mode = Environment.GetEnvironmentVariable("OTOPCUA_SDK2_SOAK_MODE");
        (mode is "diagnostic" or "profile" or "full" or "saturation" or LateSlopeRun.Mode).ShouldBeTrue("an explicit soak mode is required");
        var saturation = mode == "saturation" ? SaturationControl.FromEnvironment() : null;
        var lateSlope = mode == LateSlopeRun.Mode ? new LateSlopeRun() : null;
        var seconds = int.Parse(Environment.GetEnvironmentVariable("OTOPCUA_SDK2_SOAK_SECONDS")
            ?? (mode == "full" || lateSlope is not null ? "7200" : saturation is not null ? "1800" : "90"), CultureInfo.InvariantCulture);
        if (mode == "full") seconds.ShouldBeGreaterThanOrEqualTo(7200);
        else if (lateSlope is not null) LateSlopeRun.ValidateRequest(seconds);
        else if (saturation is not null) seconds.ShouldBe(1800, "the prospective saturation cutoff is fixed");
        else if (mode == "profile") seconds.ShouldBeInRange(60, 1800); // Investigative only; never satisfies the full-duration flag.
        else seconds.ShouldBeInRange(60, 120);
        var output = Environment.GetEnvironmentVariable("OTOPCUA_SDK2_SOAK_OUTPUT")
            ?? throw new InvalidOperationException("An absolute, fresh OTOPCUA_SDK2_SOAK_OUTPUT JSONL path is required.");
        Path.IsPathFullyQualified(output).ShouldBeTrue();
        Directory.CreateDirectory(Path.GetDirectoryName(output)!);
        await using var evidence = new StreamWriter(new FileStream(output, FileMode.CreateNew, FileAccess.Write, FileShare.Read));
        using var run = CancellationTokenSource.CreateLinkedTokenSource(TestContext.Current.CancellationToken);
        run.CancelAfter(TimeSpan.FromSeconds(saturation is null && lateSlope is null ? seconds + 120 : LateSlopeRun.StartupDeadline));
        var ct = run.Token;
        var watch = Stopwatch.StartNew();
        var clientRoot = Path.Combine(Path.GetTempPath(), $"otua-soak-client-{Guid.NewGuid():N}");
        ApplicationInstance? application = null;
        ServerEventRig? rig = null;
        string? serverRoot = null;
        var cycles = 0;
        var restarts = 0;
        Sample? baseline = null;
        ExceptionDispatchInfo? firstRejectedBudget = null;
        var retained = new Queue<Sample>();
        try
        {
            var telemetry = DefaultTelemetry.Create(_ => { });
            var security = TestClientSecurity.Build(clientRoot);
            security.ApplicationCertificate!.SubjectName = "CN=SDK2LifecycleSoak";
            var config = new ApplicationConfiguration(telemetry)
            {
                ApplicationName = "SDK2LifecycleSoak", ApplicationUri = $"urn:SDK2LifecycleSoak:{Guid.NewGuid():N}",
                ApplicationType = ApplicationType.Client, SecurityConfiguration = security,
                ClientConfiguration = new ClientConfiguration { DefaultSessionTimeout = 15_000 },
                TransportQuotas = new TransportQuotas { OperationTimeout = 5000 },
            };
            application = new ApplicationInstance(telemetry) { ApplicationConfiguration = config };
            await config.ValidateAsync(ApplicationType.Client, ct);
            await application.CheckApplicationInstanceCertificatesAsync(true, null, ct);
            rig = await ServerEventRig.StartAsync("SDK2LifecycleSoak", encrypted: true, ct);
            serverRoot = RequiredField<string>(rig, "_pkiRoot");
            Configure(rig);
            // The required observation period excludes initial certificate creation and server startup.
            var startupSeconds = watch.Elapsed.TotalSeconds;
            if (lateSlope is not null)
            {
                ct.ThrowIfCancellationRequested();
                startupSeconds.ShouldBeLessThanOrEqualTo(LateSlopeRun.StartupDeadline);
            }
            watch.Restart();
            if (saturation is not null) run.CancelAfter(TimeSpan.FromSeconds(1800));
            if (lateSlope is not null)
            {
                lateSlope.ClientManager = config.CertificateManager;
                lateSlope.PhaseStarted = true;
                run.CancelAfter(TimeSpan.FromSeconds(LateSlopeRun.PhaseDeadline));
            }
            var start = new
            {
                kind = "start", utc = DateTimeOffset.UtcNow, processId = Environment.ProcessId,
                mode, requestedSeconds = seconds, minimumFullSeconds = 7200,
                build = Environment.GetEnvironmentVariable("OTOPCUA_SOAK_BUILD_SHA"),
                sdk = typeof(OperationContext).Assembly.GetCustomAttribute<AssemblyInformationalVersionAttribute>()?.InformationalVersion,
                diagnosticInstrumentation = "SDK2.0.1 CP lists and certificate-validation cached stores; read-only",
                limits = new { retainedManagedGrowthBytes = 64L << 20, rssGrowthBytes = 128L << 20, fdGrowth = 32 },
                coverage = "encrypted sessions; scalar/Int16[] Raw+UNS; subscription removal; browse abandonment; history release/cross-session/close; surgical changes/rebuild; revocation; paused TCP read/recovery; same-PKI restart",
            };
            if (lateSlope is null) await LogAsync(evidence, start);
            else await LateSlopeRun.LogStartAsync(evidence, start, startupSeconds);
            while (saturation is not null ? cycles < 385
                : lateSlope is not null ? lateSlope.LastCompletedCycleSeconds < LateSlopeRun.Duration
                : watch.Elapsed < TimeSpan.FromSeconds(seconds))
            {
                if (saturation is not null)
                {
                    if (cycles == 320) await saturation.AttachAsync(ct);
                    if (cycles == 320) saturation.Boundary("window1-start", 321, restarts);
                    if (cycles == 352) saturation.Boundary("window2-start", 353, restarts);
                    if (cycles == 384)
                    {
                        await saturation.DetachAndDrainAsync(ct);
                        await LogAsync(evidence, new { kind = "saturation-settling-start", cycle = 385, restarts });
                    }
                }
                var cycle = Stopwatch.StartNew();
                using var cycleDeadline = lateSlope is null ? null : LateSlopeRun.StartCycleDeadline(ct);
                var cycleToken = cycleDeadline?.Token ?? ct;
                try
                {
                    await ExerciseAsync(rig, config, ++cycles, evidence, cycleToken);
                    await EventuallyAsync(() => PublicCounts(rig) == (0, 0, 0), cycleToken);
                    var sample = await CaptureAsync(rig, config.CertificateManager, watch.Elapsed, cycles, clientRoot, serverRoot, cycleToken);
                    if (lateSlope is not null) sample = sample with { Seconds = watch.Elapsed.TotalSeconds };
                    // Retain the failing observation too; rejected diagnostic rows never count as passing quiescent samples.
                    await LogAsync(evidence, new { kind = "resource-observation", resources = sample });
                    sample.Sessions.ShouldBe(0); sample.Subscriptions.ShouldBe(0); sample.Items.ShouldBe(0);
                    sample.BrowsePoints.ShouldBe(0); sample.HistoryPoints.ShouldBe(0);
                    sample.LiveDataItems.ShouldBe(0); sample.LiveEventItems.ShouldBe(0); sample.MonitoredNodes.ShouldBe(0);
                    sample.PkiDescriptors.ShouldBe(0, "quiescent certificate file descriptors must be closed");
                    if (cycles >= 3)
                    {
                        baseline ??= sample;
                        sample.CachedCertificateStores.ShouldBe(baseline.CachedCertificateStores);
                        sample.Variables.ShouldBe(baseline.Variables); sample.Folders.ShouldBe(baseline.Folders);
                        sample.Conditions.ShouldBe(baseline.Conditions);
                        if (firstRejectedBudget is null)
                        {
                            try
                            {
                                sample.Descriptors.ShouldBeLessThanOrEqualTo(baseline.Descriptors + 32);
                                sample.ManagedBytes.ShouldBeLessThanOrEqualTo(baseline.ManagedBytes + (64L << 20));
                                sample.RssBytes.ShouldBeLessThanOrEqualTo(baseline.RssBytes + (128L << 20));
                            }
                            catch (ShouldAssertException error) when (mode == "profile" || saturation is not null || lateSlope is not null)
                            {
                                // Investigative sampling may continue after rejection, but can never turn it into a pass.
                                firstRejectedBudget = ExceptionDispatchInfo.Capture(error);
                                await LogAsync(evidence, new { kind = "budget-rejected", cycle = cycles,
                                    seconds = watch.Elapsed.TotalSeconds, resources = sample, baseline, failure = error.ToString() });
                            }
                        }
                        retained.Enqueue(sample);
                        if (retained.Count > 24) retained.Dequeue();
                        if (retained.Count == 24)
                        {
                            var window = retained.ToArray();
                            var strictlyIncreasing = window.Zip(window.Skip(1)).All(pair => pair.Second.ManagedBytes > pair.First.ManagedBytes);
                            (strictlyIncreasing && window[^1].ManagedBytes - window[0].ManagedBytes > (16L << 20)).ShouldBeFalse(
                                "retained managed memory must not grow monotonically across 24 quiescent cycles");
                        }
                    }
                    await LogAsync(evidence, new { kind = firstRejectedBudget is null ? "quiescent" : "quiescent-budget-rejected",
                        cycleSeconds = cycle.Elapsed.TotalSeconds, resources = sample });
                    if (cycles % 4 == 0)
                    {
                        var restart = Stopwatch.StartNew();
                        double? boundedRestartSeconds = null;
                        if (lateSlope is null)
                            await RestartWithActiveResourcesAsync(rig, config, cycleToken).WaitAsync(TimeSpan.FromSeconds(25), cycleToken);
                        else
                        {
                            using var restartDeadline = LateSlopeRun.StartLinkedDeadline(cycleToken, TimeSpan.FromSeconds(25));
                            var boundedRestart = Stopwatch.StartNew();
                            try
                            {
                                await RestartWithActiveResourcesAsync(rig, config, restartDeadline.Token);
                                restartDeadline.Token.ThrowIfCancellationRequested();
                                boundedRestartSeconds = boundedRestart.Elapsed.TotalSeconds;
                                boundedRestartSeconds.Value.ShouldBeLessThanOrEqualTo(25);
                            }
                            finally { restartDeadline.Cancel(); }
                        }
                        Configure(rig);
                        restarts++;
                        if (lateSlope is null)
                            await LogAsync(evidence, new { kind = "restart", cycles, restarts, seconds = restart.Elapsed.TotalSeconds });
                        else
                            await LogAsync(evidence, new { kind = "restart", cycles, restarts, seconds = restart.Elapsed.TotalSeconds,
                                boundedRestartSeconds = boundedRestartSeconds!.Value });
                    }
                    if (saturation is not null)
                    {
                        if (cycles == 352) saturation.Boundary("window1-end", cycles, restarts);
                        if (cycles == 384) saturation.Boundary("window2-end", cycles, restarts);
                    }
                    if (lateSlope is not null)
                    {
                        cycleToken.ThrowIfCancellationRequested();
                        var cycleSeconds = cycle.Elapsed.TotalSeconds;
                        cycleSeconds.ShouldBeLessThanOrEqualTo(LateSlopeRun.CycleDeadline);
                        var elapsed = watch.Elapsed.TotalSeconds;
                        await LogAsync(evidence, new { kind = "late-slope-cycle", cycle = cycles, restarts,
                            elapsedSeconds = elapsed, cycleSeconds });
                        cycleToken.ThrowIfCancellationRequested();
                        cycle.Elapsed.TotalSeconds.ShouldBeLessThanOrEqualTo(LateSlopeRun.CycleDeadline);
                        lateSlope.LastCompletedCycleSeconds = elapsed;
                    }
                }
                finally { cycleDeadline?.Cancel(); }
            }
            if (lateSlope is not null)
            {
                watch.Stop();
                ct.ThrowIfCancellationRequested();
                watch.Elapsed.TotalSeconds.ShouldBeInRange(LateSlopeRun.Duration, LateSlopeRun.PhaseDeadline);
                cycles.ShouldBeGreaterThanOrEqualTo(4);
                restarts.ShouldBe(cycles / 4);
                lateSlope.DurationCompleted = true;
            }
            else
            {
                // Preserve the first assertion and its original stack even if later RSS samples decrease.
                // This happens before any completion record; finally still releases the owned rig and client.
                if (saturation is null) firstRejectedBudget?.Throw();
                var phaseSeconds = watch.Elapsed.TotalSeconds;
                if (saturation is not null)
                {
                    cycles.ShouldBe(385); restarts.ShouldBe(96);
                    phaseSeconds.ShouldBeLessThanOrEqualTo(1800, "there is no saturation cutoff extension");
                }
                else
                {
                    cycles.ShouldBeGreaterThanOrEqualTo(4, "diagnostic mode must exercise at least one actual restart");
                    restarts.ShouldBeGreaterThan(0);
                    watch.Elapsed.TotalSeconds.ShouldBeGreaterThanOrEqualTo(seconds);
                }
                var cleanupToken = saturation is null ? ct : TestContext.Current.CancellationToken;
                var finalClientCores = CertificateCores(config.CertificateManager);
                var finalServerCores = CertificateCores(rig.Server.CertificateManager!);
                await rig.DisposeAsync().AsTask().WaitAsync(TimeSpan.FromSeconds(20), cleanupToken);
                rig = null;
                await application.DisposeAsync().AsTask().WaitAsync(TimeSpan.FromSeconds(20), cleanupToken);
                application = null;
                finalClientCores.Sum(CachedStores).ShouldBe(0);
                finalServerCores.Sum(CachedStores).ShouldBe(0);
                (await DescriptorCountsAsync(cleanupToken, clientRoot, serverRoot)).Pki.ShouldBe(0);
                if (saturation is not null)
                {
                    await LogAsync(evidence, new { kind = "saturation-result", utc = DateTimeOffset.UtcNow,
                        canonicalCycles = 384, settlingCycles = 1, restarts, phaseSeconds,
                        readiness = saturation.Readiness, coldBudgetRejected = firstRejectedBudget is not null,
                        shutdownOwnershipVerified = true, acceptance = false });
                    firstRejectedBudget?.Throw();
                    saturation.Readiness.ShouldBeTrue("diagnostic readiness requires both fixed observation windows");
                    return;
                }
                await LogAsync(evidence, new { kind = "complete", utc = DateTimeOffset.UtcNow, mode, cycles, restarts, seconds = watch.Elapsed.TotalSeconds,
                    satisfiesTwoHourDuration = mode == "full" && watch.Elapsed.TotalSeconds >= 7200 });
            }
        }
        catch (Exception error) when (lateSlope is not null) { lateSlope.Record(error); }
        catch (Exception error)
        {
            await LogAsync(evidence, new { kind = "failed", cycles, restarts, seconds = watch.Elapsed.TotalSeconds,
                exceptionType = error.GetType().FullName });
            if (saturation is not null) firstRejectedBudget?.Throw();
            throw;
        }
        finally
        {
            if (lateSlope is not null)
            {
                watch.Stop();
                await lateSlope.ShutdownAsync(rig, application, clientRoot, serverRoot);
            }
            else
            {
                if (rig is not null) await rig.DisposeAsync().AsTask().WaitAsync(TimeSpan.FromSeconds(20), TestContext.Current.CancellationToken);
                if (application is not null) await application.DisposeAsync().AsTask().WaitAsync(TimeSpan.FromSeconds(20), TestContext.Current.CancellationToken);
                if (Directory.Exists(clientRoot)) Directory.Delete(clientRoot, true);
                if (serverRoot is not null) Directory.Exists(serverRoot).ShouldBeFalse("the task-owned server PKI must be released");
            }
        }
        if (lateSlope is not null)
            await lateSlope.EmitAndPreserveFailuresAsync(evidence, cycles, restarts, watch.Elapsed.TotalSeconds, firstRejectedBudget);
    }

    private static void Configure(ServerEventRig rig)
    {
        rig.MaterialiseNative();
        rig.Nm.HistorianDataSource = new EventHistoryPagingOverTheWireTests.SqlLikeEventSource(Enumerable.Range(0, 40)
            .Select(i => new HistoricalEvent($"soak-{i}", "source", Epoch.AddSeconds(i), Epoch, "soak", 500)).ToList());
        rig.Nm.NodeWriteGateway = new AcceptWrites();
        MaterialiseValues(rig);
        rig.Enforce(true, [ServerEventRig.Key()], Grant());
    }

    private static ZB.MOM.WW.OtOpcUa.Configuration.Entities.NodeAcl Grant() => ServerEventRig.ClusterGrant(
        ServerEventRig.KeyPrincipal, "soak-grant", NodePermissions.Browse | NodePermissions.Read | NodePermissions.Subscribe
            | NodePermissions.WriteOperate | NodePermissions.HistoryRead);

    private static void MaterialiseValues(ServerEventRig rig)
    {
        rig.Sink.EnsureFolder(ServerEventRig.RawDeviceFolder, null, "Device", AddressSpaceRealm.Raw);
        rig.Sink.EnsureVariable(Scalar, ServerEventRig.RawDeviceFolder, "Scalar", "Int32", true, AddressSpaceRealm.Raw);
        rig.Sink.EnsureVariable(ArrayTag, ServerEventRig.RawDeviceFolder, "Array", "Int16", true, AddressSpaceRealm.Raw,
            isArray: true, arrayLength: 4);
        rig.Sink.EnsureFolder("SoakEquipment", null, "Equipment", AddressSpaceRealm.Uns);
        rig.Sink.EnsureVariable(UnsScalar, "SoakEquipment", "Scalar", "Int32", true, AddressSpaceRealm.Uns);
        for (var i = 0; i < 16; i++) rig.Sink.EnsureVariable($"{ServerEventRig.RawDeviceFolder}/Browse{i}",
            ServerEventRig.RawDeviceFolder, $"Browse{i}", "Int32", false, AddressSpaceRealm.Raw);
        rig.Sink.WriteValue(Scalar, 0, OpcUaStatus.Good, DateTime.UtcNow, AddressSpaceRealm.Raw);
        rig.Sink.WriteValue(ArrayTag, new short[] { 0, 1, 2, 3 }, OpcUaStatus.Good, DateTime.UtcNow, AddressSpaceRealm.Raw);
        rig.Sink.WriteValue(UnsScalar, 0, OpcUaStatus.Good, DateTime.UtcNow, AddressSpaceRealm.Uns);
    }

    private static async Task ExerciseAsync(ServerEventRig rig, ApplicationConfiguration config, int cycle,
        StreamWriter evidence, CancellationToken ct)
    {
        await using var proxy = new PauseProxy(new Uri(rig.EndpointUrl).Port);
        using var owner = await ConnectAsync(rig, config, proxy.Port, ct);
        using var other = await ConnectAsync(rig, config, null, ct);
        var ownedSessions = rig.Server.CurrentInstance.SessionManager.GetSessions().ToArray();
        ownedSessions.Length.ShouldBe(2);
        var scalar = Raw(owner, Scalar);
        var array = Raw(owner, ArrayTag);
        var unsNs = owner.NamespaceUris.GetIndex(V3NodeIds.UnsNamespaceUri);
        unsNs.ShouldBeGreaterThan(0);
        var uns = new NodeId(UnsScalar, (ushort)unsNs);
        owner.NamespaceUris.GetString(scalar.NamespaceIndex).ShouldBe(V3NodeIds.RawNamespaceUri);
        owner.NamespaceUris.GetString(uns.NamespaceIndex).ShouldBe(V3NodeIds.UnsNamespaceUri);
        using var subscription = new Subscription(owner.DefaultSubscription) { PublishingInterval = 100, KeepAliveCount = 10 };
        owner.AddSubscription(subscription);
        try
        {
            await subscription.CreateAsync(ct);
            var seen = new ConcurrentDictionary<NodeId, Variant>();
            foreach (var id in new[] { scalar, array, uns })
            {
                var item = new MonitoredItem(subscription.DefaultItem) { StartNodeId = id, AttributeId = Attributes.Value,
                    SamplingInterval = 50, QueueSize = 10 };
                item.Notification += (sender, _) =>
                {
                    foreach (var value in sender.DequeueValues()) if (StatusCode.IsGood(value.StatusCode)) seen[id] = value.WrappedValue;
                };
                subscription.AddItem(item);
            }
            await subscription.ApplyChangesAsync(ct);
            subscription.MonitoredItems.ShouldAllBe(item => ServiceResult.IsGood(item.Status.Error));
            PublicCounts(rig).ShouldBe((2, 1, 3));
            var values = new short[] { (short)(cycle % 30000), 2, 3, 4 };
            var writes = await owner.WriteAsync(null, new[]
            {
                new WriteValue { NodeId = scalar, AttributeId = Attributes.Value, Value = new DataValue(Variant.From(cycle)) },
                new WriteValue { NodeId = array, AttributeId = Attributes.Value, Value = new DataValue(SdkTestValues.FromClr(values)) },
                new WriteValue { NodeId = uns, AttributeId = Attributes.Value, Value = new DataValue(Variant.From(cycle)) },
            }, ct);
            writes.Results.ToList().ShouldAllBe(status => status == StatusCodes.Good);
            (await owner.ReadValueAsync(scalar, ct)).WrappedValue.GetInt32().ShouldBe(cycle);
            (await owner.ReadValueAsync(array, ct)).ClrValue().ShouldBeOfType<short[]>().ShouldBe(values);
            (await owner.ReadValueAsync(uns, ct)).WrappedValue.GetInt32().ShouldBe(cycle);
            await EventuallyAsync(() => seen.TryGetValue(scalar, out var value) && value.GetInt32() == cycle
                && seen.ContainsKey(array) && seen.TryGetValue(uns, out var mirror) && mirror.GetInt32() == cycle, ct);

            // One explicitly released and one deliberately abandoned browse point.
            var browse = await BrowseAsync(owner, ct);
            browse.IsEmpty.ShouldBeFalse();
            (await owner.BrowseNextAsync(null, true, [browse], ct)).Results[0].StatusCode.ShouldBe(StatusCodes.Good);
            (await BrowseAsync(owner, ct)).IsEmpty.ShouldBeFalse();
            // Production history points: cross-session denial, single-use resume, explicit release and close cleanup.
            var first = (await PageAsync(owner, default, false, ct)).ContinuationPoint;
            first.IsEmpty.ShouldBeFalse();
            (await PageAsync(other, first, false, ct)).StatusCode.ShouldBe(StatusCodes.BadContinuationPointInvalid);
            var next = (await PageAsync(owner, first, false, ct)).ContinuationPoint;
            next.IsEmpty.ShouldBeFalse();
            (await PageAsync(owner, next, true, ct)).StatusCode.ShouldBe(StatusCodes.Good);
            for (var i = 0; i < 6; i++) (await PageAsync(owner, default, false, ct)).ContinuationPoint.IsEmpty.ShouldBeFalse();
            var held = ownedSessions.Select(ContinuationCounts).ToArray();
            held.Sum(count => count.Browse).ShouldBeGreaterThan(0);
            held.Sum(count => count.History).ShouldBeGreaterThan(0);
            var active = PublicCounts(rig);
            await LogAsync(evidence, new { kind = "active", cycle, sessions = active.Sessions,
                subscriptions = active.Subscriptions, items = active.Items, browsePoints = held.Sum(x => x.Browse), historyPoints = held.Sum(x => x.History) });

            rig.Nm.UpdateTagAttributes(Scalar, false, null, "Int32", false, null, AddressSpaceRealm.Raw).ShouldBeTrue();
            rig.Nm.UpdateTagAttributes(Scalar, true, null, "Int32", false, null, AddressSpaceRealm.Raw).ShouldBeTrue();
            rig.Nm.RemoveVariableNode(ArrayTag, AddressSpaceRealm.Raw).ShouldBeTrue();
            rig.Sink.EnsureVariable(ArrayTag, ServerEventRig.RawDeviceFolder, "Array", "Int16", true, AddressSpaceRealm.Raw,
                isArray: true, arrayLength: 4);
            rig.Sink.WriteValue(ArrayTag, values, OpcUaStatus.Good, DateTime.UtcNow, AddressSpaceRealm.Raw);
            var authority = rig.Nm.DataPlaneAuthority.ShouldBeOfType<DataPlaneAuthorityState>();
            authority.Apply(ServerEventRig.Cluster, $"revoked-{cycle}", [], [Grant()],
                new Dictionary<AclScopeRef, IReadOnlyList<NodeAclPath>>());
            var denied = await owner.ReadAsync(null, 0, TimestampsToReturn.Neither,
                [new ReadValueId { NodeId = scalar, AttributeId = Attributes.Value }], ct);
            denied.Results[0].StatusCode.ShouldBe(StatusCodes.BadUserAccessDenied);
            rig.Enforce(true, [ServerEventRig.Key()], Grant());

            proxy.Pause();
            var stalled = Stopwatch.StartNew();
            string stalledStatus;
            try
            {
                using var timeout = CancellationTokenSource.CreateLinkedTokenSource(ct);
                timeout.CancelAfter(TimeSpan.FromSeconds(1));
                await owner.ReadValueAsync(scalar, timeout.Token).WaitAsync(TimeSpan.FromSeconds(5), ct);
                throw new InvalidOperationException("The deliberately paused network unexpectedly completed the read.");
            }
            catch (OperationCanceledException) when (!ct.IsCancellationRequested) { stalledStatus = "OperationCanceled"; }
            // Exact-tag ChannelAsyncOperation maps a canceled completion waiter to BadRequestInterrupted.
            catch (ServiceResultException error) when (error.StatusCode == StatusCodes.BadTimeout
                || error.StatusCode == StatusCodes.BadRequestCancelledByClient || error.StatusCode == StatusCodes.BadRequestTimeout
                || error.StatusCode == StatusCodes.BadRequestInterrupted) { stalledStatus = error.StatusCode.ToString(); }
            finally { proxy.Resume(); }
            (await owner.ReadValueAsync(scalar, ct)).StatusCode.ShouldBe(StatusCodes.Good);
            await LogAsync(evidence, new { kind = "network-stall-recovered", cycle, status = stalledStatus, seconds = stalled.Elapsed.TotalSeconds });
            await subscription.DeleteAsync(true, ct);
            await owner.RemoveSubscriptionAsync(subscription, ct);
            rig.Sink.RebuildAddressSpace();
            Configure(rig);
        }
        finally
        {
            proxy.Resume();
            await owner.CloseAsync(ct).WaitAsync(TimeSpan.FromSeconds(10), ct);
            await other.CloseAsync(ct).WaitAsync(TimeSpan.FromSeconds(10), ct);
            foreach (var closed in ownedSessions)
            {
                ContinuationCounts(closed).ShouldBe((0, 0));
            }
        }
    }

    private static async Task RestartWithActiveResourcesAsync(ServerEventRig rig, ApplicationConfiguration config,
        CancellationToken ct)
    {
        using var client = await ConnectAsync(rig, config, null, ct);
        using var subscription = new Subscription(client.DefaultSubscription) { PublishingInterval = 100 };
        client.AddSubscription(subscription);
        await subscription.CreateAsync(ct);
        subscription.AddItem(new MonitoredItem(subscription.DefaultItem)
        {
            StartNodeId = Raw(client, Scalar), AttributeId = Attributes.Value, SamplingInterval = 50,
        });
        await subscription.ApplyChangesAsync(ct);
        PublicCounts(rig).ShouldBe((1, 1, 1));
        (await BrowseAsync(client, ct)).IsEmpty.ShouldBeFalse();
        (await PageAsync(client, default, false, ct)).ContinuationPoint.IsEmpty.ShouldBeFalse();
        var oldInternal = rig.Server.CurrentInstance;
        var oldSessionManager = oldInternal.SessionManager;
        var oldSubscriptionManager = oldInternal.SubscriptionManager;
        var oldSessions = oldSessionManager.GetSessions().ToArray();
        var oldSubscriptions = oldSubscriptionManager.GetSubscriptions().ToArray();
        var oldManager = rig.Server.CertificateManager!;
        var oldCores = CertificateCores(oldManager);
        await rig.RestartAsync(ct).WaitAsync(TimeSpan.FromSeconds(20), ct);
        oldSessionManager.GetSessions().ShouldBeEmpty();
        oldSubscriptionManager.GetSubscriptions().ShouldBeEmpty();
        oldSubscriptions.ShouldAllBe(item => item.MonitoredItemCount == 0);
        foreach (var session in oldSessions) ContinuationCounts(session).ShouldBe((0, 0));
        RequiredField<bool>(oldManager, "m_disposed").ShouldBeTrue();
        oldCores.Sum(CachedStores).ShouldBe(0, "stopped server validators must release every cached store");
        PublicCounts(rig).ShouldBe((0, 0, 0));
    }

    private static async Task<ClientSession> ConnectAsync(ServerEventRig rig, ApplicationConfiguration config,
        int? proxyPort, CancellationToken ct)
    {
        var telemetry = DefaultTelemetry.Create(_ => { });
        var endpoint = await CoreClientUtils.SelectEndpointAsync(config, rig.EndpointUrl, true, telemetry, ct)
            ?? throw new InvalidOperationException("No encrypted endpoint.");
        endpoint.SecurityMode.ShouldBe(MessageSecurityMode.SignAndEncrypt);
        if (proxyPort is { } port) endpoint.EndpointUrl = $"opc.tcp://127.0.0.1:{port}/OtOpcUa";
        return await new DefaultSessionFactory(telemetry).CreateAsync(config,
            new ConfiguredEndpoint(null, endpoint, EndpointConfiguration.Create(config)), false, false,
            "SDK2LifecycleSoak", 15_000, new UserIdentity("apikey", Encoding.UTF8.GetBytes(ServerEventRig.KeyPassword)), default, ct);
    }

    private static NodeId Raw(ClientSession session, string id)
    {
        var ns = session.NamespaceUris.GetIndex(V3NodeIds.RawNamespaceUri);
        ns.ShouldBeGreaterThan(0);
        return new NodeId(id, (ushort)ns);
    }

    private static async Task<ByteString> BrowseAsync(ClientSession session, CancellationToken ct) =>
        (await session.BrowseAsync(null, null, 1, [new BrowseDescription
        {
            NodeId = Raw(session, ServerEventRig.RawDeviceFolder), BrowseDirection = BrowseDirection.Forward,
            ReferenceTypeId = ReferenceTypeIds.HierarchicalReferences, IncludeSubtypes = true,
            ResultMask = (uint)BrowseResultMask.All,
        }], ct)).Results[0].ContinuationPoint;

    private static async Task<Opc.Ua.HistoryReadResult> PageAsync(ClientSession session, ByteString point,
        bool release, CancellationToken ct) => (await session.HistoryReadAsync(null, new ExtensionObject(new ReadEventDetails
        {
            StartTime = Epoch.AddDays(-1), EndTime = Epoch.AddDays(1), NumValuesPerNode = 4,
            Filter = new EventFilter { SelectClauses = [new SimpleAttributeOperand(ObjectTypeIds.BaseEventType,
                QualifiedName.From(BrowseNames.EventId))] },
        }), TimestampsToReturn.Both, release,
        [new HistoryReadValueId { NodeId = Raw(session, ServerEventRig.RawDeviceFolder), ContinuationPoint = point }], ct)).Results[0];

    private static (int Sessions, int Subscriptions, int Items) PublicCounts(ServerEventRig rig)
    {
        var subscriptions = rig.Server.CurrentInstance.SubscriptionManager.GetSubscriptions();
        return (rig.Server.CurrentInstance.SessionManager.GetSessions().Count, subscriptions.Count,
            subscriptions.Sum(subscription => subscription.MonitoredItemCount));
    }

    // Exact-tag diagnostic instrumentation only. Fail if fields/types change; never replace or mutate a store.
    private static (int Browse, int History) ContinuationCounts(ServerSession session)
    {
        var points = session.ContinuationPoints;
        using (RequiredField<System.Threading.Lock>(points, "m_lock").EnterScope())
            return (NullableCollectionCount(points, "m_browse"), NullableCollectionCount(points, "m_history"));
    }

    private static object[] CertificateCores(object manager)
    {
        using (RequiredField<System.Threading.Lock>(manager, "m_certificatesLock").EnterScope())
        {
            var cores = new List<object>();
            foreach (var field in new[] { "m_peerCore", "m_userCore", "m_httpsCore" })
                if (Field(manager, field).GetValue(manager) is { } core) cores.Add(core);
            cores.AddRange(RequiredField<IDictionary>(manager, "m_customCores").Values.Cast<object>());
            return cores.Distinct(System.Collections.Generic.ReferenceEqualityComparer.Instance).ToArray();
        }
    }

    private static int CachedStores(object core) => RequiredField<IDictionary>(core, "m_stores").Count;
    private static int NullableCollectionCount(object owner, string name) => Field(owner, name).GetValue(owner) switch
    {
        null => 0,
        ICollection items => items.Count,
        _ => throw new InvalidOperationException($"SDK diagnostic field {name} no longer has the expected collection type."),
    };
    private static FieldInfo Field(object owner, string name) => owner.GetType().GetField(name,
        BindingFlags.Instance | BindingFlags.NonPublic) ?? throw new InvalidOperationException($"Missing SDK diagnostic field {name}.");
    private static T RequiredField<T>(object owner, string name) => Field(owner, name).GetValue(owner) is T value ? value
        : throw new InvalidOperationException($"SDK diagnostic field {name} no longer has the expected type {typeof(T).Name}.");

    private sealed record Sample(int Cycle, double Seconds, int Sessions, int Subscriptions, int Items,
        int BrowsePoints, int HistoryPoints, int LiveDataItems, int LiveEventItems, int MonitoredNodes,
        int Variables, int Folders, int Conditions, int CachedCertificateStores,
        int Descriptors, int PkiDescriptors, long ManagedBytes, long RssBytes, long AllocatedBytes,
        int Gen0, int Gen1, int Gen2, long GcCommittedBytes, long GcHeapBytes, long GcFragmentedBytes,
        long PinnedObjects, long PendingFinalizers, long JitMethods, long JitIlBytes, long ActiveTimers, int Threads);

    private static async Task<Sample> CaptureAsync(ServerEventRig rig, object clientManager, TimeSpan elapsed, int cycle,
        string clientRoot, string serverRoot, CancellationToken ct)
    {
        var counts = PublicCounts(rig);
        var points = rig.Server.CurrentInstance.SessionManager.GetSessions().Select(ContinuationCounts).ToArray();
        var descriptors = await DescriptorCountsAsync(ct, clientRoot, serverRoot);
        var retained = GC.GetTotalMemory(forceFullCollection: true);
        var memory = GC.GetGCMemoryInfo();
        using var process = Process.GetCurrentProcess();
        process.Refresh();
        return new Sample(cycle, elapsed.TotalSeconds, counts.Sessions, counts.Subscriptions, counts.Items,
            points.Sum(x => x.Browse), points.Sum(x => x.History), rig.Nm.LiveItemCount, rig.Nm.LiveEventItemCount,
            rig.Nm.MonitoredNodeCount, rig.Nm.VariableCount, rig.Nm.FolderCount, rig.Nm.AlarmConditionCount,
            CertificateCores(rig.Server.CertificateManager!).Sum(CachedStores) + CertificateCores(clientManager).Sum(CachedStores),
            descriptors.Total, descriptors.Pki, retained, process.WorkingSet64, GC.GetTotalAllocatedBytes(),
            GC.CollectionCount(0), GC.CollectionCount(1), GC.CollectionCount(2), memory.TotalCommittedBytes,
            memory.HeapSizeBytes, memory.FragmentedBytes, memory.PinnedObjectsCount, memory.FinalizationPendingCount,
            System.Runtime.JitInfo.GetCompiledMethodCount(), System.Runtime.JitInfo.GetCompiledILBytes(), Timer.ActiveCount,
            process.Threads.Count);
    }

    public static bool DescriptorProbeSupported => OperatingSystem.IsLinux() || OperatingSystem.IsMacOS();

    [Fact(SkipUnless = nameof(DescriptorProbeSupported), Skip = "Descriptor probes require Linux procfs or macOS lsof.")]
    public async Task Descriptor_probe_observes_an_open_PKI_file_through_a_symlink_alias_and_its_close()
    {
        var ct = TestContext.Current.CancellationToken;
        var temporary = Path.Combine(OperatingSystem.IsMacOS() ? "/tmp" : Path.GetTempPath(), $"otua-pki-probe-{Guid.NewGuid():N}");
        var pki = Path.Combine(temporary, "pki");
        var sibling = Path.Combine(temporary, "pki-other");
        var alias = Path.Combine(temporary, "pki-alias");
        Directory.CreateDirectory(pki);
        Directory.CreateDirectory(sibling);
        try
        {
            // On macOS the temporary root itself can use /var -> /private/var. The additional
            // task-owned alias also checks links whose target contains a linked parent component.
            Directory.CreateSymbolicLink(alias, pki);
            await using var outside = new FileStream(Path.Combine(sibling, "control"), FileMode.Create,
                FileAccess.ReadWrite, FileShare.ReadWrite | FileShare.Delete);
            CanonicalPath(alias).ShouldBe(CanonicalPath(pki));
            var before = (await DescriptorCountsAsync(ct, pki, alias)).Pki;
            before.ShouldBe(0, "a sibling sharing the PKI path prefix is outside the root");
            int opened;
            await using (var held = new FileStream(Path.Combine(alias, "control"), FileMode.Create,
                FileAccess.ReadWrite, FileShare.ReadWrite | FileShare.Delete))
            {
                opened = (await DescriptorCountsAsync(ct, pki, alias)).Pki;
                opened.ShouldBeGreaterThan(0, "the positive control must observe the task-owned file opened through the alias");
            }
            var closed = (await DescriptorCountsAsync(ct, pki, alias)).Pki;
            closed.ShouldBe(0, "closing the positive control must release its descriptor");
            TestContext.Current.TestOutputHelper!.WriteLine($"PKI descriptor control: before={before}, opened={opened}, closed={closed}; siblingExcluded=true; aliasResolved=true");
        }
        finally
        {
            if (Directory.Exists(alias)) Directory.Delete(alias);
            if (Directory.Exists(temporary)) Directory.Delete(temporary, recursive: true);
        }
    }

    [Fact(SkipUnless = nameof(DescriptorProbeSupported), Skip = "The bounded probe control uses a task-owned Unix process.")]
    public async Task Descriptor_probe_times_out_and_reaps_a_process_that_keeps_both_streams_open()
    {
        var ct = TestContext.Current.CancellationToken;
        using var process = new Process
        {
            StartInfo = new ProcessStartInfo("/bin/sh")
            {
                RedirectStandardOutput = true, RedirectStandardError = true, UseShellExecute = false,
                ArgumentList = { "-c", "printf control; printf control >&2; exec /bin/sleep 30" },
            },
        };
        var elapsed = Stopwatch.StartNew();
        await Should.ThrowAsync<TimeoutException>(() => ReadDescriptorProbeAsync(process, TimeSpan.FromMilliseconds(250), ct));
        elapsed.Elapsed.ShouldBeLessThan(TimeSpan.FromSeconds(4), "timeout includes stream drains and bounded process cleanup");
        process.HasExited.ShouldBeTrue("the task-owned timed-out process must be killed and reaped before returning");
        TestContext.Current.TestOutputHelper!.WriteLine($"Descriptor timeout control: deadlineMs=250, elapsedMs={elapsed.Elapsed.TotalMilliseconds:F0}, childExited={process.HasExited}");
    }

    private static async Task<(int Total, int Pki)> DescriptorCountsAsync(CancellationToken ct, params string[] roots)
    {
        ct.ThrowIfCancellationRequested();
        var canonicalRoots = roots.Select(root => CanonicalPath(root)).Distinct(StringComparer.Ordinal).ToArray();
        if (OperatingSystem.IsLinux())
        {
            var targets = Directory.EnumerateFileSystemEntries("/proc/self/fd")
                .Select(path => new FileInfo(path).LinkTarget).ToArray();
            return (targets.Length, targets.Count(path => IsWithinRoots(path, canonicalRoots)));
        }
        if (OperatingSystem.IsMacOS())
        {
            using var process = new Process { StartInfo = new ProcessStartInfo("/usr/sbin/lsof")
                { RedirectStandardOutput = true, RedirectStandardError = true, UseShellExecute = false } };
            foreach (var arg in new[] { "-a", "-p", Environment.ProcessId.ToString(CultureInfo.InvariantCulture), "-Fnf" })
                process.StartInfo.ArgumentList.Add(arg);
            var text = await ReadDescriptorProbeAsync(process, TimeSpan.FromSeconds(5), ct);
            process.ExitCode.ShouldBe(0);
            var total = 0;
            var pki = 0;
            var descriptor = false;
            foreach (var line in text.Split('\n'))
            {
                if (line.StartsWith('f'))
                {
                    descriptor = line.Length > 1 && char.IsDigit(line[1]);
                    if (descriptor) total++;
                }
                else if (descriptor && line.StartsWith('n') && IsWithinRoots(line[1..], canonicalRoots)) pki++;
            }
            return (total, pki);
        }
        throw new PlatformNotSupportedException("This soak requires Linux procfs or macOS lsof for actual descriptor evidence.");
    }

    private static async Task<string> ReadDescriptorProbeAsync(Process process, TimeSpan timeout, CancellationToken ct)
    {
        using var deadline = CancellationTokenSource.CreateLinkedTokenSource(ct);
        deadline.CancelAfter(timeout);
        process.Start();
        var stdout = process.StandardOutput.ReadToEndAsync(deadline.Token);
        var stderr = process.StandardError.ReadToEndAsync(deadline.Token);
        var exited = process.WaitForExitAsync(deadline.Token);
        try
        {
            // Waiting for exit alone does not bound reads: both redirected pipes must also finish.
            await Task.WhenAll(stdout, stderr, exited).WaitAsync(deadline.Token).ConfigureAwait(false);
            return await stdout.ConfigureAwait(false);
        }
        catch (OperationCanceledException)
        {
            try { if (!process.HasExited) process.Kill(entireProcessTree: true); }
            catch (InvalidOperationException) when (process.HasExited) { /* It exited between the check and kill. */ }
            await process.WaitForExitAsync(CancellationToken.None).WaitAsync(TimeSpan.FromSeconds(2)).ConfigureAwait(false);
            // Observe both cancelled drains; never leave a pipe read running after returning.
            try { await Task.WhenAll(stdout, stderr).WaitAsync(TimeSpan.FromSeconds(2)).ConfigureAwait(false); }
            catch (OperationCanceledException) { }
            if (ct.IsCancellationRequested) throw;
            throw new TimeoutException("Descriptor probe exceeded its process and redirected-stream deadline.");
        }
    }

    private static bool IsWithinRoots(string? path, IReadOnlyList<string> roots)
    {
        if (path is null || !Path.IsPathFullyQualified(path)) return false;
        // procfs marks an unlinked but still open file; its containing physical directory still counts.
        const string deleted = " (deleted)";
        if (path.EndsWith(deleted, StringComparison.Ordinal)) path = path[..^deleted.Length];
        var canonical = CanonicalPath(path);
        return roots.Any(root => canonical.Equals(root, StringComparison.Ordinal)
            || canonical.StartsWith(Path.EndsInDirectorySeparator(root) ? root : root + Path.DirectorySeparatorChar,
                StringComparison.Ordinal));
    }

    private static string CanonicalPath(string path, int depth = 0)
    {
        if (depth > 32) throw new IOException("Descriptor path link resolution exceeded its bound.");
        var full = Path.GetFullPath(path);
        var current = Path.GetPathRoot(full)!;
        foreach (var component in full[current.Length..].Split(Path.DirectorySeparatorChar, StringSplitOptions.RemoveEmptyEntries))
        {
            current = Path.Combine(current, component);
            FileSystemInfo entry = Directory.Exists(current) ? new DirectoryInfo(current) : new FileInfo(current);
            if (entry.LinkTarget is not null)
            {
                var target = entry.ResolveLinkTarget(returnFinalTarget: true)
                    ?? throw new IOException("Descriptor path link resolution returned no target.");
                // Resolve parent links too: an alias target can itself begin with /var or /tmp.
                current = CanonicalPath(target.FullName, depth + 1);
            }
        }
        return Path.TrimEndingDirectorySeparator(current);
    }

    private static async Task EventuallyAsync(Func<bool> condition, CancellationToken ct)
    {
        var deadline = Stopwatch.StartNew();
        while (!condition() && deadline.Elapsed < TimeSpan.FromSeconds(10)) await Task.Delay(25, ct);
        condition().ShouldBeTrue("resource or notification condition did not settle within ten seconds");
    }
    private static async Task LogAsync(StreamWriter output, object value)
    {
        await output.WriteLineAsync(JsonSerializer.Serialize(value));
        await output.FlushAsync();
    }

    // Test-only IPC. Receipts are fresh, run/PID-bound and published atomically by a separate observer.
    private sealed class SaturationControl(string directory, string runId)
    {
        private const string Protocol = "sdk2-saturation-v1";
        private int _observerPid;
        private int _nextMarker;
        internal bool Readiness { get; private set; }
        internal static SaturationControl FromEnvironment()
        {
            var directory = Environment.GetEnvironmentVariable("OTOPCUA_SDK2_SATURATION_CONTROL")
                ?? throw new InvalidOperationException("A fresh saturation control directory is required.");
            var runId = Environment.GetEnvironmentVariable("OTOPCUA_SDK2_SATURATION_RUN_ID")
                ?? throw new InvalidOperationException("An explicit saturation run id is required.");
            if (!Path.IsPathFullyQualified(directory) || !Guid.TryParseExact(runId, "D", out _))
                throw new InvalidOperationException("Saturation control identity is malformed.");
            Directory.CreateDirectory(directory);
            if (Directory.EnumerateFileSystemEntries(directory).Any())
                throw new InvalidOperationException("Saturation control directory must be empty.");
            _ = SaturationMarkers.Log;
            return new SaturationControl(directory, runId);
        }
        internal async Task AttachAsync(CancellationToken ct)
        {
            await PublishAsync("attach-request.json", new { protocol = Protocol, runId,
                targetPid = Environment.ProcessId, firstCycle = 321, completedRestarts = 80 }, ct);
            using var receipt = await WaitAsync("active.json", ct);
            var root = receipt.RootElement;
            ValidateIdentity(root);
            root.GetProperty("active").GetBoolean().ShouldBeTrue();
            _observerPid = root.GetProperty("observerPid").GetInt32();
            if (_observerPid <= 0 || _observerPid == Environment.ProcessId)
                throw new InvalidDataException("A distinct observer process is required.");
            using var enabled = CancellationTokenSource.CreateLinkedTokenSource(ct);
            enabled.CancelAfter(TimeSpan.FromSeconds(5));
            while (!SaturationMarkers.Log.IsEnabled()) await Task.Delay(10, enabled.Token);
        }
        internal void Boundary(string name, int cycle, int restarts)
        {
            (string Name, int Cycle, int Restarts)[] expected =
            [ ("window1-start", 321, 80), ("window1-end", 352, 88),
              ("window2-start", 353, 88), ("window2-end", 384, 96) ];
            if (_observerPid == 0 || _nextMarker >= expected.Length || expected[_nextMarker] != (name, cycle, restarts))
                throw new InvalidOperationException("Saturation boundary is out of order.");
            if (!SaturationMarkers.Log.IsEnabled()) throw new InvalidOperationException("Saturation observer is not active.");
            SaturationMarkers.Log.Boundary(runId, name, cycle, restarts);
            _nextMarker++;
        }
        internal async Task DetachAndDrainAsync(CancellationToken ct)
        {
            if (_nextMarker != 4) throw new InvalidOperationException("Both fixed observation windows are required.");
            await PublishAsync("detach-request.json", new { protocol = Protocol, runId,
                targetPid = Environment.ProcessId, lastCycle = 384, completedRestarts = 96 }, ct);
            using var receipt = await WaitAsync("drained.json", ct);
            Readiness = ValidateDrained(receipt.RootElement, runId, Environment.ProcessId, _observerPid);
            using var disabled = CancellationTokenSource.CreateLinkedTokenSource(ct);
            disabled.CancelAfter(TimeSpan.FromSeconds(5));
            while (SaturationMarkers.Log.IsEnabled()) await Task.Delay(10, disabled.Token);
        }
        internal static bool ValidateDrained(JsonElement root, string expectedRun, int target, int observer)
        {
            if (root.GetProperty("protocol").GetString() != Protocol || root.GetProperty("runId").GetString() != expectedRun
                || root.GetProperty("targetPid").GetInt32() != target || root.GetProperty("observerPid").GetInt32() != observer
                || !root.GetProperty("drained").GetBoolean() || root.GetProperty("active").GetBoolean())
                throw new InvalidDataException("Saturation drained receipt identity or ownership is invalid.");
            var windows = root.GetProperty("windows").EnumerateArray().ToArray();
            if (windows.Length != 2) throw new InvalidDataException("Exactly two fixed windows are required.");
            var ready = root.GetProperty("markersValid").GetBoolean();
            foreach (var name in new[] { "eventsLost", "sdkFactoryJit", "sdkFactoryLoads", "sdkAssemblyLoads" })
            {
                var value = root.GetProperty(name).GetInt64();
                if (value < 0) throw new InvalidDataException("Negative diagnostic count.");
                ready &= value == 0;
            }
            for (var index = 0; index < 2; index++)
            {
                var window = windows[index];
                if (window.GetProperty("name").GetString() != $"window{index + 1}"
                    || window.GetProperty("firstCycle").GetInt32() != 321 + index * 32
                    || window.GetProperty("lastCycle").GetInt32() != 352 + index * 32
                    || window.GetProperty("startRestarts").GetInt32() != 80 + index * 8
                    || window.GetProperty("endRestarts").GetInt32() != 88 + index * 8)
                    throw new InvalidDataException("Saturation window boundaries are invalid.");
                foreach (var name in new[] { "sdkFactoryJit", "sdkFactoryLoads", "sdkAssemblyLoads" })
                {
                    var value = window.GetProperty(name).GetInt64();
                    if (value < 0) throw new InvalidDataException("Negative window diagnostic count.");
                    ready &= value == 0;
                }
            }
            return ready;
        }
        private void ValidateIdentity(JsonElement root)
        {
            if (root.GetProperty("protocol").GetString() != Protocol || root.GetProperty("runId").GetString() != runId
                || root.GetProperty("targetPid").GetInt32() != Environment.ProcessId)
                throw new InvalidDataException("Saturation observer receipt identity is invalid.");
        }
        private async Task PublishAsync(string name, object value, CancellationToken ct)
        {
            var temporary = Path.Combine(directory, name + ".tmp");
            await using (var stream = new FileStream(temporary, FileMode.CreateNew, FileAccess.Write))
                await JsonSerializer.SerializeAsync(stream, value, cancellationToken: ct);
            File.Move(temporary, Path.Combine(directory, name));
        }
        private async Task<JsonDocument> WaitAsync(string name, CancellationToken ct)
        {
            using var bounded = CancellationTokenSource.CreateLinkedTokenSource(ct);
            bounded.CancelAfter(TimeSpan.FromSeconds(45));
            var path = Path.Combine(directory, name);
            while (!File.Exists(path))
            {
                if (File.Exists(Path.Combine(directory, "observer-failed.json")))
                    throw new InvalidOperationException("The saturation observer failed before completing its receipt.");
                await Task.Delay(25, bounded.Token);
            }
            return JsonDocument.Parse(await File.ReadAllTextAsync(path, bounded.Token));
        }
    }

    [Fact]
    public void Saturation_protocol_rejects_stale_identity_and_nonzero_readiness_counts()
    {
        var run = Guid.NewGuid().ToString("D");
        JsonDocument Receipt(string identity, int pid = 100, int lost = 0, int jit = 0, int loads = 0, int assemblies = 0) =>
            JsonDocument.Parse(JsonSerializer.Serialize(new { protocol = "sdk2-saturation-v1", runId = identity,
                targetPid = pid, observerPid = 200, active = false, drained = true, markersValid = true,
                eventsLost = lost, sdkFactoryJit = jit, sdkFactoryLoads = loads, sdkAssemblyLoads = assemblies,
                windows = new[] {
                    new { name = "window1", firstCycle = 321, lastCycle = 352, startRestarts = 80, endRestarts = 88,
                        sdkFactoryJit = jit, sdkFactoryLoads = loads, sdkAssemblyLoads = assemblies },
                    new { name = "window2", firstCycle = 353, lastCycle = 384, startRestarts = 88, endRestarts = 96,
                        sdkFactoryJit = 0, sdkFactoryLoads = 0, sdkAssemblyLoads = 0 } } }));
        using var good = Receipt(run);
        SaturationControl.ValidateDrained(good.RootElement, run, 100, 200).ShouldBeTrue();
        using var stale = Receipt(Guid.NewGuid().ToString("D"));
        Should.Throw<InvalidDataException>(() => SaturationControl.ValidateDrained(stale.RootElement, run, 100, 200));
        using var wrongPid = Receipt(run, pid: 101);
        Should.Throw<InvalidDataException>(() => SaturationControl.ValidateDrained(wrongPid.RootElement, run, 100, 200));
        foreach (var denied in new[] { Receipt(run, lost: 1), Receipt(run, jit: 1), Receipt(run, loads: 1), Receipt(run, assemblies: 1) })
        {
            using (denied) SaturationControl.ValidateDrained(denied.RootElement, run, 100, 200).ShouldBeFalse();
        }
    }

    [Fact]
    public async Task Saturation_protocol_missing_observer_is_bounded()
    {
        var directory = Path.Combine(Path.GetTempPath(), $"otua-saturation-control-{Guid.NewGuid():N}");
        Directory.CreateDirectory(directory);
        try
        {
            using var deadline = CancellationTokenSource.CreateLinkedTokenSource(TestContext.Current.CancellationToken);
            deadline.CancelAfter(TimeSpan.FromMilliseconds(100));
            var control = new SaturationControl(directory, Guid.NewGuid().ToString("D"));
            await Should.ThrowAsync<OperationCanceledException>(() => control.AttachAsync(deadline.Token));
            File.Exists(Path.Combine(directory, "active.json")).ShouldBeFalse();
        }
        finally { Directory.Delete(directory, recursive: true); }
    }

    public static bool SaturationObserverAvailable => File.Exists(Environment.GetEnvironmentVariable("OTOPCUA_SDK2_SATURATION_OBSERVER"));

    [Theory(SkipUnless = nameof(SaturationObserverAvailable), Skip = "Requires the separately built diagnostic observer.")]
    [InlineData(false)]
    [InlineData(true)]
    public async Task Saturation_protocol_actual_observer_drains_and_rejects_extra_foreign_markers(bool foreignMarker)
    {
        var directory = Path.Combine(Path.GetTempPath(), $"otua-saturation-observer-{Guid.NewGuid():N}");
        Directory.CreateDirectory(directory);
        var run = Guid.NewGuid().ToString("D");
        var control = new SaturationControl(directory, run);
        var observerPath = Environment.GetEnvironmentVariable("OTOPCUA_SDK2_SATURATION_OBSERVER")!;
        var frameworkDependent = observerPath.EndsWith(".dll", StringComparison.OrdinalIgnoreCase);
        using var observer = new Process { StartInfo = new ProcessStartInfo(frameworkDependent ? "dotnet" : observerPath)
            { RedirectStandardOutput = true, RedirectStandardError = true, UseShellExecute = false } };
        if (frameworkDependent) observer.StartInfo.ArgumentList.Add(observerPath);
        observer.StartInfo.ArgumentList.Add(directory);
        observer.StartInfo.ArgumentList.Add(run);
        using var deadline = CancellationTokenSource.CreateLinkedTokenSource(TestContext.Current.CancellationToken);
        deadline.CancelAfter(TimeSpan.FromSeconds(40));
        try
        {
            observer.Start().ShouldBeTrue();
            var stdout = observer.StandardOutput.ReadToEndAsync(deadline.Token);
            var stderr = observer.StandardError.ReadToEndAsync(deadline.Token);
            await control.AttachAsync(deadline.Token);
            if (!foreignMarker) SaturationProtocolRuntimeProbe(1).ShouldBe(2);
            if (foreignMarker) SaturationMarkers.Log.Boundary(Guid.NewGuid().ToString("D"), "window1-start", 321, 80);
            control.Boundary("window1-start", 321, 80);
            await Task.Delay(20, deadline.Token);
            control.Boundary("window1-end", 352, 88);
            control.Boundary("window2-start", 353, 88);
            await Task.Delay(20, deadline.Token);
            control.Boundary("window2-end", 384, 96);
            await control.DetachAndDrainAsync(deadline.Token);
            control.Readiness.ShouldBe(!foreignMarker);
            if (!foreignMarker)
            {
                using var parsed = JsonDocument.Parse(await File.ReadAllTextAsync(Path.Combine(directory, "drained.json"), deadline.Token));
                parsed.RootElement.GetProperty("runtimeControlJit").GetInt32().ShouldBeGreaterThan(0,
                    "the known non-SDK probe must prove runtime-provider delivery and method classification");
            }
            SaturationMarkers.Log.IsEnabled().ShouldBeFalse("Stop and stream drain precede return to the target");
            await observer.WaitForExitAsync(deadline.Token);
            observer.ExitCode.ShouldBe(0, await stderr);
            await stdout;
        }
        finally
        {
            if (!observer.HasExited) { observer.Kill(entireProcessTree: true); await observer.WaitForExitAsync(TestContext.Current.CancellationToken); }
            Directory.Delete(directory, recursive: true);
        }
    }

    [System.Runtime.CompilerServices.MethodImpl(System.Runtime.CompilerServices.MethodImplOptions.NoInlining
        | System.Runtime.CompilerServices.MethodImplOptions.NoOptimization)]
    private static int SaturationProtocolRuntimeProbe(int value) => value + 1;

    [EventSource(Name = "OtOpcUa-Sdk2-Saturation")]
    private sealed class SaturationMarkers : EventSource
    {
        internal static readonly SaturationMarkers Log = new();
        [Event(1, Level = EventLevel.Informational)]
        public void Boundary(string runId, string name, int cycle, int restarts) => WriteEvent(1, runId, name, cycle, restarts);
    }
    private sealed class AcceptWrites : IOpcUaNodeWriteGateway
    {
        public Task<NodeWriteOutcome> WriteAsync(string id, object? value, AddressSpaceRealm realm, CancellationToken ct) =>
            Task.FromResult(new NodeWriteOutcome(true, null));
    }

    /// <summary>Task-owned TCP relay. Pause blocks forwarding in both directions without closing either socket.</summary>
    private sealed class PauseProxy : IAsyncDisposable
    {
        private readonly TcpListener _listener = new(IPAddress.Loopback, 0);
        private readonly CancellationTokenSource _stop = new();
        private readonly ConcurrentBag<TcpClient> _sockets = [];
        private readonly ConcurrentBag<Task> _pumps = [];
        private readonly Task _accept;
        private TaskCompletionSource? _pause;
        internal int Port => ((IPEndPoint)_listener.LocalEndpoint).Port;
        internal PauseProxy(int targetPort)
        {
            _listener.Start();
            _accept = AcceptAsync(targetPort);
        }
        internal void Pause() => Volatile.Write(ref _pause, new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously));
        internal void Resume() => Interlocked.Exchange(ref _pause, null)?.TrySetResult();
        private async Task AcceptAsync(int port)
        {
            try
            {
                while (!_stop.IsCancellationRequested)
                {
                    var incoming = await _listener.AcceptTcpClientAsync(_stop.Token);
                    _sockets.Add(incoming);
                    var outgoing = new TcpClient(); _sockets.Add(outgoing);
                    await outgoing.ConnectAsync(IPAddress.Loopback, port, _stop.Token);
                    _pumps.Add(PumpAsync(incoming.GetStream(), outgoing.GetStream()));
                    _pumps.Add(PumpAsync(outgoing.GetStream(), incoming.GetStream()));
                }
            }
            catch (OperationCanceledException) when (_stop.IsCancellationRequested) { }
        }
        private async Task PumpAsync(NetworkStream source, NetworkStream destination)
        {
            var buffer = new byte[16 * 1024];
            try
            {
                while (true)
                {
                    var length = await source.ReadAsync(buffer, _stop.Token);
                    if (length == 0) return;
                    if (Volatile.Read(ref _pause) is { } paused) await paused.Task.WaitAsync(_stop.Token);
                    await destination.WriteAsync(buffer.AsMemory(0, length), _stop.Token);
                }
            }
            catch (OperationCanceledException) when (_stop.IsCancellationRequested) { }
            catch (IOException) when (_stop.IsCancellationRequested) { }
        }
        public async ValueTask DisposeAsync()
        {
            Resume(); _stop.Cancel(); _listener.Stop();
            foreach (var socket in _sockets) socket.Dispose();
            try { await _accept; await Task.WhenAll(_pumps); }
            finally { _stop.Dispose(); }
        }
    }
}