test(opcua): add fixed candidate saturation diagnosis with observer barriers

scripts/validation/opcua-lifecycle-memory/SATURATION.md

@@ -0,0 +1,114 @@
+# Candidate saturation experiment — review before launch
+
+This is a prospective diagnostic protocol, not an acceptance run or a replacement
+for the failed cold server soak. Production code, the cycle-3 resource baseline,
+and the +64 MiB managed / +128 MiB RSS / +32 descriptor caps remain unchanged.
+Only `OTOPCUA_SDK2_SOAK_MODE=saturation` selects this behavior; the existing
+`diagnostic`, `profile`, and `full` modes keep their existing stopping rules.
+
+The fixed canonical phase performs exactly 384 lifecycle cycles, with the
+existing active-resource restart after every fourth cycle: 96 completed restarts.
+It then performs exactly one settling cycle, without another restart. The phase,
+including observer handshakes, must finish within 1,800 seconds. Initial fixture
+startup has a separate 120-second bound. There is no early completion, extension,
+RSS-selected endpoint, budget reset, or two-hour/full-pass flag. Cleanup remains
+bounded independently of the phase deadline.
+
+All canonical operations remain in use: encrypted opc.tcp sessions, namespace URI
+binding, scalar/typed-array read/write, subscriptions, browse abandonment, history
+continuation churn/release/cross-session/close, surgical changes/rebuilds, key
+revocation, paused network reads/recovery, and same-PKI server restart. All
+ownership, continuation, PKI-descriptor, store and node inventory checks remain
+active. Cold budget rejection is latched at its original observation and rethrown,
+even if diagnostic readiness later succeeds. A rejected readiness receipt does
+not become a passing test or a new resource baseline.
+
+## Observer boundaries
+
+A fresh task-owned control directory and UUID bind every request/receipt to this
+run and target PID. Files are published atomically; existing, malformed, foreign,
+or incomplete receipts fail closed. The observer is a separate diagnostic process
+using public `DiagnosticsClient` / EventPipe APIs. Product dependencies are
+unchanged.
+
+One continuous trace covers both fixed windows, including their boundary gap:
+
+| Marker | Before/after operation | Completed restarts |
+|---|---|---:|
+| window1-start | Before cycle 321 | 80 |
+| window1-end | After cycle 352 and its restart | 88 |
+| window2-start | Before cycle 353 | 88 |
+| window2-end | After cycle 384 and its restart | 96 |
+
+The target waits for the observer's active receipt and enabled marker provider
+before its first window operation. The parser requires exactly these four
+markers, the correct UUID and cycle/restart values, and strictly ordered
+timestamps. A continuous veto also covers events outside the marker ranges
+within the attached interval. Both windows and the entire attached interval
+require zero `Opc.Ua.OpcUaExtensions` JIT starts and method-load events, zero new
+`Opc.Ua.*` assembly-load events, and zero lost events. Other JIT counts and loaded
+native method-byte sums are reported for review; they do not establish that all
+runtime paths are warmed. Method-size sums do not equal resident memory.
+
+Runtime capture is Verbose `Microsoft-Windows-DotNETRuntime`, keywords `0x18`
+(Loader + JIT), plus Informational `OtOpcUa-Sdk2-Saturation` markers, no rundown,
+and a prospectively fixed **16 MiB** circular-buffer capacity. This is a new
+experiment and does not reinterpret the earlier 256 MiB-capacity traces.
+
+After restart 96, the target requests stop and waits. The observer must stop its
+session, drain the event stream, close it, parse the entire trace, and atomically
+publish a PID/run-bound drained receipt. Its Stop/copy/dispose/parse stages have
+10/10/5/10-second bounds. Target receipt waits are bounded at 45 seconds and remain
+inside the 1,800-second phase deadline. Observer faults propagate immediately.
+The target also verifies its marker provider is disabled before the settling
+cycle. Missing markers, no events, event loss, or malformed stream cannot prove
+readiness. A parser failure produces a failed observer receipt, not empty counts.
+
+The `saturation-result` separates readiness, cold budget rejection, fixed operation
+completion and verified shutdown ownership. It always records `acceptance=false`.
+It is emitted before rethrowing the original cold assertion. The launcher retains
+nonzero test/container exits and records diagnostic readiness separately.
+
+## Prepare immutable Linux artifacts
+
+Commit the reviewed source first. Preparation refuses a dirty checkout and
+publishes fresh, separate app/observer artifacts for `linux-arm64`, self-contained
+.NET **10.0.14**, without trimming or single-file publishing. It never starts the
+experiment. The bundle records source revision, source hashes, exact publish
+command, fixed protocol, image digest and SHA-256 of every app/observer/tool file.
+
+```bash
+python3 scripts/validation/opcua-lifecycle-memory/prepare_saturation.py \
+  --output /tmp/opcua-saturation-reviewed \
+  --libraries "$diagnosticToolLibraries" \
+  --image sha256:b04611ee9e1b7d36797012ed921962c57189cb4dbb3dff12afb8e158110f293b
+```
+
+`diagnosticToolLibraries` is the installed dotnet-trace 9.0.661903
+`tools/net8.0/any` directory. No product package reference is added. Artifact
+preparation and review are prerequisites; root and Astra must review the actual
+code, control tests and receipt before launch. The coordinator owns the long
+execution.
+
+```bash
+python3 /tmp/opcua-saturation-reviewed/tools/launch_saturation.py \
+  --bundle /tmp/opcua-saturation-reviewed --receipt-sha256 "$reviewedReceiptHash" \
+  --output /tmp/opcua-saturation-run --verify-only
+# After root/Astra review, root runs the same command without --verify-only.
+```
+
+The launcher rechecks every immutable file before and after the run. It creates a
+fresh offline task-owned container, writable output/PKI directories, and read-only
+app/observer/tool mounts. Descriptor controls run first in a separate process.
+TRX guards require two descriptor tests and one canonical test actually executed,
+with no skipped results. The launcher verifies the actual target and observer
+mapped their own self-contained `libcoreclr.so`; the older SDK-image controller
+runtime is not the target runtime. A 2,040-second outer timeout includes startup,
+phase and bounded cleanup; it does not extend the 1,800-second phase.
+
+No long saturation run has been launched during implementation. Existing cold
+failures remain untouched. No original-SDK full-workload adapter, warmed
+acceptance baseline, additional two-hour server run, or production correction is
+part of this change. The separately demonstrated Mac native certificate-related
+inventory growth remains unresolved even if generated-factory compilation
+saturates.

