test: verify the mapped CLR inode in late-slope diagnostics

test: add immutable Linux late-slope diagnostic tooling
Merge master SDK1 ownership repair with SDK2 disposal adaptation
2026-11-10 18:02:08 -05:00 · 2026-11-10 17:59:15 -05:00 · 2026-11-10 17:17:00 -05:00 · 2026-11-10 17:16:47 -05:00 · 2026-11-10 17:06:27 -05:00 · 2026-11-10 17:05:22 -05:00
12 changed files with 2114 additions and 20 deletions
@@ -63,5 +63,5 @@ lib/
 # Subagent-driven-development git worktrees
 .worktrees/

-# OPC UA SDK upgrade verification artifacts
+# Local OPC UA validation runs; selected receipts belong under docs/plans/evidence.
 /artifacts/opcua2/
@@ -0,0 +1,349 @@
+{
+  "issue": 807,
+  "scope": "SDK1 host terminal-disposal repair; independent of SDK2 migration release acceptance.",
+  "originalMaster": "52d1373e8129b7f4f0ea2e35f86a8a5d0ed1de36",
+  "redSource": "84bfe6cd20285ccf136998998ef5c051917d8201",
+  "greenSource": "10edc44f94afc42300518f1504091c58217a89a9",
+  "redProductionEqualsOriginalMaster": true,
+  "redOriginalProductionSha256": "346362c8c18b5bcdb90757f654c02c68bad44fc5875abc8b4672f4c7ac49b381",
+  "greenProductionSha256": "c4934b15e695a4110e8cea8fbdef8faabad80aadeca4ccfb007b11400057ff9b",
+  "resolvedSdkPackages": [
+    "OPCFoundation.NetStandard.Opc.Ua.Configuration/1.5.378.182",
+    "OPCFoundation.NetStandard.Opc.Ua.Core/1.5.378.182",
+    "OPCFoundation.NetStandard.Opc.Ua.Security.Certificates/1.5.378.182",
+    "OPCFoundation.NetStandard.Opc.Ua.Server/1.5.378.182",
+    "OPCFoundation.NetStandard.Opc.Ua.Types/1.5.378.182"
+  ],
+  "builds": [
+    {
+      "rawLogPath": "/tmp/opcua-sdk1-terminal-red-build.log",
+      "sha256": "38f5561a0db2a2d26f8462f36e114ced057c189effcd6e87dc5a62a42c06d52e",
+      "warnings": 0,
+      "errors": 0
+    },
+    {
+      "rawLogPath": "/tmp/opcua-sdk1-terminal-green-build.log",
+      "sha256": "5fae318d45d9b5b8b4ba2812ef87869edf6d851e09848154d5f4a51f47ed68d2",
+      "warnings": 0,
+      "errors": 0
+    },
+    {
+      "rawLogPath": "/tmp/opcua-sdk1-terminal-solution-build.log",
+      "sha256": "13317ee89238730216874531fad82e3591a6d0d1b5503c8bd896dcc95c885daf",
+      "warnings": 0,
+      "errors": 0
+    }
+  ],
+  "runs": [
+    {
+      "source": "84bfe6cd20285ccf136998998ef5c051917d8201",
+      "rawDirectory": "artifacts/opcua2/84bfe6cd/sdk1-terminal-red",
+      "command": [
+        "dotnet",
+        "test",
+        "tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests/ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests.csproj",
+        "-c",
+        "Release",
+        "--no-build",
+        "--no-restore",
+        "--filter",
+        "FullyQualifiedName~OpcUaApplicationHostDisposalTests",
+        "--logger",
+        "trx;LogFileName=close.trx",
+        "--results-directory",
+        "artifacts/opcua2/84bfe6cd/sdk1-terminal-red"
+      ],
+      "testExit": 1,
+      "guardPassed": true,
+      "runnerSeconds": 3.8918902919394895,
+      "counters": {
+        "total": "5",
+        "executed": "5",
+        "passed": "0",
+        "failed": "5",
+        "error": "0",
+        "timeout": "0",
+        "aborted": "0",
+        "inconclusive": "0",
+        "passedButRunAborted": "0",
+        "notRunnable": "0",
+        "notExecuted": "0",
+        "disconnected": "0",
+        "warning": "0",
+        "completed": "0",
+        "inProgress": "0",
+        "pending": "0"
+      },
+      "selectedResults": [
+        {
+          "name": "ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests.OpcUaApplicationHostDisposalTests.Host_terminally_disposes_its_booted_server_once",
+          "outcome": "Failed",
+          "error": "Shouldly.ShouldAssertException : server.TerminalDisposals\r\n    should be\r\n1\r\n    but was\r\n0"
+        },
+        {
+          "name": "ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests.OpcUaApplicationHostDisposalTests.Concurrent_disposers_share_completion_and_one_cleanup_owner",
+          "outcome": "Failed",
+          "error": "Shouldly.ShouldAssertException : server.StopCalls\r\n    should be\r\n1\r\n    but was\r\n4"
+        },
+        {
+          "name": "ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests.OpcUaApplicationHostDisposalTests.Disposed_host_rejects_start_without_taking_server_ownership",
+          "outcome": "Failed",
+          "error": "Shouldly.ShouldAssertException : Task `host.StartAsync(server, Ct)`\r\n    should throw\r\nSystem.ObjectDisposedException\r\n    but did not"
+        },
+        {
+          "name": "ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests.OpcUaApplicationHostDisposalTests.Stop_failure_still_terminally_disposes_the_server",
+          "outcome": "Failed",
+          "error": "Shouldly.ShouldAssertException : server.TerminalDisposals\r\n    should be\r\n1\r\n    but was\r\n0"
+        },
+        {
+          "name": "ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests.OpcUaApplicationHostDisposalTests.Partial_startup_failure_does_not_strand_the_server",
+          "outcome": "Failed",
+          "error": "Shouldly.ShouldAssertException : server.TerminalDisposals\r\n    should be\r\n1\r\n    but was\r\n0"
+        }
+      ],
+      "rawHashes": {
+        "close.trx": "ec832633adf184a07f31075ff342bb2c1b9d3f3893c518f5737dc02526f886bb",
+        "run.log": "cdb7c4982d45807de53f70dfc22fad69ccda1db740db05407cef8df8b6b6d904",
+        "receipt.json": "4469c071c0d178446ccfdbdca4bf11ebad43c33feb91d8a8c5f2fa891579f2dc"
+      }
+    },
+    {
+      "source": "10edc44f94afc42300518f1504091c58217a89a9",
+      "rawDirectory": "artifacts/opcua2/10edc44f/sdk1-terminal-green",
+      "command": [
+        "dotnet",
+        "test",
+        "tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests/ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests.csproj",
+        "-c",
+        "Release",
+        "--no-build",
+        "--no-restore",
+        "--filter",
+        "FullyQualifiedName~OpcUaApplicationHostDisposalTests",
+        "--logger",
+        "trx;LogFileName=close.trx",
+        "--results-directory",
+        "artifacts/opcua2/10edc44f/sdk1-terminal-green"
+      ],
+      "testExit": 0,
+      "guardPassed": true,
+      "runnerSeconds": 3.336757709039375,
+      "counters": {
+        "total": "5",
+        "executed": "5",
+        "passed": "5",
+        "failed": "0",
+        "error": "0",
+        "timeout": "0",
+        "aborted": "0",
+        "inconclusive": "0",
+        "passedButRunAborted": "0",
+        "notRunnable": "0",
+        "notExecuted": "0",
+        "disconnected": "0",
+        "warning": "0",
+        "completed": "0",
+        "inProgress": "0",
+        "pending": "0"
+      },
+      "selectedResults": [
+        {
+          "name": "ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests.OpcUaApplicationHostDisposalTests.Partial_startup_failure_does_not_strand_the_server",
+          "outcome": "Passed"
+        },
+        {
+          "name": "ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests.OpcUaApplicationHostDisposalTests.Concurrent_disposers_share_completion_and_one_cleanup_owner",
+          "outcome": "Passed"
+        },
+        {
+          "name": "ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests.OpcUaApplicationHostDisposalTests.Host_terminally_disposes_its_booted_server_once",
+          "outcome": "Passed"
+        },
+        {
+          "name": "ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests.OpcUaApplicationHostDisposalTests.Stop_failure_still_terminally_disposes_the_server",
+          "outcome": "Passed"
+        },
+        {
+          "name": "ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests.OpcUaApplicationHostDisposalTests.Disposed_host_rejects_start_without_taking_server_ownership",
+          "outcome": "Passed"
+        }
+      ],
+      "rawHashes": {
+        "close.trx": "c44ba287659af567fcba1209c147aa5bfb6f2fc27753c0923d08b92372c007b4",
+        "run.log": "7f68723461a5ac4154c247b8bdfcc6d3770a5c1c1a88016051ef492f1684c88e",
+        "receipt.json": "bea02609e11dcc4569a26add9ae5836b101bf107d0346dd34514fc1d114ba525"
+      }
+    },
+    {
+      "source": "10edc44f94afc42300518f1504091c58217a89a9",
+      "rawDirectory": "artifacts/opcua2/10edc44f/sdk1-server-suite",
+      "command": [
+        "dotnet",
+        "test",
+        "tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests/ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests.csproj",
+        "-c",
+        "Release",
+        "--no-build",
+        "--no-restore",
+        "--logger",
+        "trx;LogFileName=close.trx",
+        "--results-directory",
+        "artifacts/opcua2/10edc44f/sdk1-server-suite"
+      ],
+      "testExit": 0,
+      "guardPassed": true,
+      "runnerSeconds": 119.36682745895814,
+      "counters": {
+        "total": "1033",
+        "executed": "1032",
+        "passed": "1032",
+        "failed": "0",
+        "error": "0",
+        "timeout": "0",
+        "aborted": "0",
+        "inconclusive": "0",
+        "passedButRunAborted": "0",
+        "notRunnable": "0",
+        "notExecuted": "0",
+        "disconnected": "0",
+        "warning": "0",
+        "completed": "0",
+        "inProgress": "0",
+        "pending": "0"
+      },
+      "selectedResults": [
+        {
+          "name": "ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests.AddressSpaceComposerDeviceHostTests.Compose_resolves_driver_device_and_device_host",
+          "outcome": "NotExecuted",
+          "error": "v3: equipment no longer binds a driver/device \u2014 Equipment lost DriverInstanceId/DeviceId and Compose no longer takes a devices list, so EquipmentNode's DriverInstanceId/DeviceId/DeviceHost projection is always null. Equipment references raw tags via UnsTagReference. The device-endpoint merge this suite characterized now lives in Runtime.Tests DeploymentArtifactRawPathTests; the host-extraction unit is Commons.Tests DeviceConfigIntentTests."
+        },
+        {
+          "name": "ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests.OpcUaApplicationHostDisposalTests.Host_terminally_disposes_its_booted_server_once",
+          "outcome": "Passed"
+        },
+        {
+          "name": "ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests.OpcUaApplicationHostSecurityTests.BuildSecurityPolicies_empty_input_falls_back_to_none",
+          "outcome": "Passed"
+        },
+        {
+          "name": "ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests.OpcUaApplicationHostTests.StartAsync_creates_application_certificate_in_pki_own",
+          "outcome": "Passed"
+        },
+        {
+          "name": "ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests.OpcUaApplicationHostQuotaTests.Configured_transport_quotas_replace_only_what_is_set",
+          "outcome": "Passed"
+        },
+        {
+          "name": "ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests.OpcUaApplicationHostSecurityTests.BuildSecurityPolicies_default_set_emits_all_three_baseline_profiles",
+          "outcome": "Passed"
+        },
+        {
+          "name": "ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests.OpcUaApplicationHostDisposalTests.Partial_startup_failure_does_not_strand_the_server",
+          "outcome": "Passed"
+        },
+        {
+          "name": "ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests.OpcUaApplicationHostSecurityTests.BuildUserTokenPolicies_withholds_anonymous_when_disallowed",
+          "outcome": "Passed"
+        },
+        {
+          "name": "ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests.OpcUaApplicationHostImpersonationTests.NullOpcUaUserAuthenticator_always_denies",
+          "outcome": "Passed"
+        },
+        {
+          "name": "ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests.OpcUaApplicationHostQuotaTests.Configured_request_quotas_are_applied",
+          "outcome": "Passed"
+        },
+        {
+          "name": "ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests.OpcUaApplicationHostImpersonationTests.HandleImpersonation_anonymous_token_is_rejected_when_disallowed",
+          "outcome": "Passed"
+        },
+        {
+          "name": "ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests.OpcUaApplicationHostImpersonationTests.HandleImpersonation_username_success_sets_identity_and_no_validation_error",
+          "outcome": "Passed"
+        },
+        {
+          "name": "ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests.OpcUaApplicationHostDisposalTests.Disposed_host_rejects_start_without_taking_server_ownership",
+          "outcome": "Passed"
+        },
+        {
+          "name": "ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests.OpcUaApplicationHostServerArrayTests.ServerArray_contains_local_uri_and_configured_peers_after_start",
+          "outcome": "Passed"
+        },
+        {
+          "name": "ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests.OpcUaApplicationHostQuotaTests.Default_request_quotas_are_the_former_hard_coded_values",
+          "outcome": "Passed"
+        },
+        {
+          "name": "ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests.OpcUaApplicationHostSecurityTests.StartAsync_with_only_signandencrypt_omits_None_endpoint",
+          "outcome": "Passed"
+        },
+        {
+          "name": "ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests.OpcUaApplicationHostDisposalTests.Concurrent_disposers_share_completion_and_one_cleanup_owner",
+          "outcome": "Passed"
+        },
+        {
+          "name": "ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests.OpcUaApplicationHostQuotaTests.Unset_transport_quotas_keep_the_sdk_defaults",
+          "outcome": "Passed"
+        },
+        {
+          "name": "ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests.OpcUaApplicationHostDisposalTests.Stop_failure_still_terminally_disposes_the_server",
+          "outcome": "Passed"
+        },
+        {
+          "name": "ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests.OpcUaApplicationHostImpersonationTests.HandleImpersonation_anonymous_token_is_admitted_with_a_role_carrying_identity",
+          "outcome": "Passed"
+        },
+        {
+          "name": "ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests.OpcUaApplicationHostSecurityTests.StartAsync_populates_ServerConfiguration_with_all_enabled_profiles",
+          "outcome": "Passed"
+        },
+        {
+          "name": "ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests.OpcUaApplicationHostTests.StartAsync_reuses_existing_certificate_on_second_boot",
+          "outcome": "Passed"
+        },
+        {
+          "name": "ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests.OpcUaApplicationHostImpersonationTests.HandleImpersonation_username_denial_sets_validation_error_and_no_identity",
+          "outcome": "Passed"
+        },
+        {
+          "name": "ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests.OpcUaApplicationHostQuotaTests.History_read_batch_cap_default_scales_with_the_machine",
+          "outcome": "Passed"
+        },
+        {
+          "name": "ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests.OpcUaApplicationHostSecurityTests.BuildSecurityPolicies_dedupes_repeated_profiles",
+          "outcome": "Passed"
+        },
+        {
+          "name": "ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests.OpcUaApplicationHostImpersonationTests.HandleImpersonation_authenticator_throwing_results_in_rejection",
+          "outcome": "Passed"
+        },
+        {
+          "name": "ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests.OpcUaApplicationHostImpersonationTests.HandleImpersonation_null_username_treated_as_empty_string",
+          "outcome": "Passed"
+        },
+        {
+          "name": "ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests.OpcUaApplicationHostImpersonationTests.HandleImpersonation_username_success_carries_roles_on_identity",
+          "outcome": "Passed"
+        },
+        {
+          "name": "ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests.OpcUaApplicationHostSecurityTests.BuildUserTokenPolicies_emits_anonymous_and_username",
+          "outcome": "Passed"
+        }
+      ],
+      "rawHashes": {
+        "close.trx": "800ac4134f48fc901da2c3519573184337c3f9a22cc5e7a5f4859a6b586fbf70",
+        "run.log": "253457c47f8801d37ce5fafc5c95cb9f5f0a9bbcead363c4ad0f5256e18fc7df",
+        "receipt.json": "b0db91f846ca883c8b8d48a946021f5e99785a26c37ff1087187b83575f62c96"
+      }
+    }
+  ],
+  "existingSkip": "AddressSpaceComposerDeviceHostTests.Compose_resolves_driver_device_and_device_host",
+  "newSkips": 0,
+  "ownership": "One caller owns detach/Stop/terminal Dispose; all Dispose callers await shared completion; Start after disposal begins is rejected.",
+  "limits": [
+    "Does not serialize an already-running asynchronous Start against Dispose.",
+    "Does not alter immutable83ed880e baseline publications or aggregate regression oracles.",
+    "Does not establish any relative RSS denominator, #799 acceptance or SDK2 release approval."
+  ],
+  "review": "Astra approved the isolated SDK1 source/result/package gate: byte-identical original production RED; five expected REDs and five GREENs; 1032 owning-suite passes with one existing skip; all build hashes and selected security/PKI results verified. No SDK2 release or performance-baseline approval."
+}
@@ -0,0 +1,71 @@
+# Linux late-slope diagnostic
+
+The [protocol](../LATE_SLOPE.md) fixes 7,200 seconds of canonical cycling from
+startup completion, with the existing active restart every fourth cycle. The
+cycle-three baseline and resource caps remain in force. A latched rejection
+keeps the original failing test and container exit. The result records
+`acceptance=false` even when the diagnostic completes its duration and shutdown.
+
+The analyzer requires an ordered active, recovered-stall, resource-observation,
+quiescent and complete-cycle ledger. It checks the 25-second restart helper
+bound, 30-second complete cycle bound and 60-second shutdown bound. It
+recomputes the first cold rejection and verifies its assertion against the
+actual TRX. It checks each fixed time window and the entire 1,800–7,200 second
+range before reporting RSS and retained-managed slopes in bytes per hour.
+The report includes sample counts, quiescent peaks, final values and coverage.
+It identifies whether the earlier saturation workload had completed by 1,800
+seconds, which bounds any interpretation relative to the previously observed
+restart step. The raw JSONL keeps every sample and the first assertion stack.
+
+Certificate allocation stack tracking remains enabled and contributes to the
+measured process. The slope report is descriptive and carries no resource
+pass threshold. Quiescent peaks refer to the sampled points.
+
+## Offline controls and artifact preparation
+
+The synthetic controls exercise the parser, receipts and bundle manifest with
+temporary files. Run them with Python 3.11 or later.
+
+```bash
+python3 scripts/validation/opcua-lifecycle-memory/late-slope/test_offline.py
+```
+
+Prepare from the reviewed clean commit containing the canonical test and tools.
+The image argument is the immutable ID of a locally cached Linux ARM64 SDK
+image. The explicitly selected Docker context must use a local Unix socket.
+The preparation command publishes the self-contained .NET 10.0.14 app and
+records the exact SDK 2.0.1 and HistorianGateway 1.0.0 graph, source hashes, file
+hashes and managed module MVIDs. It uses the repository's read-only PE metadata
+inspector. Preparation completes before any descriptor or lifecycle test runs.
+
+```bash
+python3 scripts/validation/opcua-lifecycle-memory/late-slope/prepare.py \
+  --source-revision "$reviewedRevision" --docker-context "$localContext" \
+  --image "$cachedImageId" --output "$bundle"
+python3 "$bundle/tools/launch.py" --bundle "$bundle" \
+  --receipt-sha256 "$reviewedBundleReceiptHash" --verify-only
+```
+
+## Reviewed run and analysis
+
+Root owns launch after review of the exact bundle receipt and controls. The
+bundled launcher uses the pinned Unix endpoint, immutable image, network-none
+container, read-only root filesystem and read-only application and tools mounts.
+A fresh private task directory holds results and fixture files. The two
+descriptor controls must execute successfully before the canonical test starts. The launcher verifies
+the actual test process's apphost, mapped CLR and CLR hash. Its outer deadline
+is 7,440 seconds, with cleanup limited to the uniquely labeled container.
+
+```bash
+python3 "$bundle/tools/launch.py" --bundle "$bundle" \
+  --receipt-sha256 "$reviewedBundleReceiptHash" --output "$run"
+python3 "$bundle/tools/analyze.py" --run "$run" \
+  --bundle-receipt "$bundle/receipt.json" --bundle-receipt-sha256 "$reviewedBundleReceiptHash"
+```
+
+A completed diagnostic with a cold-cap rejection has a nonzero launcher exit.
+Its receipt records diagnostic completion separately from the actual test and
+container status. The offline analyzer verifies the reviewed bundle receipt,
+raw evidence hashes, test identities and outcomes, fixed duration and verified
+shutdown before emitting a report. It can run from copies of the small bundle
+receipt and run evidence on a machine without Docker.
@@ -0,0 +1,423 @@
+#!/usr/bin/env python3
+"""Validate a recorded Linux late-slope diagnostic and describe its quiescent samples."""
+
+import argparse
+import hashlib
+import json
+import math
+from pathlib import Path
+import re
+from statistics import median
+import sys
+import xml.etree.ElementTree as ET
+
+
+PROTOCOL = "sdk2-late-slope-v1"
+DOCKER_LABEL = "org.otopcua.late-slope-run"
+MODE = "late-slope"
+SECONDS = 7200
+MAX_CYCLE_SECONDS = 30
+RESTART_SECONDS = 25
+SHUTDOWN_SECONDS = 60
+CONTAINER_SECONDS = 7440
+RUNTIME = "10.0.14"
+SDK = "2.0.1+07e62f72500cc0923a2bab940653c8c1116fa286"
+APP = "ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests"
+TEST_CLASS = APP + ".Sdk2LifecycleSoakTests."
+TEST_NAME = TEST_CLASS + "Repeated_lifecycles_return_resources_to_a_steady_baseline"
+DESCRIPTOR_TESTS = {
+    TEST_CLASS + "Descriptor_probe_observes_an_open_PKI_file_through_a_symlink_alias_and_its_close",
+    TEST_CLASS + "Descriptor_probe_times_out_and_reaps_a_process_that_keeps_both_streams_open",
+}
+STALL_STATUSES = {"OperationCanceled", "BadTimeout", "BadRequestCancelledByClient", "BadRequestTimeout", "BadRequestInterrupted"}
+WINDOWS = ((1800, 2400), (3600, 4200), (6600, 7200))
+BUDGETS = {"fdGrowth": 32, "retainedManagedGrowthBytes": 64 << 20, "rssGrowthBytes": 128 << 20}
+SETTINGS = {
+    "protocol": PROTOCOL, "mode": MODE, "requestedSeconds": SECONDS, "rid": "linux-arm64",
+    "runtimeVersion": RUNTIME, "selfContained": True, "acceptance": False,
+    "startupTimeoutSeconds": 120, "cycleTimeoutSeconds": MAX_CYCLE_SECONDS,
+    "cycleDeadlineSeconds": SECONDS + MAX_CYCLE_SECONDS, "restartTimeoutSeconds": RESTART_SECONDS,
+    "ownerTimeoutSeconds": 20, "shutdownTimeoutSeconds": SHUTDOWN_SECONDS,
+    "containerTimeoutSeconds": CONTAINER_SECONDS, "certificateLeakTracking": True,
+    "windowsSeconds": [list(window) for window in WINDOWS], "limits": BUDGETS,
+}
+CAP_FIELDS = (("Descriptors", "fdGrowth"), ("ManagedBytes", "retainedManagedGrowthBytes"), ("RssBytes", "rssGrowthBytes"))
+ZERO_FIELDS = (
+    "Sessions", "Subscriptions", "Items", "BrowsePoints", "HistoryPoints",
+    "LiveDataItems", "LiveEventItems", "MonitoredNodes", "PkiDescriptors",
+)
+STEADY_FIELDS = ("Variables", "Folders", "Conditions", "CachedCertificateStores")
+MEASUREMENTS = (
+    "Descriptors", "ManagedBytes", "RssBytes", "AllocatedBytes", "Gen0", "Gen1", "Gen2",
+    "GcCommittedBytes", "GcHeapBytes", "GcFragmentedBytes", "PinnedObjects", "PendingFinalizers",
+    "JitMethods", "JitIlBytes", "ActiveTimers", "Threads",
+)
+SAMPLE_FIELDS = {"Cycle", "Seconds", *ZERO_FIELDS, *STEADY_FIELDS, *MEASUREMENTS}
+EVIDENCE_FILES = (
+    "launch.json", "container.log", "container-exit-status.txt", "controller-runtime.txt",
+    "descriptor-controls.log", "test.log", "target-runtime.txt", "resources.jsonl",
+    "test-exit-status.txt", "results/descriptor-controls.trx", "results/late-slope.trx",
+)
+TRX = {"t": "http://microsoft.com/schemas/VisualStudio/TeamTest/2010"}
+
+
+def require(condition, message):
+    if not condition:
+        raise ValueError(message)
+
+
+def local_context(context):
+    return (type(context) is dict and set(context) == {"name", "endpoint"}
+            and isinstance(context["name"], str) and re.fullmatch(r"[A-Za-z0-9_.-]+", context["name"])
+            and isinstance(context["endpoint"], str) and context["endpoint"].startswith("unix:///")
+            and not any(character in context["endpoint"] for character in "\r\n\0"))
+
+
+def container_command(root, output, bundle, run_id):
+    require(all(path.is_absolute() and ":" not in str(path) for path in (root, output)), "Invalid Docker bind paths")
+    require(local_context(bundle["dockerContext"]), "Expected the reviewed local Unix Docker context")
+    return ["docker", "--host", bundle["dockerContext"]["endpoint"], "run", "--rm", "--init", "--pull", "never",
+            "--network", "none", "--platform", "linux/arm64", "--read-only", "--workdir", "/task",
+            "--name", f"opcua-late-slope-{run_id}",
+            "--label", f"{DOCKER_LABEL}={run_id}", "-e", "OTOPCUA_SOAK_BUILD_SHA=" + bundle["sourceRevision"],
+            "-v", f"{root / 'app'}:/app:ro", "-v", f"{root / 'tools'}:/tools:ro", "-v", f"{output}:/task",
+            bundle["image"], "bash", "/tools/container.sh"]
+
+
+def digest(path):
+    with path.open("rb") as stream:
+        return hashlib.file_digest(stream, "sha256").hexdigest()
+
+
+def unique_object(pairs):
+    result = {}
+    for key, value in pairs:
+        require(key not in result, f"Duplicate JSON key {key}")
+        result[key] = value
+    return result
+
+
+def reject_constant(value):
+    raise ValueError(f"Invalid JSON constant {value}")
+
+
+def json_object(text):
+    value = json.loads(text, object_pairs_hook=unique_object, parse_constant=reject_constant)
+    require(type(value) is dict, "Expected a JSON object")
+    return value
+
+
+def read_object(path):
+    return json_object(path.read_text(encoding="utf-8"))
+
+
+def read_rows(path):
+    with path.open(encoding="utf-8") as stream:
+        for line_number, line in enumerate(stream, 1):
+            require(line.endswith("\n"), f"Truncated JSONL row {line_number}")
+            try:
+                yield json_object(line)
+            except ValueError as error:
+                raise ValueError(f"Invalid JSONL row {line_number}: {error}") from error
+
+
+def count(value, label, minimum=0):
+    require(type(value) is int and minimum <= value <= (1 << 63) - 1, f"Invalid {label}")
+    return value
+
+
+def seconds(value, label, minimum=0, maximum=CONTAINER_SECONDS):
+    require(type(value) in (int, float) and minimum <= value <= maximum, f"Invalid {label}")
+    return value
+
+
+def assertion_message(text):
+    require(isinstance(text, str), "Missing original resource-cap assertion")
+    message = text.replace("\r", "").split("\n   at ", 1)[0].strip()
+    normalized = re.sub(r"^Shouldly\.ShouldAssertException\s*:\s*", "Shouldly.ShouldAssertException: ", message)
+    require(normalized.startswith("Shouldly.ShouldAssertException: ") and len(normalized) > 32,
+            "The recorded failure is not a resource-cap assertion")
+    return normalized
+
+
+def sample(row, cycle):
+    value = row["resources"]
+    require(type(value) is dict and set(value) == SAMPLE_FIELDS, f"Invalid sample fields at cycle {cycle}")
+    require(count(value["Cycle"], "sample cycle", 1) == cycle, f"Out-of-order sample cycle {cycle}")
+    seconds(value["Seconds"], "sample time", maximum=SECONDS + MAX_CYCLE_SECONDS)
+    for field in SAMPLE_FIELDS - {"Cycle", "Seconds"}:
+        count(value[field], field)
+    require(all(value[field] == 0 for field in ZERO_FIELDS), f"Quiescent ownership failure at cycle {cycle}")
+    return value
+
+
+def validate_rows(rows, revision):
+    """Validate the canonical row order, original budget and fixed-duration outcome."""
+    rows = iter(rows)
+
+    def take(kind):
+        row = next(rows, {})
+        require(row.get("kind") == kind, f"Expected {kind}, found {row.get('kind')!r}")
+        require(row.get("acceptance", False) is False and "satisfiesTwoHourDuration" not in row,
+                "An ordinary acceptance result is forbidden")
+        return row
+
+    start = take("start")
+    require(start.get("mode") == MODE and start.get("requestedSeconds") == SECONDS
+            and type(start["requestedSeconds"]) is int and start.get("minimumFullSeconds") == SECONDS
+            and start.get("acceptance") is False, "Wrong fixed-duration diagnostic mode")
+    require(start.get("build") == revision and re.fullmatch(r"[0-9a-f]{40}", revision), "Wrong source revision")
+    require(start.get("runtime") == RUNTIME and start.get("architecture") == "Arm64" and start.get("sdk") == SDK,
+            "Wrong Linux ARM64 runtime or SDK version")
+    count(start.get("processId"), "test process ID", 1)
+    seconds(start.get("startupSeconds"), "startup duration", maximum=SETTINGS["startupTimeoutSeconds"])
+    limits = start.get("limits")
+    require(type(limits) is dict and limits == BUDGETS and all(type(value) is int for value in limits.values()),
+            "Original cold resource caps changed")
+
+    samples = []
+    first_rejection = None
+    completions = []
+    # Buffer only a single row so the actual record order cannot be reconstructed by sorting.
+    pending = next(rows, {})
+    while pending.get("kind") == "active":
+        cycle = len(samples) + 1
+        require(count(pending.get("cycle"), "active cycle", 1) == cycle, "Missing or duplicate active cycle")
+        active_counts = tuple(count(pending.get(field), field) for field in ("sessions", "subscriptions", "items"))
+        require(active_counts == (2, 1, 3), f"Incomplete active session, subscription or item exercise at cycle {cycle}")
+        for field in ("browsePoints", "historyPoints"):
+            count(pending.get(field), field, 1)
+        stall = take("network-stall-recovered")
+        require(count(stall.get("cycle"), "recovered stall cycle", 1) == cycle, "Missing or duplicate recovered stall")
+        seconds(stall.get("seconds"), "network stall time", maximum=MAX_CYCLE_SECONDS)
+        require(isinstance(stall.get("status"), str) and stall["status"] in STALL_STATUSES,
+                "Unexpected recovered network stall status")
+        value = sample(take("resource-observation"), cycle)
+        require(not samples or value["Seconds"] > samples[-1]["Seconds"], "Nonmonotonic or duplicate sample time")
+        require(not completions or value["Seconds"] > completions[-1]["elapsedSeconds"],
+                "Sample precedes completion of the previous cycle")
+        samples.append(value)
+
+        if cycle >= 3:
+            baseline = samples[2]
+            require(all(value[field] == baseline[field] for field in STEADY_FIELDS),
+                    f"Quiescent model or certificate-store inventory changed at cycle {cycle}")
+            exceeded = [field for field, cap in CAP_FIELDS if value[field] > baseline[field] + BUDGETS[cap]]
+            if exceeded and first_rejection is None:
+                rejection = take("budget-rejected")
+                require(count(rejection.get("cycle"), "rejection cycle", 1) == cycle
+                        and sample(rejection, cycle) == value
+                        and sample({"resources": rejection.get("baseline")}, 3) == baseline,
+                        "First rejection does not match the failing sample and original cycle-three baseline")
+                seconds(rejection.get("seconds"), "rejection time", minimum=value["Seconds"],
+                        maximum=SECONDS + MAX_CYCLE_SECONDS)
+                first_rejection = {**rejection, "exceededFields": exceeded,
+                                   "assertionMessage": assertion_message(rejection.get("failure"))}
+            if cycle >= 26:
+                retained = [row["ManagedBytes"] for row in samples[-24:]]
+                require(not (all(a < b for a, b in zip(retained, retained[1:]))
+                             and retained[-1] - retained[0] > (16 << 20)),
+                        f"Managed growth guard failed at cycle {cycle}")
+
+        quiescent = take("quiescent-budget-rejected" if first_rejection else "quiescent")
+        require(sample(quiescent, cycle) == value, "Quiescent sample differs from its pre-assertion observation")
+        quiescent_seconds = seconds(quiescent.get("cycleSeconds"), "quiescent cycle duration", maximum=MAX_CYCLE_SECONDS)
+        restart_seconds = 0
+        if cycle % 4 == 0:
+            restart = take("restart")
+            require(count(restart.get("cycles"), "restart cycle", 1) == cycle
+                    and count(restart.get("restarts"), "restart count", 1) == cycle // 4, "Missing or duplicate active restart")
+            bounded = seconds(restart.get("boundedRestartSeconds"), "bounded restart duration", maximum=RESTART_SECONDS)
+            restart_seconds = seconds(restart.get("seconds"), "combined restart duration", minimum=bounded, maximum=MAX_CYCLE_SECONDS)
+        completion = take("late-slope-cycle")
+        require(count(completion.get("cycle"), "completed cycle", 1) == cycle
+                and count(completion.get("restarts"), "completed restarts") == cycle // 4, "Invalid completed cycle ledger")
+        elapsed = seconds(completion.get("elapsedSeconds"), "cycle completion time", minimum=value["Seconds"],
+                          maximum=SECONDS + MAX_CYCLE_SECONDS)
+        seconds(completion.get("cycleSeconds"), "complete cycle duration", minimum=quiescent_seconds + restart_seconds,
+                maximum=MAX_CYCLE_SECONDS)
+        if first_rejection and first_rejection["cycle"] == cycle:
+            require(first_rejection["seconds"] <= elapsed, "Rejection follows its completed cycle")
+        require(not completions or completions[-1]["elapsedSeconds"] < SECONDS,
+                "Another cycle began after the fixed duration")
+        completions.append(completion)
+        pending = next(rows, {})
+
+    result = pending
+    require(result.get("kind") == "late-slope-result" and next(rows, None) is None,
+            "Expected a unique final diagnostic result with no ordinary completion or trailing failure")
+    require(result.get("mode") == MODE and result.get("acceptance") is False
+            and "satisfiesTwoHourDuration" not in result and result.get("durationCompleted") is True
+            and result.get("shutdownVerified") is True and result.get("fatalFailure") is False and result.get("failureTypes") == [],
+            "Duration, shutdown or diagnostic outcome was incomplete")
+    cycles = count(result.get("cycles"), "result cycles", 4)
+    require(cycles == len(samples) and count(result.get("restarts"), "result restarts", 1) == cycles // 4,
+            "Final cycle or restart count differs from the ledger")
+    phase = seconds(result.get("cycleSeconds"), "canonical cycle phase", minimum=SECONDS, maximum=SECONDS + MAX_CYCLE_SECONDS)
+    require(phase >= completions[-1]["elapsedSeconds"] and completions[-1]["elapsedSeconds"] >= SECONDS,
+            "The final cycle does not cover the fixed duration")
+    seconds(result.get("shutdownSeconds"), "shutdown duration", maximum=SHUTDOWN_SECONDS)
+    require(result.get("coldBudgetRejected") is (first_rejection is not None), "The first cold-cap rejection was not preserved")
+    return start, samples, completions, first_rejection, result
+
+
+def sample_summary(samples, fields):
+    times = [value["Seconds"] for value in samples]
+    summary = {"sampleCount": len(samples), "firstSeconds": times[0], "lastSeconds": times[-1],
+               "spanSeconds": times[-1] - times[0], "maximumGapSeconds": max((b - a for a, b in zip(times, times[1:])), default=0)}
+    for field in fields:
+        values = [value[field] for value in samples]
+        summary[field] = {"first": values[0], "last": values[-1], "min": min(values), "max": max(values), "median": median(values)}
+    return summary
+
+
+def slope(samples, field):
+    times = [value["Seconds"] for value in samples]
+    mean = math.fsum(times) / len(times)
+    mean_value = math.fsum(value[field] for value in samples) / len(samples)
+    variance = math.fsum((time - mean) ** 2 for time in times)
+    return 3600 * math.fsum((time - mean) * (value[field] - mean_value)
+                           for time, value in zip(times, samples)) / variance
+
+
+def describe(samples, completions):
+    windows = []
+    for begin, end in WINDOWS:
+        subset = [value for value in samples if begin <= value["Seconds"] <= end]
+        require(len(subset) >= 10, f"Insufficient observations in {begin}–{end} seconds")
+        summary = sample_summary(subset, MEASUREMENTS)
+        require(summary["firstSeconds"] <= begin + 60 and summary["lastSeconds"] >= end - 60
+                and summary["spanSeconds"] >= 480 and summary["maximumGapSeconds"] <= 60,
+                f"Incomplete coverage in {begin}–{end} seconds")
+        windows.append({"windowSeconds": [begin, end], **summary})
+    late = [value for value in samples if 1800 <= value["Seconds"] <= SECONDS]
+    coverage = sample_summary(late, ())
+    require(coverage["maximumGapSeconds"] <= 60, "An observation gap exceeds 60 seconds across the late phase")
+    previous = [row for row in completions if row["elapsedSeconds"] <= 1800]
+    cycles = len(previous)
+    restarts = previous[-1]["restarts"] if previous else 0
+    work = {"cycles": cycles, "restarts": restarts, "earlierSaturationWorkloadReached": cycles >= 385 and restarts >= 96}
+    if not work["earlierSaturationWorkloadReached"]:
+        work["interpretation"] = "Post-restart-step attribution withheld because the earlier workload counts fell short by 1800 seconds."
+    return {
+        "allQuiescent": sample_summary(samples, (*ZERO_FIELDS, *STEADY_FIELDS, *MEASUREMENTS)),
+        "finalQuiescentSample": samples[-1],
+        "windows": windows,
+        "leastSquares": {**coverage, "rangeSeconds": [1800, SECONDS], "units": "bytes/hour",
+                         "rss": slope(late, "RssBytes"), "retainedManaged": slope(late, "ManagedBytes")},
+        "workCompletedBy1800Seconds": work,
+    }
+
+
+def check_trx(path, expected_names, failure=None):
+    document = ET.parse(path).getroot()
+    counters = document.findall("./t:ResultSummary/t:Counters", TRX)
+    require(len(counters) == 1, f"Missing unique counters in {path.name}")
+    counters = counters[0].attrib
+    total = len(expected_names)
+    failed = 1 if failure else 0
+    expected = {"total": total, "executed": total, "passed": total - failed, "failed": failed,
+                "error": 0, "timeout": 0, "aborted": 0, "notExecuted": 0}
+    for key, value in expected.items():
+        require(counters.get(key) == str(value), f"Unexpected {key} count in {path.name}")
+    for key in ("inconclusive", "passedButRunAborted", "notRunnable", "disconnected", "inProgress", "pending"):
+        require(counters.get(key, "0") == "0", f"Unexpected {key} count in {path.name}")
+    results = document.findall("./t:Results/t:UnitTestResult", TRX)
+    require(len(results) == total and {row.get("testName") for row in results} == expected_names,
+            f"Wrong test identities in {path.name}")
+    require(all(row.get("outcome") == ("Failed" if failure else "Passed") for row in results),
+            f"Unexpected test outcome in {path.name}")
+    if failure:
+        actual = results[0].findtext("./t:Output/t:ErrorInfo/t:Message", namespaces=TRX)
+        require(assertion_message(actual) == failure, "TRX failure differs from the first cold-cap assertion")
+    return {"executed": total, "passed": total - failed, "failed": failed, "skipped": 0}
+
+
+def exit_status(path):
+    value = path.read_text(encoding="ascii").strip()
+    require(re.fullmatch(r"0|[1-9][0-9]{0,2}", value) and int(value) <= 255, f"Invalid {path.name}")
+    return int(value)
+
+
+def inspect_run(path, provenance, wall_seconds):
+    """Check actual run evidence. This function neither launches Docker nor changes the evidence."""
+    start, samples, completions, rejection, result = validate_rows(read_rows(path / "resources.jsonl"), provenance["sourceRevision"])
+    wall = seconds(wall_seconds, "container wall time", minimum=SECONDS)
+    require(start["startupSeconds"] + result["cycleSeconds"] + result["shutdownSeconds"] <= wall,
+            "Startup, cycle and shutdown phases exceed measured wall time")
+    test_exit = exit_status(path / "test-exit-status.txt")
+    container_exit = exit_status(path / "container-exit-status.txt")
+    require(test_exit == container_exit and ((test_exit != 0) == bool(rejection)), "Original test/container exit was not preserved")
+    mapping = (path / "target-runtime.txt").read_text(encoding="ascii").splitlines()
+    require(mapping == [f"targetPid={start['processId']}", f"actualExecutable=/app/{APP}",
+                        "mappedCoreClr=/app/libcoreclr.so", f"coreClrSha256={provenance['coreClrSha256']}"],
+            "Published Linux apphost or mapped CLR was not verified for the recorded PID")
+    controls = check_trx(path / "results/descriptor-controls.trx", DESCRIPTOR_TESTS)
+    workload = check_trx(path / "results/late-slope.trx", {TEST_NAME}, rejection and rejection["assertionMessage"])
+    return {"diagnosticOutcome": result, "firstColdRejection": rejection, "descriptorTests": controls,
+            "experimentTests": workload, "testExitStatus": test_exit, "containerExitStatus": container_exit,
+            **describe(samples, completions)}
+
+
+def evidence_hashes(path):
+    result = {}
+    for name in EVIDENCE_FILES:
+        file = path / name
+        require(file.is_file() and file.resolve() == file, f"Missing or linked evidence file {name}")
+        result[name] = digest(file)
+    return result
+
+
+def analyze_run(path, bundle_receipt_path, bundle_hash):
+    require(re.fullmatch(r"[0-9a-f]{64}", bundle_hash) and digest(bundle_receipt_path) == bundle_hash,
+            "Reviewed bundle receipt hash differs")
+    bundle = read_object(bundle_receipt_path)
+    require(bundle.get("kind") == "bundle" and all(type(bundle.get(key)) is type(value) and bundle[key] == value
+                                                    for key, value in SETTINGS.items()),
+            "Invalid immutable bundle protocol")
+    receipt = read_object(path / "run-receipt.json")
+    provenance = read_object(path / "launch.json")
+    require(receipt.get("kind") == "run" and provenance.get("kind") == "launch", "Wrong receipt types")
+    for record in (receipt, provenance):
+        require(record.get("protocol") == PROTOCOL and record.get("acceptance") is False
+                and record.get("bundleReceiptSha256") == bundle_hash, "Wrong diagnostic provenance")
+    for field in ("runId", "sourceRevision", "image", "coreClrSha256", "dockerContext", "command", "containerTimeoutSeconds"):
+        require(receipt[field] == provenance[field], f"Launch and run receipts disagree on {field}")
+    require(receipt["sourceRevision"] == bundle["sourceRevision"] and receipt["image"] == bundle["image"]
+            and receipt["coreClrSha256"] == bundle["files"]["app/libcoreclr.so"]
+            and receipt["dockerContext"] == bundle["dockerContext"],
+            "Recorded source, image or runtime differs from the reviewed bundle")
+    require(re.fullmatch(r"[0-9a-f]{32}", receipt["runId"])
+            and re.fullmatch(r"sha256:[0-9a-f]{64}", receipt["image"])
+            and re.fullmatch(r"[0-9a-f]{64}", receipt["coreClrSha256"]), "Invalid run identity or immutable provenance")
+    require(receipt.get("diagnosticCompleted") is True and receipt.get("containerRemoved") is True,
+            "The launcher did not verify duration, shutdown and container cleanup")
+    require("validationErrors" not in receipt and provenance["containerTimeoutSeconds"] == CONTAINER_SECONDS
+            and provenance["command"] == container_command(Path(provenance["bundlePath"]), Path(provenance["outputPath"]),
+                                                           bundle, receipt["runId"]), "The recorded container invocation differs")
+    require(receipt.get("evidenceSha256") == evidence_hashes(path), "Recorded evidence hashes differ")
+    report = inspect_run(path, provenance, receipt["wallSeconds"])
+    for field in ("diagnosticOutcome", "testExitStatus", "containerExitStatus", "descriptorTests", "experimentTests"):
+        require(receipt[field] == report[field], f"Run receipt differs from the raw {field}")
+    require(receipt.get("launcherExitStatus") == report["testExitStatus"], "The launcher did not preserve the original exit")
+    return {"protocol": PROTOCOL, "acceptance": False, "bundleReceiptSha256": bundle_hash,
+            "sourceRevision": receipt["sourceRevision"], "evidenceSha256": receipt["evidenceSha256"], **report}
+
+
+def main():
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--run", type=Path, required=True)
+    parser.add_argument("--bundle-receipt", type=Path, required=True)
+    parser.add_argument("--bundle-receipt-sha256", required=True)
+    options = parser.parse_args()
+    try:
+        print(json.dumps(analyze_run(options.run.resolve(strict=True), options.bundle_receipt,
+                                     options.bundle_receipt_sha256), indent=2, allow_nan=False))
+    except (OSError, ValueError, KeyError, TypeError, ET.ParseError) as error:
+        print(f"Incomplete late-slope evidence: {error}", file=sys.stderr)
+        return 1
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())
@@ -0,0 +1,129 @@
+"""Checks shared by immutable artifact preparation and launch."""
+
+import json
+import os
+from pathlib import Path
+import re
+import stat
+import subprocess
+import sys
+from uuid import UUID
+
+sys.dont_write_bytecode = True
+
+from analyze import APP, RUNTIME, SETTINGS, digest, local_context, read_object, require
+
+
+TOOLS = ("analyze.py", "bundle.py", "launch.py", "container.sh", "README.md")
+TEST_DIRECTORY = f"tests/Server/{APP}"
+CANONICAL_SOURCES = tuple(f"{TEST_DIRECTORY}/{name}" for name in (
+    "Sdk2LifecycleSoakTests.cs", "Sdk2LifecycleLateSlopeTests.cs", f"{APP}.csproj"))
+SDK_PACKAGES = {
+    f"OPCFoundation.NetStandard.Opc.Ua.{component}/2.0.1"
+    for component in ("Client", "Configuration", "Core", "Core.Schema", "Core.Types",
+                      "Security.Certificates", "Server", "Types")
+}
+GATEWAY_PACKAGES = {f"ZB.MOM.WW.HistorianGateway.{component}/1.0.0" for component in ("Client", "Contracts")}
+
+
+def context_identity(name):
+    require(re.fullmatch(r"[A-Za-z0-9_.-]+", name), "Invalid local Docker context name")
+    require(not any(os.environ.get(key) for key in ("DOCKER_HOST", "DOCKER_TLS", "DOCKER_TLS_VERIFY", "DOCKER_CERT_PATH"))
+            and os.environ.get("DOCKER_CONTEXT", name) in ("", name), "Docker endpoint or TLS override is set")
+    result = subprocess.check_output(["docker", "context", "inspect", name], text=True, timeout=15)
+    contexts = json.loads(result)
+    require(len(contexts) == 1 and contexts[0]["Name"] == name, "Expected one explicit Docker context")
+    identity = {"name": name, "endpoint": contexts[0]["Endpoints"]["docker"]["Host"]}
+    require(local_context(identity) and Path(identity["endpoint"][len("unix://"):]).is_socket(),
+            "The reviewed Docker context must use a local Unix socket")
+    return identity
+
+
+def image_identity(image, context):
+    require(re.fullmatch(r"sha256:[0-9a-f]{64}", image), "Use a locally cached immutable image ID")
+    result = subprocess.check_output(["docker", "--host", context["endpoint"], "image", "inspect", image], text=True, timeout=15)
+    images = json.loads(result)
+    require(len(images) == 1, "Expected one cached image")
+    value = images[0]
+    require((value["Id"], value["Os"], value["Architecture"]) == (image, "linux", "arm64"),
+            "The immutable image must be Linux ARM64")
+    return {"id": image, "os": "linux", "architecture": "arm64"}
+
+
+def manifest(root):
+    """Hash the exact regular-file tree and exclude only the receipt that contains it."""
+    result = {}
+    for path in sorted(root.rglob("*")):
+        kind = path.lstat().st_mode
+        require(stat.S_ISDIR(kind) or stat.S_ISREG(kind), f"Unsupported artifact type {path.relative_to(root)}")
+        if stat.S_ISREG(kind) and path != root / "receipt.json":
+            result[path.relative_to(root).as_posix()] = digest(path)
+    return result
+
+
+def check_elf(path):
+    with path.open("rb") as stream:
+        header = stream.read(20)
+    require(header[:6] == b"\x7fELF\x02\x01" and header[18:20] == b"\xb7\x00",
+            f"{path.name} is not ELF64 AArch64")
+
+
+def check_application(app):
+    check_elf(app / APP)
+    require(os.access(app / APP, os.X_OK), "The Linux apphost is not executable")
+    check_elf(app / "libcoreclr.so")
+    options = read_object(app / f"{APP}.runtimeconfig.json")["runtimeOptions"]
+    frameworks = options["includedFrameworks"]
+    require(options["tfm"] == "net10.0" and "framework" not in options and "frameworks" not in options
+            and isinstance(frameworks, list) and any(value["name"] == "Microsoft.NETCore.App" for value in frameworks)
+            and all(value["version"] == RUNTIME for value in frameworks), "Wrong self-contained application runtime")
+    dependencies = read_object(app / f"{APP}.deps.json")
+    require(dependencies["runtimeTarget"]["name"] == ".NETCoreApp,Version=v10.0/linux-arm64", "Wrong application RID")
+    libraries = dependencies["libraries"]
+    sdk = {key for key in libraries if key.startswith("OPCFoundation.")}
+    gateway = {key for key in libraries if key.startswith("ZB.MOM.WW.HistorianGateway.")}
+    require(sdk == SDK_PACKAGES and gateway == GATEWAY_PACKAGES, "Expected exact SDK 2.0.1 and HistorianGateway 1.0.0 graph")
+    return sorted(sdk), sorted(gateway)
+
+
+def check_modules(modules, files):
+    require(isinstance(modules, list), "Missing managed module identities")
+    seen = set()
+    for module in modules:
+        name = module["file"]
+        require(name in files and name.startswith("app/") and name.endswith(".dll") and name not in seen,
+                "Unexpected or duplicate managed module file")
+        require(module["assembly"] and re.fullmatch(r"\d+(?:\.\d+){3}", module["version"])
+                and UUID(module["mvid"]).int != 0, "Invalid assembly version or MVID")
+        seen.add(name)
+    required = {f"app/{APP}.dll", "app/Opc.Ua.Server.dll", "app/System.Private.CoreLib.dll",
+                "app/ZB.MOM.WW.HistorianGateway.Client.dll", "app/ZB.MOM.WW.HistorianGateway.Contracts.dll"}
+    require(required <= seen, "Missing application, runtime or dependency module identity")
+
+
+def verify_bundle(root, expected_hash):
+    path = root / "receipt.json"
+    require(path.is_file() and not path.is_symlink() and re.fullmatch(r"[0-9a-f]{64}", expected_hash)
+            and digest(path) == expected_hash, "Reviewed bundle receipt hash differs")
+    receipt = read_object(path)
+    require(receipt.get("kind") == "bundle" and all(type(receipt.get(key)) is type(value) and receipt[key] == value
+                                                    for key, value in SETTINGS.items()), "Invalid fixed diagnostic settings")
+    require(re.fullmatch(r"[0-9a-f]{40}", receipt["sourceRevision"]), "Invalid committed source revision")
+    files = manifest(root)
+    require(receipt["files"] == files, "The immutable bundle file set or a file hash changed")
+    require({name for name in files if name.startswith("tools/")} == {f"tools/{name}" for name in TOOLS},
+            "Unexpected bundled tools")
+    sdk, gateway = check_application(root / "app")
+    require(receipt["sdkLibraries"] == sdk and receipt["gatewayLibraries"] == gateway,
+            "Dependency receipt differs from the published application")
+    check_modules(receipt["managedModules"], files)
+    require(receipt["imageIdentity"] == {"id": receipt["image"], "os": "linux", "architecture": "arm64"}
+            and re.fullmatch(r"sha256:[0-9a-f]{64}", receipt["image"]), "Invalid immutable image identity")
+    require(local_context(receipt["dockerContext"]), "Invalid reviewed local Docker context")
+    require(receipt["sourceFiles"] and all(re.fullmatch(r"[0-9a-f]{64}", value) for value in receipt["sourceFiles"].values()),
+            "Missing committed source hashes")
+    require(set(CANONICAL_SOURCES) <= receipt["sourceFiles"].keys(), "Missing canonical test or project hash")
+    for name in TOOLS:
+        source = "scripts/validation/opcua-lifecycle-memory/late-slope/" + name
+        require(receipt["sourceFiles"].get(source) == files[f"tools/{name}"], f"Bundled {name} differs from source")
+    return receipt
@@ -0,0 +1,83 @@
+#!/usr/bin/env bash
+# Execute the immutable self-contained apphost through the cached Linux VSTest controller.
+verify_mapped_coreclr() {
+    awk -v identity="$2" '
+        $6 ~ /\/libcoreclr.so$/ {
+            if (NF != 6 || ($5 " " $6) != identity) bad = 1
+            else matched = 1
+        }
+        END { exit (matched && !bad) ? 0 : 1 }
+    ' "$1"
+}
+
+# The offline controls source only the mapping check above.
+if [[ ${BASH_SOURCE[0]} != "$0" ]]; then return 0; fi
+
+set -euo pipefail
+umask 077
+app=ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests
+test_class="$app.Sdk2LifecycleSoakTests"
+mkdir -p /task/tmp /task/data /task/cli /task/results
+export TMPDIR=/task/tmp XDG_DATA_HOME=/task/data DOTNET_CLI_HOME=/task/cli
+export OTOPCUA_SDK2_SOAK=1 OTOPCUA_SDK2_SOAK_MODE=late-slope OTOPCUA_SDK2_SOAK_SECONDS=7200
+export OTOPCUA_SDK2_SOAK_OUTPUT=/task/resources.jsonl OPCUA_CERTIFICATE_LEAK_TRACKING=1
+export DOTNET_CLI_TELEMETRY_OPTOUT=1 DOTNET_NOLOGO=1
+[[ $(uname -s) == Linux && $(uname -m) == aarch64 && -d /proc/self/fd ]]
+[[ ${OTOPCUA_SOAK_BUILD_SHA:-} =~ ^[0-9a-f]{40}$ ]]
+dotnet --info > /task/controller-runtime.txt
+
+dotnet vstest "/app/$app.dll" \
+    --TestCaseFilter:"FullyQualifiedName~$test_class.Descriptor_probe_" \
+    --ResultsDirectory:/task/results '--Logger:trx;LogFileName=descriptor-controls.trx' > /task/descriptor-controls.log 2>&1
+
+# Verify actual execution before starting the long test. The host parses the complete TRX independently.
+controls=/task/results/descriptor-controls.trx
+counters=$(grep -E '^[[:space:]]*<Counters ' "$controls")
+[[ $(grep -Ec '^[[:space:]]*<Counters ' "$controls") == 1 ]]
+for attribute in total=2 executed=2 passed=2 failed=0 notExecuted=0 error=0 timeout=0 aborted=0; do
+    [[ $counters == *" ${attribute%=*}=\"${attribute#*=}\""* ]]
+done
+[[ $(grep -Ec '^[[:space:]]*<UnitTestResult ' "$controls") == 2 ]]
+for name in Descriptor_probe_observes_an_open_PKI_file_through_a_symlink_alias_and_its_close \
+            Descriptor_probe_times_out_and_reaps_a_process_that_keeps_both_streams_open; do
+    result=$(grep -E '^[[:space:]]*<UnitTestResult ' "$controls" | grep -F "testName=\"$test_class.$name\"")
+    [[ $result == *' outcome="Passed"'* && $result != *$'\n'* ]]
+done
+
+vstest_pid=
+cleanup() {
+    if [[ -n "$vstest_pid" ]]; then kill "$vstest_pid" 2>/dev/null || true; fi
+}
+trap cleanup EXIT
+dotnet vstest "/app/$app.dll" \
+    --TestCaseFilter:"FullyQualifiedName=$test_class.Repeated_lifecycles_return_resources_to_a_steady_baseline" \
+    --ResultsDirectory:/task/results '--Logger:trx;LogFileName=late-slope.trx' > /task/test.log 2>&1 &
+vstest_pid=$!
+
+# The canonical startup deadline is 120 seconds. Allow controller startup before observing its first complete row.
+proof_deadline=$((SECONDS + 150))
+observed=0
+while ((SECONDS < proof_deadline)) && kill -0 "$vstest_pid" 2>/dev/null; do
+    if [[ -s /task/resources.jsonl ]] && IFS= read -r header < /task/resources.jsonl; then
+        [[ $header == *'"kind":"start"'* && $header == *'"mode":"late-slope"'* ]]
+        pid=$(sed -n 's/.*"processId":\([1-9][0-9]*\),.*/\1/p' <<< "$header")
+        [[ $pid =~ ^[1-9][0-9]*$ ]]
+        executable=$(readlink "/proc/$pid/exe")
+        [[ $executable == "/app/$app" ]]
+        runtime=/app/libcoreclr.so
+        verify_mapped_coreclr "/proc/$pid/maps" "$(stat -c '%i %n' "$runtime")"
+        runtime_hash=$(sha256sum "$runtime")
+        printf 'targetPid=%s\nactualExecutable=%s\nmappedCoreClr=%s\ncoreClrSha256=%s\n' \
+            "$pid" "$executable" "$runtime" "${runtime_hash%% *}" > /task/target-runtime.txt
+        observed=1
+        break
+    fi
+    sleep 0.1
+done
+if [[ $observed != 1 ]]; then kill "$vstest_pid" 2>/dev/null || true; fi
+test_status=0
+wait "$vstest_pid" || test_status=$?
+vstest_pid=
+printf '%s\n' "$test_status" > /task/test-exit-status.txt
+if [[ $test_status != 0 ]]; then exit "$test_status"; fi
+[[ $observed == 1 ]]
@@ -0,0 +1,130 @@
+#!/usr/bin/env python3
+"""Verify an immutable bundle or launch its fixed-duration Linux diagnostic."""
+
+import argparse
+import json
+from pathlib import Path
+import re
+import subprocess
+import sys
+import time
+import uuid
+import xml.etree.ElementTree as ET
+
+sys.dont_write_bytecode = True
+
+from analyze import CONTAINER_SECONDS, DOCKER_LABEL, PROTOCOL, container_command, evidence_hashes, inspect_run, require
+from bundle import context_identity, image_identity, verify_bundle
+
+
+def remove_container(context, run_id):
+    docker = ["docker", "--host", context["endpoint"]]
+
+    def containers():
+        result = subprocess.check_output([*docker, "ps", "--all", "--quiet", "--no-trunc",
+                                          "--filter", f"label={DOCKER_LABEL}={run_id}"], text=True, timeout=10)
+        ids = result.split()
+        require(all(re.fullmatch(r"[0-9a-f]{64}", value) for value in ids), "Unexpected container ID")
+        return ids
+
+    ids = containers()
+    if ids:
+        subprocess.run([*docker, "rm", "--force", *ids], stdout=subprocess.DEVNULL,
+                       stderr=subprocess.DEVNULL, timeout=15)
+    require(not containers(), "The task-owned container is still present")
+
+
+def write(path, value):
+    with path.open("x", encoding="utf-8") as stream:
+        json.dump(value, stream, indent=2, allow_nan=False)
+        stream.write("\n")
+
+
+def verify_local_image(bundle):
+    context = bundle["dockerContext"]
+    require(context_identity(context["name"]) == context, "Reviewed Docker context endpoint changed")
+    require(image_identity(bundle["image"], context) == bundle["imageIdentity"], "Wrong cached image identity")
+
+
+def launch(root, bundle, bundle_hash, output):
+    require(Path(__file__).resolve() == root / "tools/launch.py", "Invoke the reviewed bundled launcher")
+    require(not output.is_relative_to(root), "Run output must be outside the immutable bundle")
+    run_id = uuid.uuid4().hex
+    invocation = container_command(root, output, bundle, run_id)
+    output.mkdir(mode=0o700, parents=True, exist_ok=False)
+    provenance = {"kind": "launch", "protocol": PROTOCOL, "acceptance": False, "runId": run_id,
+                  "bundleReceiptSha256": bundle_hash, "sourceRevision": bundle["sourceRevision"], "image": bundle["image"],
+                  "coreClrSha256": bundle["files"]["app/libcoreclr.so"], "command": invocation,
+                  "dockerContext": bundle["dockerContext"], "bundlePath": str(root), "outputPath": str(output),
+                  "containerTimeoutSeconds": CONTAINER_SECONDS}
+    write(output / "launch.json", provenance)
+    errors = []
+    removed = False
+    started = time.monotonic()
+    try:
+        with (output / "container.log").open("x", encoding="utf-8") as log:
+            process = subprocess.run(invocation, stdout=log, stderr=subprocess.STDOUT, timeout=CONTAINER_SECONDS)
+        status = process.returncode
+    except subprocess.TimeoutExpired:
+        status = 124
+        errors.append("The container reached the fixed outer deadline")
+    except KeyboardInterrupt:
+        status = 130
+        errors.append("Container execution was interrupted")
+    except OSError as error:
+        status = 125
+        errors.append(str(error))
+    finally:
+        wall = time.monotonic() - started
+        try:
+            remove_container(bundle["dockerContext"], run_id)
+            removed = True
+        except (OSError, subprocess.SubprocessError, ValueError) as error:
+            errors.append(f"Container cleanup failed: {error}")
+    (output / "container-exit-status.txt").write_text(f"{status}\n", encoding="ascii")
+    receipt = {**provenance, "kind": "run", "containerExitStatus": status, "wallSeconds": wall,
+               "containerRemoved": removed, "diagnosticCompleted": False}
+    try:
+        verify_bundle(root, bundle_hash)
+        verify_local_image(bundle)
+        receipt["evidenceSha256"] = evidence_hashes(output)
+        inspected = inspect_run(output, provenance, wall)
+        for field in ("diagnosticOutcome", "testExitStatus", "descriptorTests", "experimentTests"):
+            receipt[field] = inspected[field]
+    except (OSError, KeyError, TypeError, ValueError, ET.ParseError, subprocess.SubprocessError) as error:
+        errors.append(f"Evidence validation failed: {error}")
+    receipt["diagnosticCompleted"] = not errors
+    if errors:
+        receipt["validationErrors"] = errors
+    # Return the original container status when the test rethrows its first resource cap.
+    receipt["launcherExitStatus"] = (128 - status if status < 0 else status) or int(bool(errors))
+    write(output / "run-receipt.json", receipt)
+    print(json.dumps(receipt, indent=2, allow_nan=False))
+    return receipt["launcherExitStatus"]
+
+
+def main():
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--bundle", type=Path, required=True)
+    parser.add_argument("--receipt-sha256", required=True)
+    action = parser.add_mutually_exclusive_group(required=True)
+    action.add_argument("--verify-only", action="store_true")
+    action.add_argument("--output", type=Path)
+    options = parser.parse_args()
+    try:
+        root = options.bundle.resolve(strict=True)
+        bundle = verify_bundle(root, options.receipt_sha256)
+        verify_local_image(bundle)
+        if options.verify_only:
+            print(json.dumps({"protocol": PROTOCOL, "sourceRevision": bundle["sourceRevision"],
+                              "receiptSha256": options.receipt_sha256, "verified": True,
+                              "acceptance": False, "workloadLaunched": False}, indent=2))
+            return 0
+        return launch(root, bundle, options.receipt_sha256, options.output.resolve())
+    except (OSError, KeyError, TypeError, ValueError, subprocess.SubprocessError) as error:
+        print(f"Late-slope launcher rejected the request: {error}", file=sys.stderr)
+        return 1
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())
@@ -0,0 +1,97 @@
+#!/usr/bin/env python3
+"""Publish and record a fresh immutable Linux ARM64 bundle without running the workload."""
+
+import argparse
+from datetime import datetime, timezone
+import json
+from pathlib import Path
+import shutil
+import subprocess
+import sys
+import tempfile
+import xml.etree.ElementTree as ET
+
+sys.dont_write_bytecode = True
+
+from analyze import APP, RUNTIME, SETTINGS, digest, require
+from bundle import CANONICAL_SOURCES, TOOLS, check_application, context_identity, image_identity, manifest, verify_bundle
+
+
+HERE = Path(__file__).resolve().parent
+REPO = HERE.parents[3]
+PROJECT = REPO / "tests/Server" / APP / (APP + ".csproj")
+SUPPORT = PROJECT.with_name("Sdk2LifecycleLateSlopeTests.cs")
+MODULE_SOURCE = HERE.parent / "ArtifactModuleInfo.cs"
+DESIGN = HERE.parent / "LATE_SLOPE.md"
+
+
+def check_source(revision):
+    current = subprocess.check_output(["git", "rev-parse", "HEAD"], cwd=REPO, text=True).strip()
+    require(revision == current, "Publish the specified committed source revision")
+    require(not subprocess.check_output(["git", "status", "--porcelain"], cwd=REPO, text=True).strip(),
+            "The publication checkout must be clean")
+    require('"late-slope-result"' in SUPPORT.read_text(encoding="utf-8"), "The committed canonical test lacks late-slope mode")
+    sources = [*(REPO / name for name in CANONICAL_SOURCES), MODULE_SOURCE, DESIGN,
+               HERE / "prepare.py", HERE / "test_offline.py", *(HERE / name for name in TOOLS)]
+    return {path.relative_to(REPO).as_posix(): digest(path) for path in sources}
+
+
+def managed_modules(build, root):
+    project = ET.Element("Project", Sdk="Microsoft.NET.Sdk")
+    settings = ET.SubElement(project, "PropertyGroup")
+    for key, value in {"OutputType": "Exe", "TargetFramework": "net10.0", "ImplicitUsings": "enable",
+                       "EnableDefaultCompileItems": "false", "TreatWarningsAsErrors": "true"}.items():
+        ET.SubElement(settings, key).text = value
+    ET.SubElement(ET.SubElement(project, "ItemGroup"), "Compile", Include=str(MODULE_SOURCE))
+    path = build / "module-info.csproj"
+    ET.ElementTree(project).write(path, encoding="unicode")
+    with (root / "module-info-build.log").open("w", encoding="utf-8") as log:
+        subprocess.run(["dotnet", "build", str(path), "-c", "Release", "-m:2", "--nologo"],
+                       stdout=log, stderr=subprocess.STDOUT, check=True, timeout=300)
+    command = ["dotnet", str(build / "bin/Release/net10.0/module-info.dll"), str(root), str(root / "app")]
+    return json.loads(subprocess.check_output(command, text=True, timeout=60))
+
+
+def main():
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--output", type=Path, required=True)
+    parser.add_argument("--source-revision", required=True)
+    parser.add_argument("--image", required=True)
+    parser.add_argument("--docker-context", required=True)
+    options = parser.parse_args()
+    sources = check_source(options.source_revision)
+    context = context_identity(options.docker_context)
+    image = image_identity(options.image, context)
+    root = options.output.resolve()
+    root.mkdir(mode=0o700, parents=True, exist_ok=False)
+    with tempfile.TemporaryDirectory(prefix="opcua-late-slope-publish-") as scratch:
+        build = Path(scratch)
+        command = ["dotnet", "publish", str(PROJECT), "-c", "Release", "-r", "linux-arm64",
+                   "--self-contained", "true", f"-p:RuntimeFrameworkVersion={RUNTIME}",
+                   "-p:PublishSingleFile=false", "-p:PublishTrimmed=false", "-p:UseAppHost=true",
+                   "--artifacts-path", str(build / "artifacts"), "-o", str(root / "app"), "-m:2", "--nologo"]
+        with (root / "publish.log").open("w", encoding="utf-8") as log:
+            subprocess.run(command, cwd=REPO, stdout=log, stderr=subprocess.STDOUT, check=True, timeout=900)
+        sdk, gateway = check_application(root / "app")
+        modules = managed_modules(build, root)
+    tools = root / "tools"
+    tools.mkdir()
+    for name in TOOLS:
+        shutil.copy2(HERE / name, tools / name)
+    require(check_source(options.source_revision) == sources, "Committed inputs changed during publication")
+    require(context_identity(options.docker_context) == context and image_identity(options.image, context) == image,
+            "The local Docker context or image changed during publication")
+    receipt = {**SETTINGS, "kind": "bundle", "sourceRevision": options.source_revision, "sourceFiles": sources, "dockerContext": context,
+               "image": options.image, "imageIdentity": image, "preparedUtc": datetime.now(timezone.utc).isoformat(),
+               "publishCommand": command, "sdkLibraries": sdk, "gatewayLibraries": gateway,
+               "managedModules": modules, "files": manifest(root)}
+    path = root / "receipt.json"
+    path.write_text(json.dumps(receipt, indent=2, allow_nan=False) + "\n", encoding="utf-8")
+    sha = digest(path)
+    verify_bundle(root, sha)
+    print(json.dumps({"bundle": str(root), "sourceRevision": options.source_revision,
+                      "receiptSha256": sha, "acceptance": False, "workloadLaunched": False}, indent=2))
+
+
+if __name__ == "__main__":
+    main()
@@ -0,0 +1,383 @@
+#!/usr/bin/env python3
+"""Synthetic evidence controls. No application, Docker, or workload execution."""
+
+import json
+import os
+from pathlib import Path
+import subprocess
+import sys
+import tempfile
+import unittest
+from unittest.mock import patch
+from uuid import UUID
+import xml.etree.ElementTree as ET
+
+sys.dont_write_bytecode = True
+
+import analyze
+import bundle
+
+
+REVISION = "a" * 40
+IMAGE = "sha256:" + "b" * 64
+CONTEXT = {"name": "local-control", "endpoint": "unix:///tmp/control.sock"}
+RUN_ID = "c" * 32
+
+
+def canonical_rows(period=4, rejected=True):
+    yield {"kind": "start", "mode": analyze.MODE, "requestedSeconds": 7200, "minimumFullSeconds": 7200,
+           "runtime": analyze.RUNTIME, "sdk": analyze.SDK, "architecture": "Arm64", "processId": 123,
+           "startupSeconds": 1, "build": REVISION, "acceptance": False, "limits": analyze.BUDGETS.copy()}
+    cycles = 7200 // period
+    baseline = None
+    for cycle in range(1, cycles + 1):
+        elapsed = cycle * period - 1
+        value = dict.fromkeys(analyze.SAMPLE_FIELDS, 0)
+        value.update(Cycle=cycle, Seconds=elapsed, Variables=19, Folders=4, Conditions=1,
+                     CachedCertificateStores=2, Descriptors=190, ManagedBytes=20_000_000 + elapsed * 32,
+                     RssBytes=200_000_000 + elapsed * 128, GcCommittedBytes=48_000_000, JitMethods=76000,
+                     JitIlBytes=11500000, ActiveTimers=12, Threads=24)
+        if cycle == 3:
+            baseline = value.copy()
+        if rejected and cycle == 4:
+            limit = baseline["RssBytes"] + analyze.BUDGETS["rssGrowthBytes"]
+            value["RssBytes"] = limit + 1
+        yield {"kind": "active", "cycle": cycle, "sessions": 2, "subscriptions": 1, "items": 3,
+               "browsePoints": 1, "historyPoints": 6}
+        yield {"kind": "network-stall-recovered", "cycle": cycle, "status": "BadRequestInterrupted", "seconds": 1}
+        yield {"kind": "resource-observation", "resources": value.copy()}
+        if rejected and cycle == 4:
+            yield {"kind": "budget-rejected", "cycle": cycle, "seconds": elapsed + 0.1,
+                   "resources": value.copy(), "baseline": baseline,
+                   "failure": (f"Shouldly.ShouldAssertException: {limit + 1}L\r\n    should be less than or equal to"
+                               f"\r\n{limit}L\r\n    but was not\n   at synthetic fixture")}
+        yield {"kind": "quiescent-budget-rejected" if rejected and cycle >= 4 else "quiescent",
+               "cycleSeconds": period - 1, "resources": value.copy()}
+        if cycle % 4 == 0:
+            yield {"kind": "restart", "cycles": cycle, "restarts": cycle // 4, "seconds": 1, "boundedRestartSeconds": 0.9}
+        yield {"kind": "late-slope-cycle", "cycle": cycle, "restarts": cycle // 4,
+               "elapsedSeconds": cycle * period, "cycleSeconds": period}
+    yield {"kind": "late-slope-result", "mode": analyze.MODE, "cycles": cycles, "restarts": cycles // 4,
+           "cycleSeconds": cycles * period, "shutdownSeconds": 2, "durationCompleted": True,
+           "shutdownVerified": True, "fatalFailure": False, "failureTypes": [], "coldBudgetRejected": rejected, "acceptance": False}
+
+
+def write_json(path, value):
+    path.write_text(json.dumps(value) + "\n", encoding="utf-8")
+
+
+def write_trx(path, names, failure=None):
+    root = ET.Element("TestRun", xmlns=analyze.TRX["t"])
+    results = ET.SubElement(root, "Results")
+    for name in sorted(names):
+        result = ET.SubElement(results, "UnitTestResult", testName=name, outcome="Failed" if failure else "Passed")
+        if failure:
+            output = ET.SubElement(result, "Output")
+            ET.SubElement(ET.SubElement(output, "ErrorInfo"), "Message").text = failure
+    total = len(names)
+    summary = ET.SubElement(root, "ResultSummary")
+    ET.SubElement(summary, "Counters", total=str(total), executed=str(total), passed=str(total - bool(failure)),
+                  failed=str(int(bool(failure))), notExecuted="0", error="0", timeout="0", aborted="0")
+    ET.ElementTree(root).write(path, encoding="utf-8")
+
+
+class RowControls(unittest.TestCase):
+    def test_fixed_windows_and_known_linear_slopes(self):
+        _, samples, completions, rejection, _ = analyze.validate_rows(canonical_rows(), REVISION)
+        report = analyze.describe(samples, completions)
+        self.assertEqual(rejection["cycle"], 4)
+        self.assertEqual(report["allQuiescent"]["sampleCount"], 1800)
+        self.assertEqual(report["allQuiescent"]["RssBytes"]["max"], rejection["resources"]["RssBytes"])
+        self.assertEqual([row["windowSeconds"] for row in report["windows"]], [[1800, 2400], [3600, 4200], [6600, 7200]])
+        self.assertEqual([row["sampleCount"] for row in report["windows"]], [150, 150, 150])
+        self.assertAlmostEqual(report["leastSquares"]["rss"], 128 * 3600)
+        self.assertAlmostEqual(report["leastSquares"]["retainedManaged"], 32 * 3600)
+        self.assertEqual(report["leastSquares"]["sampleCount"], 1350)
+        self.assertTrue(report["workCompletedBy1800Seconds"]["earlierSaturationWorkloadReached"])
+
+    def test_slopes_remain_descriptive_below_earlier_workload(self):
+        _, samples, completions, rejection, _ = analyze.validate_rows(canonical_rows(25, rejected=False), REVISION)
+        report = analyze.describe(samples, completions)
+        self.assertIsNone(rejection)
+        work = report["workCompletedBy1800Seconds"]
+        self.assertEqual((work["cycles"], work["restarts"]), (72, 18))
+        self.assertFalse(work["earlierSaturationWorkloadReached"])
+        self.assertIn("attribution withheld", work["interpretation"])
+        self.assertAlmostEqual(report["leastSquares"]["rss"], 128 * 3600)
+
+    def test_wrong_mode_duration_runtime_or_startup_are_rejected(self):
+        for field, value in (("mode", "full"), ("requestedSeconds", 90), ("runtime", "10.0.3"),
+                             ("architecture", "X64"), ("sdk", "2.0.0"), ("startupSeconds", 121), ("acceptance", True)):
+            with self.subTest(field=field):
+                rows = list(canonical_rows(25))
+                rows[0][field] = value
+                with self.assertRaises(ValueError):
+                    analyze.validate_rows(rows, REVISION)
+
+    def test_wrong_final_state_never_yields_a_report(self):
+        for field, value in (("kind", "complete"), ("cycleSeconds", 7199.9), ("cycleSeconds", 7231),
+                             ("shutdownVerified", False), ("durationCompleted", False), ("fatalFailure", True),
+                             ("shutdownSeconds", 61), ("coldBudgetRejected", False), ("acceptance", True), ("restarts", 1)):
+            with self.subTest(field=field, value=value):
+                rows = list(canonical_rows(25))
+                rows[-1][field] = value
+                with self.assertRaises(ValueError):
+                    analyze.validate_rows(rows, REVISION)
+
+    def test_duplicate_reordered_missing_or_malformed_samples(self):
+        def change(rows, position, field, value):
+            rows[position]["resources"][field] = value
+
+        edits = (
+            lambda rows: rows.insert(4, rows[3]),
+            lambda rows: rows.pop(3),
+            lambda rows: change(rows, 8, "Cycle", 1),
+            lambda rows: change(rows, 8, "Seconds", rows[3]["resources"]["Seconds"]),
+            lambda rows: change(rows, 3, "RssBytes", -1),
+            lambda rows: change(rows, 3, "ManagedBytes", float("inf")),
+            lambda rows: change(rows, 3, "PkiDescriptors", 1),
+            lambda rows: rows.append({"kind": "complete"}),
+        )
+        for edit in edits:
+            rows = list(canonical_rows(25))
+            edit(rows)
+            with self.assertRaises(ValueError):
+                analyze.validate_rows(rows, REVISION)
+
+    def test_first_rejection_cannot_disappear_or_change_baseline(self):
+        rows = list(canonical_rows(25))
+        first = next(i for i, row in enumerate(rows) if row["kind"] == "budget-rejected")
+        rows[first]["baseline"]["RssBytes"] -= 1
+        with self.assertRaisesRegex(ValueError, "baseline"):
+            analyze.validate_rows(rows, REVISION)
+        rows = list(canonical_rows(25))
+        later = next(row for row in rows if row["kind"] == "quiescent-budget-rejected" and row["resources"]["Cycle"] == 5)
+        later["kind"] = "quiescent"
+        with self.assertRaisesRegex(ValueError, "quiescent-budget-rejected"):
+            analyze.validate_rows(rows, REVISION)
+        rows = list(canonical_rows(25))
+        rows.pop(first)
+        with self.assertRaisesRegex(ValueError, "budget-rejected"):
+            analyze.validate_rows(rows, REVISION)
+
+    def test_bounded_restart_and_complete_cycle_durations(self):
+        for kind, field, value, expected in (
+            ("restart", "boundedRestartSeconds", 25.1, "bounded restart duration"),
+            ("restart", "seconds", 31, "combined restart duration"),
+            ("late-slope-cycle", "cycleSeconds", 31, "complete cycle duration"),
+        ):
+            with self.subTest(field=field):
+                rows = list(canonical_rows(25))
+                row = next(row for row in rows if row["kind"] == kind)
+                row[field] = value
+                with self.assertRaisesRegex(ValueError, expected):
+                    analyze.validate_rows(rows, REVISION)
+
+    def test_missing_windows_edges_and_internal_or_between_window_gaps(self):
+        _, samples, completions, _, _ = analyze.validate_rows(canonical_rows(25), REVISION)
+        for begin, end in ((3600, 4200), (1800, 1880), (7110, 7200), (1900, 1980), (2900, 2980)):
+            with self.subTest(begin=begin, end=end):
+                incomplete = [sample for sample in samples if not begin <= sample["Seconds"] <= end]
+                with self.assertRaises(ValueError):
+                    analyze.describe(incomplete, completions)
+
+    def test_jsonl_rejects_duplicate_keys_nonfinite_literals_and_truncation(self):
+        with tempfile.TemporaryDirectory() as temporary:
+            path = Path(temporary) / "samples.jsonl"
+            for content in ('{"kind":"start","kind":"complete"}\n', '{"seconds":NaN}\n', '{"kind":"start"}'):
+                path.write_text(content, encoding="utf-8")
+                with self.assertRaises(ValueError):
+                    list(analyze.read_rows(path))
+
+
+class ReceiptControls(unittest.TestCase):
+    def setUp(self):
+        temporary = tempfile.TemporaryDirectory()
+        self.addCleanup(temporary.cleanup)
+        self.root = Path(temporary.name).resolve()
+        self.run = self.root / "run"
+        self.bundle_receipt = self.root / "bundle.json"
+        (self.run / "results").mkdir(parents=True)
+        self.build_evidence()
+
+    def build_evidence(self, rejected=True):
+        self.bundle = {**analyze.SETTINGS, "kind": "bundle", "sourceRevision": REVISION, "image": IMAGE,
+                       "dockerContext": CONTEXT, "files": {"app/libcoreclr.so": "d" * 64}}
+        write_json(self.bundle_receipt, self.bundle)
+        self.bundle_hash = analyze.digest(self.bundle_receipt)
+        self.provenance = {"kind": "launch", "protocol": analyze.PROTOCOL, "acceptance": False,
+                           "sourceRevision": REVISION, "image": IMAGE, "dockerContext": CONTEXT, "runId": RUN_ID,
+                           "bundleReceiptSha256": self.bundle_hash, "coreClrSha256": "d" * 64,
+                           "containerTimeoutSeconds": 7440, "bundlePath": str(self.root / "bundle"), "outputPath": str(self.run)}
+        self.provenance["command"] = analyze.container_command(self.root / "bundle", self.run, self.bundle, RUN_ID)
+        write_json(self.run / "launch.json", self.provenance)
+        rows = list(canonical_rows(25, rejected))
+        (self.run / "resources.jsonl").write_text("".join(json.dumps(row) + "\n" for row in rows), encoding="utf-8")
+        status = int(rejected)
+        for filename in ("test-exit-status.txt", "container-exit-status.txt"):
+            (self.run / filename).write_text(f"{status}\n", encoding="ascii")
+        (self.run / "target-runtime.txt").write_text(
+            f"targetPid=123\nactualExecutable=/app/{analyze.APP}\nmappedCoreClr=/app/libcoreclr.so\ncoreClrSha256={'d' * 64}\n",
+            encoding="ascii")
+        for filename in ("container.log", "controller-runtime.txt", "descriptor-controls.log", "test.log"):
+            (self.run / filename).write_text("Synthetic control fixture\n", encoding="utf-8")
+        write_trx(self.run / "results/descriptor-controls.trx", analyze.DESCRIPTOR_TESTS)
+        failure = next((row["failure"] for row in rows if row["kind"] == "budget-rejected"), None)
+        message = analyze.assertion_message(failure).replace("Exception:", "Exception :", 1) if failure else None
+        write_trx(self.run / "results/late-slope.trx", {analyze.TEST_NAME}, message)
+        actual = analyze.inspect_run(self.run, self.provenance, 7205)
+        self.receipt = {**self.provenance, "kind": "run", "diagnosticCompleted": True, "containerRemoved": True,
+                        "wallSeconds": 7205, "launcherExitStatus": status, **{key: actual[key] for key in (
+                            "testExitStatus", "containerExitStatus", "descriptorTests", "experimentTests", "diagnosticOutcome")}}
+        self.refresh_receipt()
+
+    def refresh_receipt(self):
+        self.receipt["evidenceSha256"] = analyze.evidence_hashes(self.run)
+        write_json(self.run / "run-receipt.json", self.receipt)
+
+    def report(self):
+        return analyze.analyze_run(self.run, self.bundle_receipt, self.bundle_hash)
+
+    def test_cold_failure_and_zero_exit_are_separate_diagnostic_results(self):
+        cold = self.report()
+        self.assertEqual(cold["testExitStatus"], 1)
+        self.assertFalse(cold["acceptance"])
+        self.assertEqual(cold["firstColdRejection"]["exceededFields"], ["RssBytes"])
+        self.build_evidence(rejected=False)
+        within_cap = self.report()
+        self.assertEqual(within_cap["testExitStatus"], 0)
+        self.assertFalse(within_cap["acceptance"])
+        self.assertIsNone(within_cap["firstColdRejection"])
+
+    def test_changed_hash_or_bad_run_receipt_rejected(self):
+        (self.run / "test.log").write_text("Changed evidence\n", encoding="utf-8")
+        with self.assertRaisesRegex(ValueError, "evidence hashes"):
+            self.report()
+        for key, value in (("launcherExitStatus", 0), ("containerRemoved", False), ("diagnosticCompleted", False),
+                           ("wallSeconds", 7441), ("bundleReceiptSha256", "f" * 64)):
+            with self.subTest(key=key):
+                self.build_evidence()
+                self.receipt[key] = value
+                self.refresh_receipt()
+                with self.assertRaises(ValueError):
+                    self.report()
+
+    def test_rehashed_exit_runtime_and_trx_cannot_mask_incomplete_evidence(self):
+        for filename, contents, pattern in (
+            ("test-exit-status.txt", "0\n", "test/container exit"),
+            ("target-runtime.txt", "targetPid=122\n", "mapped CLR"),
+        ):
+            with self.subTest(filename=filename):
+                self.build_evidence()
+                (self.run / filename).write_text(contents, encoding="ascii")
+                self.refresh_receipt()
+                with self.assertRaisesRegex(ValueError, pattern):
+                    self.report()
+        self.build_evidence()
+        write_trx(self.run / "results/late-slope.trx", {analyze.TEST_NAME}, "Shouldly.ShouldAssertException: wrong assertion")
+        self.refresh_receipt()
+        with self.assertRaisesRegex(ValueError, "first cold-cap assertion"):
+            self.report()
+        write_trx(self.run / "results/descriptor-controls.trx", {"SkippedOrWrongControl"})
+        self.refresh_receipt()
+        with self.assertRaisesRegex(ValueError, "descriptor-controls.trx"):
+            self.report()
+
+    def test_recorded_command_must_use_the_same_local_isolation(self):
+        for isolation in ("network", "read-only rootfs"):
+            with self.subTest(isolation=isolation):
+                self.build_evidence()
+                invocation = self.provenance["command"].copy()
+                if isolation == "network":
+                    invocation[invocation.index("--network") + 1] = "host"
+                else:
+                    invocation.remove("--read-only")
+                self.provenance["command"] = self.receipt["command"] = invocation
+                write_json(self.run / "launch.json", self.provenance)
+                self.refresh_receipt()
+                with self.assertRaisesRegex(ValueError, "container invocation"):
+                    self.report()
+
+
+class BundleControls(unittest.TestCase):
+    def test_mapped_runtime_requires_the_mounted_inode_and_live_file(self):
+        script = Path(__file__).with_name("container.sh")
+        with tempfile.TemporaryDirectory() as temporary:
+            maps = Path(temporary) / "maps"
+            live = "1000-2000 r-xp 00000000 00:01 42 /app/libcoreclr.so"
+            for line, expected in ((live, 0), (live + " (deleted)", 1), (live.replace(" 42 ", " 43 "), 1),
+                                   (live + "\n" + live.replace("/app/", "/different/"), 1)):
+                with self.subTest(line=line):
+                    maps.write_text(line + "\n", encoding="ascii")
+                    result = subprocess.run(["bash", "-c", 'source "$1"\nverify_mapped_coreclr "$2" "$3"',
+                                             "map-control", str(script), str(maps), "42 /app/libcoreclr.so"],
+                                            capture_output=True, text=True, timeout=5)
+                    self.assertEqual(result.returncode, expected, result.stderr)
+
+    def test_remote_override_fails_before_asking_docker(self):
+        with patch.dict(os.environ, {"DOCKER_HOST": "tcp://example.invalid:2375"}), patch("bundle.subprocess.check_output") as docker:
+            with self.assertRaisesRegex(ValueError, "override"):
+                bundle.context_identity("local-control")
+            docker.assert_not_called()
+
+    def test_bundle_file_set_runtime_and_package_graph(self):
+        with tempfile.TemporaryDirectory() as temporary:
+            root = Path(temporary).resolve()
+            app = root / "app"
+            app.mkdir()
+            elf = b"\x7fELF\x02\x01" + b"\x00" * 12 + b"\xb7\x00"
+            (app / analyze.APP).write_bytes(elf)
+            (app / analyze.APP).chmod(0o700)
+            (app / "libcoreclr.so").write_bytes(elf)
+            runtime = app / f"{analyze.APP}.runtimeconfig.json"
+            write_json(runtime, {"runtimeOptions": {"tfm": "net10.0", "includedFrameworks": [
+                {"name": "Microsoft.NETCore.App", "version": analyze.RUNTIME}]}})
+            dependencies = app / f"{analyze.APP}.deps.json"
+            write_json(dependencies, {"runtimeTarget": {"name": ".NETCoreApp,Version=v10.0/linux-arm64"},
+                                      "libraries": {key: {} for key in bundle.SDK_PACKAGES | bundle.GATEWAY_PACKAGES}})
+            modules = []
+            for index, assembly in enumerate((analyze.APP, "Opc.Ua.Server", "System.Private.CoreLib",
+                                               "ZB.MOM.WW.HistorianGateway.Client", "ZB.MOM.WW.HistorianGateway.Contracts"), 1):
+                filename = f"{assembly}.dll"
+                (app / filename).write_text("Synthetic metadata fixture\n", encoding="utf-8")
+                modules.append({"file": f"app/{filename}", "assembly": assembly, "version": "1.0.0.0", "mvid": str(UUID(int=index))})
+            tools = root / "tools"
+            tools.mkdir()
+            for name in bundle.TOOLS:
+                (tools / name).write_text("Synthetic tool fixture\n", encoding="utf-8")
+            receipt = {**analyze.SETTINGS, "kind": "bundle", "sourceRevision": REVISION, "image": IMAGE,
+                       "imageIdentity": {"id": IMAGE, "os": "linux", "architecture": "arm64"}, "dockerContext": CONTEXT,
+                       "sdkLibraries": sorted(bundle.SDK_PACKAGES), "gatewayLibraries": sorted(bundle.GATEWAY_PACKAGES),
+                       "managedModules": modules, "sourceFiles": {**dict.fromkeys(bundle.CANONICAL_SOURCES, "e" * 64), **{
+                           f"scripts/validation/opcua-lifecycle-memory/late-slope/{name}": analyze.digest(tools / name)
+                           for name in bundle.TOOLS}}, "files": bundle.manifest(root)}
+            path = root / "receipt.json"
+            write_json(path, receipt)
+            approved = analyze.digest(path)
+            bundle.verify_bundle(root, approved)
+            extra = app / "extra"
+            extra.touch()
+            with self.assertRaisesRegex(ValueError, "file set"):
+                bundle.verify_bundle(root, approved)
+            extra.unlink()
+            extra.symlink_to(app / "libcoreclr.so")
+            with self.assertRaisesRegex(ValueError, "artifact type"):
+                bundle.verify_bundle(root, approved)
+            extra.unlink()
+            write_json(runtime, {"runtimeOptions": {"tfm": "net10.0", "includedFrameworks": [
+                {"name": "Microsoft.NETCore.App", "version": "10.0.3"}]}})
+            receipt["files"] = bundle.manifest(root)
+            write_json(path, receipt)
+            with self.assertRaisesRegex(ValueError, "self-contained application runtime"):
+                bundle.verify_bundle(root, analyze.digest(path))
+            write_json(runtime, {"runtimeOptions": {"tfm": "net10.0", "includedFrameworks": [
+                {"name": "Microsoft.NETCore.App", "version": analyze.RUNTIME}]}})
+            write_json(dependencies, {"runtimeTarget": {"name": ".NETCoreApp,Version=v10.0/linux-arm64"}, "libraries": {}})
+            receipt["files"] = bundle.manifest(root)
+            write_json(path, receipt)
+            with self.assertRaisesRegex(ValueError, "dependency|graph"):
+                bundle.verify_bundle(root, analyze.digest(path))
+
+
+if __name__ == "__main__":
+    unittest.main(verbosity=2)
@@ -175,6 +175,9 @@ public sealed class OpcUaApplicationHost : IAsyncDisposable
    private StandardServer? _server;
    private readonly ITelemetryContext _telemetry = DefaultTelemetry.Create(_ => { });
    private readonly List<IUserTokenAuthenticator> _identityAuthenticators = [];
+    private readonly object _lifetimeLock = new();
+    private readonly TaskCompletionSource _disposeCompletion = new(TaskCreationOptions.RunContinuationsAsynchronously);
+    private int _disposeStarted;

    /// <summary>Initializes a new instance of the OPC UA application host.</summary>
    /// <param name="options">The host configuration options.</param>
@@ -202,18 +205,21 @@ public sealed class OpcUaApplicationHost : IAsyncDisposable
    /// <returns>A task that represents the asynchronous operation.</returns>
    public async Task StartAsync(StandardServer server, CancellationToken cancellationToken)
    {
-        if (server is not OtOpcUaSdkServer sdkServer)
-            throw new ArgumentException("The server must support application authentication before its listeners open.", nameof(server));
-        _server = server;
-        // ApplicationInstance receives the host's ITelemetryContext. DefaultTelemetry.Create wires
-        // the SDK's internal logging; an empty builder keeps
-        // the SDK's trace off our ILogger (the host keeps its own _logger) — sufficient for the bootstrap.
-        _application = new ApplicationInstance(_telemetry)
+        OtOpcUaSdkServer sdkServer;
+        lock (_lifetimeLock)
        {
-            ApplicationName = _options.ApplicationName,
-            ApplicationType = ApplicationType.Server,
-            ConfigSectionName = "OtOpcUa",
-        };
+            ObjectDisposedException.ThrowIf(Volatile.Read(ref _disposeStarted) != 0, this);
+            sdkServer = server as OtOpcUaSdkServer
+                ?? throw new ArgumentException("The server must support application authentication before its listeners open.", nameof(server));
+            _server = server;
+            // The empty telemetry builder keeps SDK tracing separate from the host's ILogger.
+            _application = new ApplicationInstance(_telemetry)
+            {
+                ApplicationName = _options.ApplicationName,
+                ApplicationType = ApplicationType.Server,
+                ConfigSectionName = "OtOpcUa",
+            };
+        }

        try
        {
@@ -640,16 +646,63 @@ public sealed class OpcUaApplicationHost : IAsyncDisposable

    /// <summary>Disposes the application host and cleans up resources.</summary>
    /// <returns>A task that represents the asynchronous operation.</returns>
-    public async ValueTask DisposeAsync()
+    public ValueTask DisposeAsync()
    {
-        // Authenticators remain installed until the SDK closes every listener and disposes its registry.
-        if (_application is { } application)
+        if (Interlocked.CompareExchange(ref _disposeStarted, 1, 0) == 0)
+            _ = CompleteDisposalAsync();
+        return new ValueTask(_disposeCompletion.Task);
+    }
+
+    private async Task CompleteDisposalAsync()
+    {
+        Exception? failure = null;
+        try
        {
-            _application = null;
-            try { await application.DisposeAsync().ConfigureAwait(false); }
-            catch (Exception ex) { _logger.LogWarning(ex, "OpcUaApplicationHost: disposal threw"); }
+            ApplicationInstance? application;
+            StandardServer? server;
+            bool adopted;
+            lock (_lifetimeLock)
+            {
+                application = _application;
+                server = _server;
+                // SDK2 adopts the server only when ApplicationInstance.StartAsync is reached.
+                adopted = server is not null && ReferenceEquals(application?.Server, server);
+                _application = null;
+            }
+            try
+            {
+                if (application is not null)
+                {
+                    try { await application.DisposeAsync().ConfigureAwait(false); }
+                    catch (Exception ex) { _logger.LogWarning(ex, "OpcUaApplicationHost: disposal threw"); }
+                }
+            }
+            finally
+            {
+                // Failed configuration/certificate work may precede the application's ownership handoff.
+                // An adopted server remains solely owned by ApplicationInstance.DisposeAsync.
+                if (!adopted)
+                {
+                    try { if (server is not null) await server.DisposeAsync().ConfigureAwait(false); }
+                    catch (Exception ex) { _logger.LogWarning(ex, "OpcUaApplicationHost: unadopted server disposal threw"); }
+                }
+            }
+        }
+        catch (Exception error) { failure = error; }
+        finally
+        {
+            try
+            {
+                // Only the cleanup owner clears adapters after every listener and registry has closed.
+                _identityAuthenticators.Clear();
+                _server = null;
+            }
+            catch (Exception error) { failure = failure is null ? error : new AggregateException(failure, error); }
+            finally
+            {
+                if (failure is null) _disposeCompletion.TrySetResult();
+                else _disposeCompletion.TrySetException(failure);
+            }
        }
-        _identityAuthenticators.Clear();
-        _server = null;
    }
 }
@@ -0,0 +1,99 @@
+using Opc.Ua;
+using Opc.Ua.Configuration;
+using Opc.Ua.Server;
+using Shouldly;
+using Xunit;
+using ClientSession = Opc.Ua.Client.ISession;
+
+namespace ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests;
+
+public sealed partial class Sdk2LifecycleSoakTests
+{
+    /// <summary>Concurrent host disposal awaits real encrypted-session shutdown before clearing ownership.</summary>
+    [Fact]
+    public async Task Concurrent_host_disposal_waits_for_the_actual_encrypted_session_to_close()
+    {
+        var ct = TestContext.Current.CancellationToken;
+        var clientRoot = Path.Combine(Path.GetTempPath(), $"otua-dispose-client-{Guid.NewGuid():N}");
+        ApplicationInstance? application = null;
+        ServerEventRig? rig = null;
+        ClientSession? client = null;
+        ISessionManager? sessionManager = null;
+        SessionEventHandler? closing = null;
+        using var release = new ManualResetEventSlim(false);
+        var entered = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
+        var calls = new List<Task>();
+        var timedOut = 0;
+        var callbackCount = 0;
+        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("SDK2ConcurrentDisposal", encrypted: true, ct);
+            Configure(rig);
+            client = await ConnectAsync(rig, config, null, ct);
+            sessionManager = rig.Server.CurrentInstance.SessionManager;
+            sessionManager.GetSessions().Count.ShouldBe(1);
+            var sessionId = client.SessionId;
+            closing = (session, reason) =>
+            {
+                if (session.Id != sessionId || reason != SessionEventReason.Closing) return;
+                Interlocked.Increment(ref callbackCount);
+                entered.TrySetResult();
+                // SDK swallows event exceptions; keep an explicit failure flag for this bounded barrier.
+                if (!release.Wait(TimeSpan.FromSeconds(15))) Interlocked.Exchange(ref timedOut, 1);
+            };
+            sessionManager.SessionClosing += closing;
+            var host = RequiredField<OpcUaApplicationHost>(rig, "_host");
+            var manager = rig.Server.CertificateManager!;
+            calls.Add(Task.Run(async () => await host.DisposeAsync(), ct));
+            await entered.Task.WaitAsync(TimeSpan.FromSeconds(10), ct);
+            for (var i = 0; i < 3; i++) calls.Add(host.DisposeAsync().AsTask());
+            calls.ShouldAllBe(call => !call.IsCompleted, "every caller must await the actual shutdown owner");
+            host.Server.ShouldBeSameAs(rig.Server);
+            RequiredField<bool>(manager, "m_disposed").ShouldBeFalse();
+            release.Set();
+            await Task.WhenAll(calls).WaitAsync(TimeSpan.FromSeconds(20), ct);
+            Volatile.Read(ref timedOut).ShouldBe(0);
+            Volatile.Read(ref callbackCount).ShouldBe(1);
+            host.Server.ShouldBeNull();
+            host.ApplicationInstance.ShouldBeNull();
+            RequiredField<bool>(manager, "m_disposed").ShouldBeTrue();
+            CertificateCores(manager).Sum(CachedStores).ShouldBe(0);
+        }
+        finally
+        {
+            // Removing a handler takes the SDK's event lock: release before awaiting or unsubscribing.
+            release.Set();
+            try { await Task.WhenAll(calls).WaitAsync(TimeSpan.FromSeconds(20), CancellationToken.None); }
+            finally
+            {
+                try { if (sessionManager is not null && closing is not null) sessionManager.SessionClosing -= closing; }
+                finally
+                {
+                    try { client?.Dispose(); }
+                    finally
+                    {
+                        try { if (rig is not null) await rig.DisposeAsync().AsTask().WaitAsync(TimeSpan.FromSeconds(20), CancellationToken.None); }
+                        finally
+                        {
+                            try { if (application is not null) await application.DisposeAsync().AsTask().WaitAsync(TimeSpan.FromSeconds(20), CancellationToken.None); }
+                            finally { if (Directory.Exists(clientRoot)) Directory.Delete(clientRoot, true); }
+                        }
+                    }
+                }
+            }
+        }
+    }
+}
@@ -0,0 +1,277 @@
+using Microsoft.Extensions.Logging.Abstractions;
+using Opc.Ua;
+using Opc.Ua.Security.Certificates;
+using Opc.Ua.Server;
+using Shouldly;
+using Xunit;
+
+namespace ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests;
+
+/// <summary>Verifies host ownership using the real sealed SDK2 server and public manager factories.</summary>
+public sealed class OpcUaApplicationHostDisposalTests : IDisposable
+{
+    private static CancellationToken Ct => TestContext.Current.CancellationToken;
+    private readonly string _pki = Path.Combine(Path.GetTempPath(), $"otopcua-sdk2-disposal-{Guid.NewGuid():N}");
+
+    /// <summary>Normal and repeated disposal complete one manager shutdown and terminal server cleanup.</summary>
+    [Fact]
+    public async Task Host_terminally_disposes_its_booted_server_once()
+    {
+        var factory = new RecordingSessionManagerFactory();
+        var (host, server) = await BootAsync(factory);
+        var manager = factory.Created.ShouldNotBeNull();
+        try
+        {
+            await host.DisposeAsync();
+            manager.ShutdownCalls.ShouldBe(1);
+            manager.DisposalCalls.ShouldBe(1);
+            await RequireTerminalServerAsync(server);
+
+            await host.DisposeAsync();
+            manager.ShutdownCalls.ShouldBe(1);
+            manager.DisposalCalls.ShouldBe(1);
+            await RequireTerminalServerAsync(server);
+        }
+        finally { await CleanupAsync(host, server); }
+    }
+
+    /// <summary>Manager disposal during Stop alone does not satisfy the terminal server assertion.</summary>
+    [Fact]
+    public async Task Stop_only_is_distinct_from_terminal_server_disposal()
+    {
+        var factory = new RecordingSessionManagerFactory();
+        var (host, server) = await BootAsync(factory);
+        var manager = factory.Created.ShouldNotBeNull();
+        try
+        {
+            await server.StopAsync(Ct);
+            manager.ShutdownCalls.ShouldBe(1);
+            manager.DisposalCalls.ShouldBe(1);
+            var probe = new NonCreatingNodeManagerFactory();
+            var stopped = await Should.ThrowAsync<InvalidOperationException>(() =>
+                server.NodeManagerLifecycle.AddAsync(probe, callerContext: null, Ct).AsTask());
+            stopped.GetType().ShouldBe(typeof(InvalidOperationException));
+            probe.CreateCalls.ShouldBe(0);
+
+            await host.DisposeAsync();
+            await RequireTerminalServerAsync(server);
+            manager.ShutdownCalls.ShouldBe(1);
+            manager.DisposalCalls.ShouldBe(1);
+        }
+        finally { await CleanupAsync(host, server); }
+    }
+
+    /// <summary>A manager shutdown failure is recorded by the SDK while terminal cleanup still completes.</summary>
+    [Fact]
+    public async Task Shutdown_stage_failure_still_terminally_disposes_the_server()
+    {
+        var factory = new RecordingSessionManagerFactory { ThrowAfterShutdown = true };
+        var (host, server) = await BootAsync(factory);
+        var manager = factory.Created.ShouldNotBeNull();
+        try
+        {
+            await host.DisposeAsync();
+            manager.ShutdownCalls.ShouldBe(1);
+            manager.DisposalCalls.ShouldBe(1);
+            ServiceResult.IsBad(server.ServerError).ShouldBeTrue();
+            server.ServerError.ToString().ShouldContain(RecordingSessionManager.ShutdownFailure);
+            await RequireTerminalServerAsync(server);
+        }
+        finally { await CleanupAsync(host, server); }
+    }
+
+    /// <summary>A factory failure after address-space initialization leaves the real server owned for cleanup.</summary>
+    [Fact]
+    public async Task Partial_startup_failure_does_not_strand_the_server()
+    {
+        var factory = new RecordingSessionManagerFactory { ThrowDuringCreate = true };
+        var host = NewHost(OpcUaHostStart.AllocateFreePort());
+        var server = new OtOpcUaSdkServer { SessionManagerFactory = factory };
+        try
+        {
+            var failure = await Should.ThrowAsync<ServiceResultException>(() => host.StartAsync(server, Ct));
+            failure.StatusCode.ShouldBe(StatusCodes.BadInternalError);
+            failure.InnerResult.ShouldNotBeNull().LocalizedText.Text.ShouldNotBeNull()
+                .ShouldContain(RecordingSessionManagerFactory.StartupFailure);
+            factory.CreateCalls.ShouldBe(1);
+            factory.Created.ShouldBeNull("the sentinel fails before constructing an unowned manager");
+            host.ApplicationInstance.ShouldBeNull();
+            host.Server.ShouldBeNull();
+            await RequireTerminalServerAsync(server);
+            await host.DisposeAsync();
+            await RequireTerminalServerAsync(server);
+        }
+        finally { await CleanupAsync(host, server); }
+    }
+
+    /// <summary>Configuration cancellation before ApplicationInstance adopts the server still releases its host owner.</summary>
+    [Fact]
+    public async Task Canceled_start_before_application_adoption_terminally_disposes_the_server()
+    {
+        var factory = new RecordingSessionManagerFactory();
+        var host = NewHost(OpcUaHostStart.AllocateFreePort());
+        var server = new OtOpcUaSdkServer { SessionManagerFactory = factory };
+        using var cancellation = new CancellationTokenSource();
+        cancellation.Cancel();
+        try
+        {
+            await Should.ThrowAsync<OperationCanceledException>(() => host.StartAsync(server, cancellation.Token));
+            factory.CreateCalls.ShouldBe(0);
+            factory.Created.ShouldBeNull();
+            host.ApplicationInstance.ShouldBeNull();
+            host.Server.ShouldBeNull();
+            await host.DisposeAsync();
+            await RequireTerminalServerAsync(server);
+        }
+        finally { await CleanupAsync(host, server); }
+    }
+
+    /// <summary>Every concurrent host disposer waits until the one actual manager shutdown is released.</summary>
+    [Fact]
+    public async Task Concurrent_disposers_share_completion_and_one_cleanup_owner()
+    {
+        var factory = new RecordingSessionManagerFactory();
+        var (host, server) = await BootAsync(factory);
+        var manager = factory.Created.ShouldNotBeNull();
+        manager.BlockShutdown = true;
+        var calls = new List<Task>();
+        try
+        {
+            calls.Add(Task.Run(async () => await host.DisposeAsync(), Ct));
+            await manager.ShutdownEntered.Task.WaitAsync(TimeSpan.FromSeconds(10), Ct);
+            for (var i = 0; i < 3; i++) calls.Add(host.DisposeAsync().AsTask());
+            calls.ShouldAllBe(task => !task.IsCompleted);
+            manager.ShutdownCalls.ShouldBe(1);
+            manager.DisposalCalls.ShouldBe(0);
+            manager.ReleaseShutdown.TrySetResult();
+            await Task.WhenAll(calls).WaitAsync(TimeSpan.FromSeconds(20), Ct);
+            manager.ShutdownCalls.ShouldBe(1);
+            manager.DisposalCalls.ShouldBe(1);
+            await RequireTerminalServerAsync(server);
+        }
+        finally
+        {
+            manager.ReleaseShutdown.TrySetResult();
+            try { await Task.WhenAll(calls).WaitAsync(TimeSpan.FromSeconds(20), CancellationToken.None); }
+            finally { await CleanupAsync(host, server); }
+        }
+    }
+
+    /// <summary>A terminal host rejects a new real server before invoking its manager factory.</summary>
+    [Fact]
+    public async Task Disposed_host_rejects_start_without_taking_server_ownership()
+    {
+        var factory = new RecordingSessionManagerFactory();
+        var host = NewHost(OpcUaHostStart.AllocateFreePort());
+        var server = new OtOpcUaSdkServer { SessionManagerFactory = factory };
+        try
+        {
+            await host.DisposeAsync();
+            await Should.ThrowAsync<ObjectDisposedException>(() => host.StartAsync(server, Ct));
+            factory.CreateCalls.ShouldBe(0, "the rejected server still belongs to its caller");
+            host.ApplicationInstance.ShouldBeNull();
+            host.Server.ShouldBeNull();
+        }
+        finally { await CleanupAsync(host, server); }
+    }
+
+    private Task<(OpcUaApplicationHost Host, OtOpcUaSdkServer Server)> BootAsync(RecordingSessionManagerFactory factory) =>
+        OpcUaHostStart.StartAsync(NewHost, () => new OtOpcUaSdkServer { SessionManagerFactory = factory }, Ct);
+
+    private OpcUaApplicationHost NewHost(int port) => new(new OpcUaApplicationHostOptions
+    {
+        ApplicationName = "OtOpcUa.SDK2.Disposal", ApplicationUri = "urn:OtOpcUa.SDK2.Disposal",
+        PublicHostname = "localhost", OpcUaPort = port, PkiStoreRoot = _pki, AllowAnonymous = false,
+    }, NullLogger<OpcUaApplicationHost>.Instance);
+
+    private static async Task RequireTerminalServerAsync(OtOpcUaSdkServer server)
+    {
+        // The public lifecycle checks terminal disposal before considering or invoking this factory.
+        var probe = new NonCreatingNodeManagerFactory();
+        var disposed = await Should.ThrowAsync<ObjectDisposedException>(() =>
+            server.NodeManagerLifecycle.AddAsync(probe, callerContext: null, CancellationToken.None).AsTask());
+        disposed.GetType().ShouldBe(typeof(ObjectDisposedException));
+        probe.CreateCalls.ShouldBe(0);
+    }
+
+    private static async Task CleanupAsync(OpcUaApplicationHost host, OtOpcUaSdkServer server)
+    {
+        try { await host.DisposeAsync().AsTask().WaitAsync(TimeSpan.FromSeconds(20), CancellationToken.None); }
+        finally
+        {
+            // RED assertions still release the actual test-owned server; never dispose a manager manually.
+            await server.DisposeAsync().AsTask().WaitAsync(TimeSpan.FromSeconds(20), CancellationToken.None);
+        }
+    }
+
+    private sealed class RecordingSessionManagerFactory : ISessionManagerFactory
+    {
+        internal const string StartupFailure = "controlled SDK2 partial startup";
+        private int _createCalls;
+        internal int CreateCalls => Volatile.Read(ref _createCalls);
+        internal RecordingSessionManager? Created { get; private set; }
+        internal bool ThrowDuringCreate { get; init; }
+        internal bool ThrowAfterShutdown { get; init; }
+
+        public ISessionManager Create(IServerInternal server, ApplicationConfiguration configuration,
+            TimeProvider timeProvider, Func<string, Certificate?> serverCertificateProvider)
+        {
+            Interlocked.Increment(ref _createCalls);
+            if (ThrowDuringCreate) throw new InvalidOperationException(StartupFailure);
+            return Created = new RecordingSessionManager(server, configuration, timeProvider)
+            {
+                ThrowAfterShutdown = ThrowAfterShutdown,
+            };
+        }
+    }
+
+    private sealed class RecordingSessionManager(IServerInternal server, ApplicationConfiguration configuration, TimeProvider timeProvider)
+        : SessionManager(server, configuration, timeProvider)
+    {
+        internal const string ShutdownFailure = "controlled SDK2 manager shutdown failure";
+        private int _shutdownCalls;
+        private int _disposalCalls;
+        internal int ShutdownCalls => Volatile.Read(ref _shutdownCalls);
+        internal int DisposalCalls => Volatile.Read(ref _disposalCalls);
+        internal bool ThrowAfterShutdown { get; init; }
+        internal bool BlockShutdown { get; set; }
+        internal TaskCompletionSource ShutdownEntered { get; } = new(TaskCreationOptions.RunContinuationsAsynchronously);
+        internal TaskCompletionSource ReleaseShutdown { get; } = new(TaskCreationOptions.RunContinuationsAsynchronously);
+
+        public override async ValueTask ShutdownAsync(CancellationToken cancellationToken = default)
+        {
+            Interlocked.Increment(ref _shutdownCalls);
+            if (BlockShutdown)
+            {
+                ShutdownEntered.TrySetResult();
+                await ReleaseShutdown.Task.WaitAsync(TimeSpan.FromSeconds(20), CancellationToken.None);
+            }
+            await base.ShutdownAsync(cancellationToken);
+            if (ThrowAfterShutdown) throw new InvalidOperationException(ShutdownFailure);
+        }
+
+        protected override void Dispose(bool disposing)
+        {
+            if (disposing) Interlocked.Increment(ref _disposalCalls);
+            base.Dispose(disposing);
+        }
+    }
+
+    private sealed class NonCreatingNodeManagerFactory : INodeManagerFactory
+    {
+        private int _createCalls;
+        internal int CreateCalls => Volatile.Read(ref _createCalls);
+        public ArrayOf<string> NamespacesUris => [];
+        public INodeManager Create(IServerInternal server, ApplicationConfiguration configuration)
+        {
+            Interlocked.Increment(ref _createCalls);
+            throw new InvalidOperationException("The terminal lifecycle probe must not create a node manager.");
+        }
+    }
+
+    /// <summary>Removes only this test's private fixture after bounded server cleanup.</summary>
+    public void Dispose()
+    {
+        if (Directory.Exists(_pki)) Directory.Delete(_pki, true);
+    }
+}
Author	SHA1	Message	Date
Joseph Doherty	9fb8d12d34	test: verify the mapped CLR inode in late-slope diagnostics	2026-11-10 18:02:08 -05:00
Joseph Doherty	8b9326ea06	test: add immutable Linux late-slope diagnostic tooling	2026-11-10 17:59:15 -05:00
Joseph Doherty	ce21f5db09	Merge master SDK1 ownership repair with SDK2 disposal adaptation	2026-11-10 17:17:00 -05:00
Joseph Doherty	2cf14f403b	test: verify SDK2 host disposal ownership and shared completion	2026-11-10 17:16:47 -05:00
Joseph Doherty	19610c6812	fix: share SDK2 host disposal completion and close unadopted servers	2026-11-10 17:06:27 -05:00
Joseph Doherty	58f8d57f5a	test: expose concurrent SDK2 host disposal completion race	2026-11-10 17:05:22 -05:00
Joseph Doherty	efb9197a0f	docs: record reviewed SDK1 terminal disposal repair v2-ci / unit-tests (push) Has been cancelled Details	2026-11-10 17:05:06 -05:00
Joseph Doherty	10edc44f94	fix: terminally dispose SDK1 server ownership once	2026-11-10 16:53:34 -05:00
Joseph Doherty	84bfe6cd20	test: cover SDK1 terminal server disposal ownership	2026-11-10 16:51:31 -05:00