scripts/validation/opcua-lifecycle-memory/SaturationObserver.cs

@@ -0,0 +1,127 @@
+// Diagnostic-only observer. The target remains the unchanged canonical lifecycle test process.
+using System.Diagnostics.Tracing;
+using System.Text.Json;
+using Microsoft.Diagnostics.NETCore.Client;
+using Microsoft.Diagnostics.Tracing;
+
+const string protocol = "sdk2-saturation-v1";
+const string markerProvider = "OtOpcUa-Sdk2-Saturation";
+var control = Path.GetFullPath(args[0]);
+var runId = args[1];
+if (!Guid.TryParseExact(runId, "D", out _)) throw new InvalidDataException("Malformed run identity");
+using var lifetime = new CancellationTokenSource(TimeSpan.FromSeconds(1980));
+var ct = lifetime.Token;
+try
+{
+    var request = await WaitJson("attach-request.json", ct);
+    using (request)
+    {
+        Identity(request.RootElement);
+        Require(request.RootElement.GetProperty("firstCycle").GetInt32() == 321
+            && request.RootElement.GetProperty("completedRestarts").GetInt32() == 80, "Invalid attach boundary");
+        var targetPid = request.RootElement.GetProperty("targetPid").GetInt32();
+        Require(targetPid > 0 && targetPid != Environment.ProcessId, "Invalid target process");
+        var providers = new[]
+        {
+            new EventPipeProvider("Microsoft-Windows-DotNETRuntime", EventLevel.Verbose, 0x18),
+            new EventPipeProvider(markerProvider, EventLevel.Informational)
+        };
+        var client = new DiagnosticsClient(targetPid);
+        var session = await Task.Run(() => client.StartEventPipeSession(providers, requestRundown: false, circularBufferMB: 16), ct)
+            .WaitAsync(TimeSpan.FromSeconds(10), ct);
+        var trace = Path.Combine(control, "readiness.nettrace");
+        await using (var output = new FileStream(trace, FileMode.CreateNew, FileAccess.Write, FileShare.Read))
+        {
+        var copy = session.EventStream.CopyToAsync(output, ct);
+        await Publish("active.json", new { protocol, runId, targetPid, observerPid = Environment.ProcessId, active = true });
+        using var stop = await WaitJson("detach-request.json", ct);
+        Identity(stop.RootElement);
+        Require(stop.RootElement.GetProperty("targetPid").GetInt32() == targetPid
+            && stop.RootElement.GetProperty("lastCycle").GetInt32() == 384
+            && stop.RootElement.GetProperty("completedRestarts").GetInt32() == 96, "Invalid detach boundary");
+        await Task.Run(session.Stop, ct).WaitAsync(TimeSpan.FromSeconds(10), ct);
+        await copy.WaitAsync(TimeSpan.FromSeconds(10), ct);
+        await output.FlushAsync(ct);
+        }
+        await Task.Run(session.Dispose, ct).WaitAsync(TimeSpan.FromSeconds(5), ct);
+        var summary = await Task.Run(() => Parse(trace, runId), ct).WaitAsync(TimeSpan.FromSeconds(10), ct);
+        // Published only after Stop, stream drain, disposal and complete parsing. Never just a file-exists ACK.
+        await Publish("drained.json", new
+        {
+            protocol, runId, targetPid, observerPid = Environment.ProcessId, active = false, drained = true,
+            summary.markersValid, summary.eventsLost, summary.sdkFactoryJit, summary.sdkFactoryLoads,
+            summary.sdkAssemblyLoads, summary.windows, summary.otherJit, summary.loadedNativeBytes,
+            configuration = new { runtimeProvider = "Microsoft-Windows-DotNETRuntime", level = "Verbose", keywords = "0x18",
+                markerProvider, circularBufferMiB = 16, requestRundown = false },
+            acceptance = false
+        });
+    }
+}
+catch (Exception error)
+{
+    await Publish("observer-failed.json", new { protocol, runId, exceptionType = error.GetType().FullName });
+    // An IPC/Stop worker can still be blocked. The parent also bounds and reaps this separate process.
+    Environment.Exit(1);
+}
+
+void Identity(JsonElement value) => Require(value.GetProperty("protocol").GetString() == protocol
+    && value.GetProperty("runId").GetString() == runId, "Mismatched observer identity");
+static void Require(bool condition, string message) { if (!condition) throw new InvalidDataException(message); }
+async Task<JsonDocument> WaitJson(string name, CancellationToken token)
+{
+    var path = Path.Combine(control, name);
+    while (!File.Exists(path)) await Task.Delay(25, token);
+    return JsonDocument.Parse(await File.ReadAllTextAsync(path, token));
+}
+async Task Publish(string name, object value)
+{
+    var temporary = Path.Combine(control, name + ".tmp");
+    await using (var stream = new FileStream(temporary, FileMode.CreateNew, FileAccess.Write))
+        await JsonSerializer.SerializeAsync(stream, value);
+    File.Move(temporary, Path.Combine(control, name));
+}
+static TraceResult Parse(string path, string expectedRun)
+{
+    var markers = new List<Marker>();
+    var events = new List<Observed>();
+    using var source = new EventPipeEventSource(path);
+    source.Clr.MethodJittingStarted += data => events.Add(new(data.TimeStampRelativeMSec,
+        data.MethodNamespace == "Opc.Ua.OpcUaExtensions" ? "factory-jit" : "other-jit", 0));
+    source.Clr.MethodLoadVerbose += data => events.Add(new(data.TimeStampRelativeMSec,
+        data.MethodNamespace == "Opc.Ua.OpcUaExtensions" ? "factory-load" : "other-load", data.MethodSize));
+    source.Clr.LoaderAssemblyLoad += data =>
+    {
+        if (data.FullyQualifiedAssemblyName.StartsWith("Opc.Ua.", StringComparison.Ordinal))
+            events.Add(new(data.TimeStampRelativeMSec, "sdk-assembly", 0));
+    };
+    source.Dynamic.All += data =>
+    {
+        if (data.ProviderName == "OtOpcUa-Sdk2-Saturation" && data.EventName == "Boundary")
+            markers.Add(new(data.TimeStampRelativeMSec, (string)data.PayloadByName("runId"),
+                (string)data.PayloadByName("name"), (int)data.PayloadByName("cycle"), (int)data.PayloadByName("restarts")));
+    };
+    source.Process();
+    (string Name, int Cycle, int Restarts)[] expected =
+    [ ("window1-start", 321, 80), ("window1-end", 352, 88),
+      ("window2-start", 353, 88), ("window2-end", 384, 96) ];
+    var valid = markers.Count == 4 && markers.Select((marker, index) =>
+        marker.Run == expectedRun && (marker.Name, marker.Cycle, marker.Restarts) == expected[index]).All(x => x)
+        && markers.Zip(markers.Skip(1)).All(pair => pair.First.Ms < pair.Second.Ms);
+    var windows = Enumerable.Range(0, 2).Select(index =>
+    {
+        var selected = valid ? events.Where(x => x.Ms >= markers[index * 2].Ms && x.Ms <= markers[index * 2 + 1].Ms).ToArray() : [];
+        return new Window($"window{index + 1}", 321 + index * 32, 352 + index * 32, 80 + index * 8, 88 + index * 8,
+            selected.Count(x => x.Kind == "factory-jit"), selected.Count(x => x.Kind == "factory-load"),
+            selected.Count(x => x.Kind == "sdk-assembly"));
+    }).ToArray();
+    // Totals include the whole attached interval, including marker gaps, startup and stopping the collector.
+    return new TraceResult(valid, source.EventsLost, events.Count(x => x.Kind == "factory-jit"),
+        events.Count(x => x.Kind == "factory-load"), events.Count(x => x.Kind == "sdk-assembly"), windows,
+        events.Count(x => x.Kind == "other-jit"), events.Sum(x => x.Bytes));
+}
+record Marker(double Ms, string Run, string Name, int Cycle, int Restarts);
+record Observed(double Ms, string Kind, long Bytes);
+record Window(string name, int firstCycle, int lastCycle, int startRestarts, int endRestarts,
+    int sdkFactoryJit, int sdkFactoryLoads, int sdkAssemblyLoads);
+record TraceResult(bool markersValid, int eventsLost, int sdkFactoryJit, int sdkFactoryLoads,
+    int sdkAssemblyLoads, Window[] windows, int otherJit, long loadedNativeBytes);

scripts/validation/opcua-lifecycle-memory/build_saturation_observer.py

@@ -0,0 +1,46 @@
+#!/usr/bin/env python3
+"""Build the opt-in saturation observer without changing product dependencies."""
+import argparse
+import json
+from pathlib import Path
+import subprocess
+import xml.etree.ElementTree as ET
+
+
+def build(output, libraries, linux):
+    output = output.resolve()
+    output.mkdir(parents=True, exist_ok=False)
+    project = ET.Element("Project", Sdk="Microsoft.NET.Sdk")
+    properties = ET.SubElement(project, "PropertyGroup")
+    for key, value in {"OutputType": "Exe", "TargetFramework": "net10.0", "ImplicitUsings": "enable",
+                       "TreatWarningsAsErrors": "true", "EnableDefaultCompileItems": "false"}.items():
+        ET.SubElement(properties, key).text = value
+    items = ET.SubElement(project, "ItemGroup")
+    ET.SubElement(items, "FrameworkReference", Include="Microsoft.AspNetCore.App")
+    ET.SubElement(items, "Compile", Include=str(Path(__file__).resolve().parent / "SaturationObserver.cs"))
+    for name in ["Microsoft.Diagnostics.NETCore.Client", "Microsoft.Diagnostics.Tracing.TraceEvent", "Microsoft.Diagnostics.FastSerialization"]:
+        path = (libraries / (name + ".dll")).resolve(strict=True)
+        reference = ET.SubElement(items, "Reference", Include=name)
+        ET.SubElement(reference, "HintPath").text = str(path)
+    project_path = output / "observer.csproj"
+    ET.indent(project)
+    ET.ElementTree(project).write(project_path, encoding="unicode")
+    command = ["dotnet", "publish" if linux else "build", str(project_path), "-c", "Release", "-m:2", "--nologo"]
+    if linux:
+        command += ["-r", "linux-arm64", "--self-contained", "true", "-p:RuntimeFrameworkVersion=10.0.14",
+                    "-p:PublishSingleFile=false", "-p:PublishTrimmed=false", "-p:UseAppHost=true", "-o", str(output / "published")]
+    with (output / "build.log").open("w") as log:
+        result = subprocess.run(command, stdout=log, stderr=subprocess.STDOUT)
+    (output / "build-command.json").write_text(json.dumps(command, indent=2) + "\n")
+    if result.returncode:
+        raise subprocess.CalledProcessError(result.returncode, command)
+    return output / ("published" if linux else "bin/Release/net10.0")
+
+
+if __name__ == "__main__":
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--output", required=True, type=Path)
+    parser.add_argument("--libraries", required=True, type=Path)
+    parser.add_argument("--linux-self-contained", action="store_true")
+    args = parser.parse_args()
+    print(build(args.output, args.libraries, args.linux_self_contained))

scripts/validation/opcua-lifecycle-memory/container_saturation.sh

@@ -0,0 +1,63 @@
+#!/usr/bin/env bash
+# Diagnostic only. The published xUnit apphost and observer are self-contained .NET 10.0.14.
+set -euo pipefail
+mkdir -p /task/tmp /task/data /task/results /task/control
+export TMPDIR=/task/tmp XDG_DATA_HOME=/task/data
+export OTOPCUA_SDK2_SOAK=1 OTOPCUA_SDK2_SOAK_MODE=saturation OTOPCUA_SDK2_SOAK_SECONDS=1800
+export OTOPCUA_SDK2_SOAK_OUTPUT=/task/resources.jsonl
+export OTOPCUA_SDK2_SATURATION_CONTROL=/task/control
+export OPCUA_CERTIFICATE_LEAK_TRACKING=1
+export DOTNET_CLI_TELEMETRY_OPTOUT=1 DOTNET_NOLOGO=1
+[[ $(uname -s) == Linux && $(uname -m) == aarch64 && -d /proc/self/fd ]]
+dotnet --info > /task/controller-runtime.txt
+vstest_pid= observer_pid=
+cleanup() {
+    for child in "$vstest_pid" "$observer_pid"; do
+        if [[ -n "$child" ]] && kill -0 "$child" 2>/dev/null; then kill "$child" 2>/dev/null || true; fi
+    done
+}
+trap cleanup EXIT
+dotnet vstest /app/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests.dll \
+  --TestCaseFilter:'FullyQualifiedName~Sdk2LifecycleSoakTests.Descriptor_probe_' \
+  --ResultsDirectory:/task/results '--Logger:trx;LogFileName=descriptor-controls.trx' > /task/descriptor-controls.log 2>&1
+/observer/observer /task/control "$OTOPCUA_SDK2_SATURATION_RUN_ID" > /task/observer.log 2>&1 &
+observer_pid=$!
+dotnet vstest /app/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests.dll \
+  --TestCaseFilter:'FullyQualifiedName=ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests.Sdk2LifecycleSoakTests.Repeated_lifecycles_return_resources_to_a_steady_baseline' \
+  --ResultsDirectory:/task/results '--Logger:trx;LogFileName=saturation.trx' > /task/test.log 2>&1 &
+vstest_pid=$!
+observed=0
+for attempt in $(seq 1 1200); do
+    if [[ -s /task/resources.jsonl ]]; then
+        target_pid=$(head -n 1 /task/resources.jsonl | sed -n 's/.*"processId":\([0-9]*\),.*/\1/p')
+        [[ -n "$target_pid" ]]
+        target_exe=$(readlink "/proc/$target_pid/exe")
+        [[ "$target_exe" == /app/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests ]]
+        printf 'targetPid=%s\nactualExecutable=%s\n' "$target_pid" "$target_exe" > /task/target-runtime.txt
+        awk '$6 ~ /\/(libcoreclr.so|libhostfxr.so|libhostpolicy.so|System.Private.CoreLib.dll)$/ {print $6}' "/proc/$target_pid/maps" | sort -u >> /task/target-runtime.txt
+        grep -Fx /app/libcoreclr.so /task/target-runtime.txt > /dev/null
+        [[ $(readlink "/proc/$observer_pid/exe") == /observer/observer ]]
+        awk '$6 ~ /\/libcoreclr.so$/ {print $6}' "/proc/$observer_pid/maps" | sort -u > /task/observer-runtime.txt
+        grep -Fx /observer/libcoreclr.so /task/observer-runtime.txt > /dev/null
+        observed=1
+        break
+    fi
+    kill -0 "$vstest_pid" 2>/dev/null || break
+    sleep 0.1
+done
+[[ "$observed" == 1 ]]
+set +e
+wait "$vstest_pid"
+test_status=$?
+printf '%s\n' "$test_status" > /task/test-exit-status.txt
+for attempt in $(seq 1 400); do
+    kill -0 "$observer_pid" 2>/dev/null || break
+    sleep 0.1
+done
+if kill -0 "$observer_pid" 2>/dev/null; then kill "$observer_pid"; fi
+wait "$observer_pid"
+observer_status=$?
+printf '%s\n' "$observer_status" > /task/observer-exit-status.txt
+# Preserve a cold-cap failure even when diagnostic readiness succeeds.
+if [[ "$test_status" != 0 ]]; then exit "$test_status"; fi
+exit "$observer_status"

scripts/validation/opcua-lifecycle-memory/launch_saturation.py

@@ -0,0 +1,114 @@
+#!/usr/bin/env python3
+"""Launch only a previously reviewed immutable candidate saturation bundle. Never an acceptance run."""
+import argparse
+import hashlib
+import json
+from pathlib import Path
+import subprocess
+import time
+import uuid
+import xml.etree.ElementTree as ET
+
+
+def digest(path):
+    return hashlib.sha256(path.read_bytes()).hexdigest()
+
+
+def require(condition, message):
+    if not condition:
+        raise ValueError(message)
+
+
+def verify(root, expected):
+    receipt_path = root / "receipt.json"
+    if digest(receipt_path) != expected:
+        raise ValueError("Receipt hash differs from the reviewed artifact")
+    receipt = json.loads(receipt_path.read_text())
+    for name, value in receipt["files"].items():
+        if digest(root / name) != value:
+            raise ValueError("Immutable bundle file changed: " + name)
+    require(receipt["protocol"] == "sdk2-saturation-v1" and receipt["acceptance"] is False, "Invalid diagnostic protocol")
+    require((receipt["canonicalCycles"], receipt["completedRestarts"], receipt["settlingCycles"], receipt["maximumPhaseSeconds"]) == (384, 96, 1, 1800), "Changed prospective limits")
+    return receipt
+
+
+def trx(path, minimum):
+    document = ET.parse(path)
+    counters = document.find(".//{*}Counters")
+    if counters is None:
+        raise ValueError("Missing TRX counters")
+    executed = int(counters.attrib.get("executed", "0"))
+    skipped = max(int(counters.attrib.get("notExecuted", "0")),
+                  sum(result.get("outcome") == "NotExecuted" for result in document.findall("./{*}Results/{*}UnitTestResult")))
+    if executed < minimum or skipped:
+        raise ValueError("Required diagnostic tests were skipped or not executed")
+    return {"executed": executed, "skipped": skipped}
+
+
+def main():
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--bundle", type=Path, required=True)
+    parser.add_argument("--receipt-sha256", required=True, help="Hash reviewed by root and Astra before launch")
+    parser.add_argument("--output", type=Path, required=True)
+    parser.add_argument("--verify-only", action="store_true")
+    options = parser.parse_args()
+    root = options.bundle.resolve(strict=True)
+    receipt = verify(root, options.receipt_sha256)
+    if options.verify_only:
+        print("Immutable bundle verified; no experiment launched")
+        return 0
+    output = options.output.resolve()
+    output.mkdir(parents=True, exist_ok=False)
+    run_id = str(uuid.uuid4())
+    container = "opcua-saturation-" + uuid.uuid4().hex[:12]
+    command = ["docker", "run", "--rm", "--name", container, "--network", "none",
+               "-e", "OTOPCUA_SOAK_BUILD_SHA=" + receipt["sourceRevision"],
+               "-e", "OTOPCUA_SDK2_SATURATION_RUN_ID=" + run_id,
+               "-v", str(root / "app") + ":/app:ro", "-v", str(root / "observer") + ":/observer:ro",
+               "-v", str(root / "tools") + ":/tools:ro", "-v", str(output) + ":/task",
+               receipt["image"], "bash", "/tools/container_saturation.sh"]
+    (output / "launch.json").write_text(json.dumps({"command": command, "receiptSha256": options.receipt_sha256,
+                                                   "sourceRevision": receipt["sourceRevision"], "acceptance": False}, indent=2) + "\n")
+    start = time.monotonic()
+    try:
+        with (output / "container.log").open("w") as log:
+            process = subprocess.run(command, stdout=log, stderr=subprocess.STDOUT, timeout=2040)
+        status = process.returncode
+    except subprocess.TimeoutExpired:
+        status = 124
+    finally:
+        subprocess.run(["docker", "stop", "--time", "10", container], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, timeout=20)
+    (output / "container-exit-status.txt").write_text(str(status) + "\n")
+    results = {"acceptance": False, "containerExitStatus": status, "wallSeconds": time.monotonic() - start,
+               "sourceRevision": receipt["sourceRevision"], "receiptSha256": options.receipt_sha256}
+    try:
+        results["descriptorTests"] = trx(output / "results/descriptor-controls.trx", 2)
+        results["experimentTests"] = trx(output / "results/saturation.trx", 1)
+        rows = [json.loads(line) for line in (output / "resources.jsonl").read_text().splitlines()]
+        header = rows[0]
+        require(header["mode"] == "saturation" and header["requestedSeconds"] == 1800 and header["build"] == receipt["sourceRevision"], "Wrong executed source or mode")
+        require(not any(row["kind"] == "complete" for row in rows), "Saturation must never emit an ordinary completion result")
+        outcome = [row for row in rows if row["kind"] == "saturation-result"]
+        require(len(outcome) == 1, "Missing unique diagnostic result")
+        outcome = outcome[0]
+        require((outcome["canonicalCycles"], outcome["settlingCycles"], outcome["restarts"]) == (384, 1, 96), "Incomplete fixed work")
+        require(outcome["phaseSeconds"] <= 1800 and outcome["shutdownOwnershipVerified"], "Cutoff or cleanup proof failed")
+        require(outcome["acceptance"] is False, "This is not an acceptance run")
+        results["diagnosticOutcome"] = outcome
+        observer_status = int((output / "observer-exit-status.txt").read_text())
+        test_status = int((output / "test-exit-status.txt").read_text())
+        require(observer_status == 0, "Observer failed")
+        require((test_status != 0) == (outcome["coldBudgetRejected"] or not outcome["readiness"]), "Cold-cap exit was not preserved")
+        results["protocolCompleted"] = True
+    except Exception as error:
+        results["protocolCompleted"] = False
+        results["validationErrorType"] = type(error).__name__
+        status = status or 1
+    verify(root, options.receipt_sha256)
+    (output / "run-receipt.json").write_text(json.dumps(results, indent=2) + "\n")
+    print(json.dumps(results, indent=2))
+    return status
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

scripts/validation/opcua-lifecycle-memory/prepare_saturation.py

@@ -0,0 +1,63 @@
+#!/usr/bin/env python3
+"""Prepare a fresh immutable candidate bundle. This command never launches the experiment."""
+import argparse
+import hashlib
+import json
+from pathlib import Path
+import shutil
+import subprocess
+from build_saturation_observer import build
+
+
+def sha(path):
+    return hashlib.sha256(path.read_bytes()).hexdigest()
+
+
+def main():
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--output", type=Path, required=True)
+    parser.add_argument("--libraries", type=Path, required=True)
+    parser.add_argument("--image", required=True)
+    options = parser.parse_args()
+    source = Path(__file__).resolve().parent
+    repo = source.parents[2]
+    revision = subprocess.check_output(["git", "rev-parse", "HEAD"], cwd=repo, text=True).strip()
+    if subprocess.check_output(["git", "status", "--porcelain"], cwd=repo, text=True).strip():
+        raise RuntimeError("Commit and review the exact source before preparing immutable artifacts")
+    if not options.image.startswith("sha256:"):
+        raise ValueError("Use an immutable Linux .NET SDK image digest")
+    root = options.output.resolve()
+    root.mkdir(parents=True, exist_ok=False)
+    project = repo / "tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests.csproj"
+    command = ["dotnet", "publish", str(project), "-c", "Release", "-r", "linux-arm64", "--self-contained", "true",
+               "-p:RuntimeFrameworkVersion=10.0.14", "-p:PublishSingleFile=false", "-p:PublishTrimmed=false", "-p:UseAppHost=true",
+               "--artifacts-path", str(root / "build"), "-o", str(root / "app"), "-m:2", "--nologo"]
+    with (root / "publish.log").open("w") as log:
+        subprocess.run(command, cwd=repo, stdout=log, stderr=subprocess.STDOUT, check=True)
+    observer = build(root / "observer-build", options.libraries, True)
+    shutil.copytree(observer, root / "observer")
+    tools = root / "tools"
+    tools.mkdir()
+    for name in ["launch_saturation.py", "container_saturation.sh", "SATURATION.md"]:
+        shutil.copy2(source / name, tools / name)
+    manifest = {str(path.relative_to(root)): sha(path) for directory in ["app", "observer", "tools"]
+                for path in sorted((root / directory).rglob("*")) if path.is_file()}
+    runtime = json.loads((root / "app/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests.runtimeconfig.json").read_text())
+    included = runtime["runtimeOptions"]["includedFrameworks"]
+    if not included or any(value["version"] != "10.0.14" for value in included):
+        raise RuntimeError("Published app is not the prospectively fixed self-contained runtime")
+    receipt = {"protocol": "sdk2-saturation-v1", "sourceRevision": revision, "image": options.image,
+               "rid": "linux-arm64", "runtimeVersion": "10.0.14", "selfContained": True,
+               "canonicalCycles": 384, "completedRestarts": 96, "settlingCycles": 1, "maximumPhaseSeconds": 1800,
+               "windows": [[321, 352], [353, 384]], "observerBufferMiB": 16, "runtimeKeywords": "0x18", "rundown": False,
+               "acceptance": False, "publishCommand": command,
+               "testSourceSha256": sha(project.parent / "Sdk2LifecycleSoakTests.cs"),
+               "observerSourceSha256": sha(source / "SaturationObserver.cs"), "files": manifest}
+    receipt_path = root / "receipt.json"
+    receipt_path.write_text(json.dumps(receipt, indent=2) + "\n")
+    print(json.dumps({"bundle": str(root), "receiptSha256": sha(receipt_path), "sourceRevision": revision,
+                      "launched": False}, indent=2))
+
+
+if __name__ == "__main__":
+    main()

scripts/validation/opcua-lifecycle-memory/saturation_tool_controls.py

@@ -0,0 +1,37 @@
+#!/usr/bin/env python3
+"""Small negative controls for artifact tampering and VSTest's misleading skipped counter."""
+import hashlib
+import json
+from pathlib import Path
+import tempfile
+from launch_saturation import trx, verify
+
+with tempfile.TemporaryDirectory(prefix="opcua-saturation-tools-") as temporary:
+    root = Path(temporary)
+    artifact = root / "immutable"
+    artifact.write_text("control")
+    receipt = {"protocol": "sdk2-saturation-v1", "acceptance": False, "canonicalCycles": 384,
+               "completedRestarts": 96, "settlingCycles": 1, "maximumPhaseSeconds": 1800,
+               "files": {"immutable": hashlib.sha256(artifact.read_bytes()).hexdigest()}}
+    path = root / "receipt.json"
+    path.write_text(json.dumps(receipt))
+    digest = hashlib.sha256(path.read_bytes()).hexdigest()
+    verify(root, digest)
+    artifact.write_text("tampered")
+    try:
+        verify(root, digest)
+    except ValueError:
+        pass
+    else:
+        raise RuntimeError("Changed artifact was accepted")
+    xml = root / "skipped.trx"
+    xml.write_text('<TestRun xmlns="http://microsoft.com/schemas/VisualStudio/TeamTest/2010"><Results>'
+                   '<UnitTestResult outcome="NotExecuted" /></Results><ResultSummary>'
+                   '<Counters executed="1" notExecuted="0" /></ResultSummary></TestRun>')
+    try:
+        trx(xml, 1)
+    except ValueError:
+        pass
+    else:
+        raise RuntimeError("Direct skipped result was hidden by the counter")
+print("Artifact verification, tamper rejection and direct skipped-result controls passed (3/3)")

tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests/Sdk2LifecycleSoakTests.cs

@@ -1,6 +1,7 @@
 using System.Collections;
 using System.Collections.Concurrent;
 using System.Diagnostics;
+using System.Diagnostics.Tracing;
 using System.Globalization;
 using System.Net;
 using System.Net.Sockets;
@@ -39,14 +40,16 @@ public sealed class Sdk2LifecycleSoakTests
     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 diagnostic/full mode.")]
+    [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").ShouldBeTrue("explicit diagnostic, profile or full mode is required");
+        (mode is "diagnostic" or "profile" or "full" or "saturation").ShouldBeTrue("an explicit soak mode is required");
+        var saturation = mode == "saturation" ? SaturationControl.FromEnvironment() : null;
         var seconds = int.Parse(Environment.GetEnvironmentVariable("OTOPCUA_SDK2_SOAK_SECONDS")
-            ?? (mode == "full" ? "7200" : "90"), CultureInfo.InvariantCulture);
+            ?? (mode == "full" ? "7200" : saturation is not null ? "1800" : "90"), CultureInfo.InvariantCulture);
         if (mode == "full") seconds.ShouldBeGreaterThanOrEqualTo(7200);
+        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")
@@ -55,7 +58,7 @@ public sealed class Sdk2LifecycleSoakTests
         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(seconds + 120));
+        run.CancelAfter(TimeSpan.FromSeconds(saturation is null ? seconds + 120 : 120));
         var ct = run.Token;
         var watch = Stopwatch.StartNew();
         var clientRoot = Path.Combine(Path.GetTempPath(), $"otua-soak-client-{Guid.NewGuid():N}");
@@ -87,6 +90,7 @@ public sealed class Sdk2LifecycleSoakTests
             Configure(rig);
             // The required observation period excludes initial certificate creation and server startup.
             watch.Restart();
+            if (saturation is not null) run.CancelAfter(TimeSpan.FromSeconds(1800));
             await LogAsync(evidence, new
             {
                 kind = "start", utc = DateTimeOffset.UtcNow, processId = Environment.ProcessId,
@@ -97,8 +101,19 @@ public sealed class Sdk2LifecycleSoakTests
                 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",
             });
-            while (watch.Elapsed < TimeSpan.FromSeconds(seconds))
+            while (saturation is not null ? cycles < 385 : 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();
                 await ExerciseAsync(rig, config, ++cycles, evidence, ct);
                 await EventuallyAsync(() => PublicCounts(rig) == (0, 0, 0), ct);
@@ -123,7 +138,7 @@ public sealed class Sdk2LifecycleSoakTests
                             sample.ManagedBytes.ShouldBeLessThanOrEqualTo(baseline.ManagedBytes + (64L << 20));
                             sample.RssBytes.ShouldBeLessThanOrEqualTo(baseline.RssBytes + (128L << 20));
                         }
-                        catch (ShouldAssertException error) when (mode == "profile")
+                        catch (ShouldAssertException error) when (mode == "profile" || saturation is not null)
                         {
                             // Investigative sampling may continue after rejection, but can never turn it into a pass.
                             firstRejectedBudget = ExceptionDispatchInfo.Capture(error);
@@ -151,22 +166,47 @@ public sealed class Sdk2LifecycleSoakTests
                     restarts++;
                     await LogAsync(evidence, new { kind = "restart", cycles, restarts, seconds = restart.Elapsed.TotalSeconds });
                 }
+                if (saturation is not null)
+                {
+                    if (cycles == 352) saturation.Boundary("window1-end", cycles, restarts);
+                    if (cycles == 384) saturation.Boundary("window2-end", cycles, restarts);
+                }
             }
             // 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.
-            firstRejectedBudget?.Throw();
-            cycles.ShouldBeGreaterThanOrEqualTo(4, "diagnostic mode must exercise at least one actual restart");
-            restarts.ShouldBeGreaterThan(0);
-            watch.Elapsed.TotalSeconds.ShouldBeGreaterThanOrEqualTo(seconds);
+            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), ct);
+            await rig.DisposeAsync().AsTask().WaitAsync(TimeSpan.FromSeconds(20), cleanupToken);
             rig = null;
-            await application.DisposeAsync().AsTask().WaitAsync(TimeSpan.FromSeconds(20), ct);
+            await application.DisposeAsync().AsTask().WaitAsync(TimeSpan.FromSeconds(20), cleanupToken);
             application = null;
             finalClientCores.Sum(CachedStores).ShouldBe(0);
             finalServerCores.Sum(CachedStores).ShouldBe(0);
-            (await DescriptorCountsAsync(ct, clientRoot, serverRoot)).Pki.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 });
         }
@@ -174,6 +214,7 @@ public sealed class Sdk2LifecycleSoakTests
         {
             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
@@ -631,6 +672,221 @@ public sealed class Sdk2LifecycleSoakTests
         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) 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);
+            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);
+        }
+    }
+
+    [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) =>