Merge remote-tracking branch 'origin/master' into test/task-assertion-return-discard

.github/workflows/v2-ci.yml

@@ -1,7 +1,6 @@
 # CI for the v2 branch — runs on every push + PR to the v2-akka-fuse / master
-# branches. Layered into two jobs:
-#   build           dotnet restore + build (fast feedback on compile errors)
-#   unit-tests      the WHOLE solution minus the env-gated tiers (self-maintaining)
+# branches. One job restores and builds once, then tests the WHOLE solution
+# minus the env-gated tiers (self-maintaining).
 #
 # Skips E2E (Category=E2E) and LiveIntegration (needs a real gateway/PLC/VPN) — both
 # are env-gated tiers with no hosted-runner fixtures. (The skip-gated *.IntegrationTests
@@ -20,38 +19,35 @@ on:
     branches: [v2-akka-fuse, master]
   workflow_dispatch: {}
 
+# #707: one active run per repository across push/PR triggers. Gitea 1.26 supports
+# workflow concurrency; newer pending runs can supersede an older pending run.
+concurrency:
+  group: lmxopcua-v2-ci
+  cancel-in-progress: false
+
 env:
   DOTNET_NOLOGO: "1"
   DOTNET_CLI_TELEMETRY_OPTOUT: "1"
 
 jobs:
 
-  build:
+  unit-tests:
     runs-on: ubuntu-latest
     steps:
       - uses: actions/checkout@v4
       - uses: actions/setup-dotnet@v4
         with:
           dotnet-version: 10.0.x
-      # #606: code-reviews/README.md must be regenerated from the per-module findings.md
-      # files; --check exits non-zero on drift without writing (see REVIEW-PROCESS.md section 5).
+      # #606: regenerated review summary must remain in sync with module findings.
       - name: code-reviews README drift check
         run: python3 code-reviews/regen-readme.py --check
+      - name: TRX skip guard fixtures
+        run: python3 scripts/ci/test-assert-not-all-skipped.py
+      # #707: bound project workers and avoid concurrent NuGet restore work.
       - name: dotnet restore
-        run: dotnet restore ZB.MOM.WW.OtOpcUa.slnx
+        run: dotnet restore ZB.MOM.WW.OtOpcUa.slnx --disable-parallel -m:2
       - name: dotnet build
-        run: dotnet build ZB.MOM.WW.OtOpcUa.slnx --no-restore --configuration Release
-
-  unit-tests:
-    needs: build
-    runs-on: ubuntu-latest
-    steps:
-      - uses: actions/checkout@v4
-      - uses: actions/setup-dotnet@v4
-        with:
-          dotnet-version: 10.0.x
-      - name: dotnet restore
-        run: dotnet restore ZB.MOM.WW.OtOpcUa.slnx
+        run: dotnet build ZB.MOM.WW.OtOpcUa.slnx --no-restore --configuration Release -m:2
       # #606: the runner's size is the context every timing-sensitive failure needs and the log never
       # recorded it (load average included: the two Gitea runners may be running jobs side by side).
       - name: runner capacity
@@ -69,14 +65,15 @@ jobs:
       # are unreachable, so they run harmlessly here (and are counted by the skip-gate below).
       - name: dotnet test (whole solution, minus E2E + LiveIntegration)
         # #606: Akka TestKit time dilation for the Runtime + ControlPlane actor suites (read by their
-        # harnesses, tests/Server/*/Harness/TestTimeFactor.cs). This leg runs every test assembly at once
-        # on the runner, and their 2–8 s presence budgets — sized on an idle dev box — failed in a long
+        # harnesses, tests/Server/*/Harness/TestTimeFactor.cs). Their 2–8 s presence budgets
+        # — sized on an idle dev box — failed in a long
         # tail, a different test per run. Dilation only stretches waits: a presence wait still fails if
         # the message never comes, and absence windows get longer (stricter). Local runs stay at 1.
         env:
           AKKA_TEST_TIMEFACTOR: "3"
         run: >
-          dotnet test ZB.MOM.WW.OtOpcUa.slnx --configuration Release --no-restore
+          dotnet test ZB.MOM.WW.OtOpcUa.slnx --configuration Release --no-build
+          -m:1 -p:TestTfmsInParallel=false --settings scripts/ci/ci.runsettings
           --filter "Category!=E2E&Category!=LiveIntegration"
           --logger "trx" --results-directory artifacts/trx
       # 07/S-2 fail-on-skip: a whole-solution run that executed ZERO tests is a masked outage

docs/AlarmHistorian.md

@@ -153,6 +153,19 @@ Ordering follows from that: the drain reads in `enqueued_at_utc` order (with `id
 as a tiebreak so the ordering is total), because a hashed key carries no
 insertion sequence.
 
+Gateway delivery carries a separate nonzero `DeliveryId` GUID. Direct event instances
+create it once; retries and record copies preserve it. It is excluded from queue JSON
+and business equality, so the existing payload fingerprint and pair deduplication stay
+unchanged. The drain derives its delivery GUID solely from the persisted row `id`,
+including migrated legacy rows. It never hashes a reconstructed payload or mints an ID
+per send. The versioned encoding in `AlarmDeliveryIds` therefore survives retry,
+restart, payload edits and replicated failover without a schema migration.
+
+Different business payloads at the same timestamp remain distinct; completely identical
+payloads deliberately converge. An explicitly empty direct delivery ID is rejected.
+The Gateway mapper sends this GUID as the wire event ID; the live alarm readback test
+requires that exact ID. SQL Server and QuestDB retain their existing event-ID schemes.
+
 `EnqueueAsync` does a single `INSERT … ON CONFLICT DO NOTHING` on the hot path.
 To avoid a `SELECT COUNT(*)` on every enqueue, the sink keeps an in-memory
 non-dead-lettered row counter (seeded at startup, kept current by every mutation,

docs/AlarmTracking.md

@@ -456,9 +456,15 @@ from the scripted `AlarmCommandRouter`:
    carrying the authored `FullName` as the `ConditionId` and the
    authenticated `OperatorUser`. The driver forwards this to the Galaxy
    gateway → AVEVA via `GatewayGalaxyAlarmAcknowledger`.
-   **Fire-and-forget** — a failed upstream ack is not surfaced back to the
-   OPC UA client (mirrors the Galaxy write-outcome limitation; the local
-   `AlarmConditionState` SDK update already committed at step 1).
+   The Galaxy adapter awaits the correlated `AcknowledgeAlarmAsync` RPC:
+   gateway protocol failures throw, while a non-zero native return code is
+   warning-logged. **The actor route back to OPC UA is fire-and-forget** —
+   `DriverInstanceActor` logs and drops failures without replying to the
+   OPC UA client, whose local `AlarmConditionState` SDK update already
+   committed at step 1. This is a native-ack routing limitation; Galaxy
+   value writes have a separate correlated write-outcome path. An awaited
+   ack RPC does not prove the ack reached AVEVA event history: the H6 live
+   verification remains [operator-gated](plans/2026-06-15-stillpending-phase-3-opcua-standards-design.md).
 
 Only the **Acknowledge** is routed to the driver. **A native condition
 supports Acknowledge and AddComment only** (Gitea #587 review): `Confirm`,

docs/ScriptedAlarms.md

@@ -325,8 +325,10 @@ routes through a separate `NativeAlarmAckRouter` seam (instead of the scripted
 FullName)` inverse map, Primary-gated) → the owning driver's
 `IAlarmSource.AcknowledgeAsync` → Galaxy gateway → AVEVA. The call carries the
 authenticated operator's display name via the `AlarmAcknowledgeRequest.OperatorUser`
-field. This is fire-and-forget — a failed upstream ack is not surfaced back to the
-OPC UA client (mirrors the Galaxy write-outcome limitation). See [AlarmTracking.md
+field. The actor route back to OPC UA is fire-and-forget — a failed upstream ack
+is logged rather than surfaced back to the OPC UA client. The Galaxy adapter
+awaits its correlated RPC; this does not establish delivery to AVEVA event history.
+Galaxy value writes use a separate correlated write-outcome path. See [AlarmTracking.md
 §Native alarm acknowledge → AVEVA](AlarmTracking.md#native-alarm-acknowledge--aveva)
 for the full routing diagram.
 

docs/drivers/S7.md

@@ -198,15 +198,53 @@ Timers are **read-only** this phase; a write attempt returns `BadNotWritable`.
 ### Counter read
 
 **Address**: `C{n}` (e.g. `C0`, `C10`). **DataType must be `Int32`**. The OPC UA node is
-a scalar `Int32` whose value is the current **count** (raw word from the PLC).
+a scalar `Int32` containing the unsigned **raw big-endian word**, widened to `Int32`:
+`(bytes[0] << 8) | bytes[1]`. `S7.Net`'s `Counter.FromByteArray` performs no BCD
+reinterpretation; for example, bytes `01 23` surface as `291`. This describes the
+current decoder, not verified agreement with a PLC's displayed count.
 Counters are **read-only** this phase; a write attempt returns `BadNotWritable`.
 
-> **Known limitation — Counter BCD encoding on S7-300/400**: `S7.Net`'s
-> `Counter.FromByteArray` returns the raw big-endian word without BCD decode.
-> On **classic S7-300/400** the C-area word is BCD-encoded (0–999), so the
-> surfaced value can differ from the actual count on that hardware. S7-1200/1500
-> use IEC/DB counters (plain integers) where the raw word is correct. BCD
-> reinterpretation for legacy C-area is a live-hardware-gated follow-up.
+The S7comm ReadVar **C-area response encoding remains unverified on hardware**.
+PLC instruction representations do not by themselves establish network encoding:
+Siemens documents a binary current value for [L: Load counter](https://docs.tia.siemens.cloud/r/en-us/v21/stl-s7-300-s7-400/stl-mnemonic-s7-300-s7-400/counter-operations-s7-300-s7-400/l-load-counter-s7-300-s7-400)
+and separate `CV` and `CV_BCD` outputs for [S_CD](https://docs.tia.siemens.cloud/r/en-us/v21/lad-s7-300-s7-400/counter-operations-s7-300-s7-400/simatic-counters-s7-300-s7-400/s_cd-assign-parameters-and-count-down-s7-300-s7-400?contentId=reR_xapFX_Ike2opHIvfzg).
+
+IEC counter values stored in DB memory use the **DB address and declared integer
+type**, a different read path from `C{n}`. CPU family alone does not select the
+counter kind or prove its wire representation. In particular, Siemens also lists
+legacy SIMATIC counters for S7-1500 in its [instruction comparison (pages 23–24)](https://cache.industry.siemens.com/dl/files/777/109986777/att_1322640/v1/s7_1500_compare_table_en.pdf?download=true).
+The driver has no counter encoding selector; hardware-faithful counter support
+requires independent raw-response evidence before any interpretation change.
+
+### Counter evidence follow-up (#616)
+
+The existing `S7_1500SpanReadTests` exercises the python-snap7 profile's synthetic
+raw words (`C0 = 4711`, `C7 = 31337`). These are outside the classic SIMATIC count
+range `0..999` documented by S_CD. These expectations describe raw-word handling;
+they do not establish classic counter count semantics. The recorded device-0 run cannot prove nonzero
+numbering. The nonzero test is environment-gated, but its options remain
+`CpuType = S71500`, rack `0`, slot `0`: setting `S7_TIMER_NUMBER_ADDRESSING=1`
+does not configure a classic CPU or make the profile a classic counter gate.
+
+Before claiming hardware validation or closing #616:
+
+- Prepare a separate target-specific semantic gate with the actual CPU type,
+  rack, slot and TSAP/connection settings. Keep the synthetic raw-word profile
+  as its own evidence of the existing decoder contract.
+- Record CPU model/order number, firmware, endpoint, connection settings and
+  driver/S7netplus plus simulator versions where applicable.
+- Observe independent PLC-side current counts and binary/BCD reference outputs
+  for valid classic counts such as `123`, a second distinguishing count (`42`
+  or `999`), and zero. Record the instruction/counter kind and DB reference
+  addresses; do not derive reference values from the driver's output.
+- Capture the actual two-byte S7comm C-area responses alongside those references.
+  Do not infer response bytes from a simulator seed or expected BCD value.
+- Verify several distinct nonzero timer/counter numbers against their independent
+  PLC values and captured requests/responses, and verify the read-only contract.
+
+The raw-byte/count and nonzero-numbering gates remain open. Password/protection
+support inherited from #303/#632 is a separate follow-up, not evidence supplied
+by these counter checks.
 
 ### Deferrals (not yet implemented)
 

docs/plans/2026-06-15-stillpending-phase-3-opcua-standards-design.md

@@ -120,7 +120,9 @@ OPC UA client Acknowledge on a NATIVE condition
   under Lock, like `AlarmCommandRouter`).
 - **Driver faulted / reconnecting** → `AcknowledgeAsync` fails or no-ops per the driver; the local condition
   state still reflects the SDK-applied ack (the OPC UA client sees Good). Surfacing a *failed* upstream ack
-  back to the client is out of scope (mirrors the Galaxy write-outcome fire-and-forget limitation).
+  back to the client is out of scope for this native-ack actor route. **Current clarification
+  (2026-10-13):** the Galaxy adapter awaits a correlated ack RPC, but the actor route does not
+  return its outcome to OPC UA; Galaxy value writes have a separate correlated write-outcome path.
 - **H4 native Enable/Disable** → `BadNotSupported` (decision #2).
 - **H2-bit:** no runtime behavior change for existing grants except closing the HistoryRead⇒HistoryUpdate
   authorization hole; the service itself stays rejected.

docs/plans/2026-10-13-alarm-delivery-identity-results.md

@@ -0,0 +1,9 @@
+# Alarm delivery identity results — lmxopcua #705
+
+Alarm events now carry a nonzero delivery GUID created once for direct delivery and preserved across retries and record copies. It is excluded from queue JSON and business-field equality/hash. Queued delivery overrides it with a frozen versioned UUID encoding derived only from the persisted queue row key, preserving existing payload JSON, fingerprints, schema, deduplication and AlarmId. The Gateway mapper sends that GUID and rejects explicitly empty identities. SQL Server and QuestDB identity logic is unchanged.
+
+Coverage includes literal historical JSON/fingerprint/GUID vectors; ten-field equality and hash behavior; direct copies/retries; queue retry/reopen/payload edits/dead-letter revival; identical-input convergence; distinct same-time business events; real replicated queue convergence; and legacy migration. Reconstructed direct objects receive new identities; durable reconstruction uses the queue. Completely identical business payloads remain intentionally deduplicated.
+
+TDD: direct wire regression failed on an empty ID; queued retry/reopen regression failed on changing IDs. Final affected runs: Core51/51, Host migration/convergence16/16, Gateway453passed/6live skipped/0failed. Full offline solution:14,635passed/611skipped/0failed across62suites, exit0. Independent final review and diffcheck clear. Fixture configuration variables were removed before all runs.
+
+No live event writes, deployed sender rollout or historical row rewrite occurred. The live round-trip test now requires exact delivery GUID readback; actual persistence remains the open Gateway #28 gate, alongside its separate #39 timestamp-offset provider failure. Upstream History controls #29 and lmxopcua #694/#700 remain open.

docs/plans/2026-10-13-cache-concurrency-results.md

@@ -0,0 +1,9 @@
+# Gateway tag-presence concurrency coverage — 2026-10-13
+
+Approved tests-only fix for #701. Added 15 deterministic cases for overlapping positive/negative cold lookups, expiry refreshes, capacity crossing, successful-result isolation from faults/cancellation, and cancellation after asynchronous completion. Fake presence calls support an async hook and atomic counter. Production cache is unchanged; no single-flight promise.
+
+The synthetic 16-caller cold/expiry bursts produced 16 catalog RPCs. Assertions permit one through16; this records amplification without making it a requirement or a live-cost claim. Newclass15/15 and Gateway449/449 passed. Independent review found no actionable defects; diff check clean.
+
+First full solution run:14,614passed/609skipped/1failed across62suites. Existing restart/reconnect wire test failed, then isolated retry passed1/1. Full repeat with detailed console logging passed14,615/609skip/0fail across62suites, exit0, including OPC UA integration122/122 and host integration268passed/12skip. #702 tracks the unresolved intermittent failure; it is not closed merely because the repeat passed.
+
+Both full runs excluded configured live fixture variables from the environment; environment-gated tests skipped. Existing NU1510/xUnit duplicate-InlineData warnings remain pre-existing. No production behavior or deployment changed by this patch.

docs/plans/2026-10-13-restart-loss-reason-results.md

@@ -0,0 +1,9 @@
+# Restart loss reason test correction — #702
+
+The full integration reproduction failed at the restart wire test reason assertion: it required BadServerHalted but correctly received BadSessionIdInvalid. Production already supports that definitive loss reason.
+
+The test now accepts an explicit semantic set: four direct definitive session statuses and two sustained closed-channel statuses. All BadNoCommunication, retained-value, unhealthy-state, frozen-read-stamp, one-notification and explicit reinitialization/recovery assertions remain unchanged. Only test expectation/comments changed; production unchanged. Existing deterministic driver tests independently enforce direct-status ordering/count and the sustained-channel threshold.
+
+Verification: isolated restart1/1, deterministic status routes30/30, full OPC UA integration122/122. Full offline solution: 14,613 passed / 611 skipped / 0 failed, exit0, 61 reported suite summaries. Host integration266 passed /14 skipped; unavailable environment/fixture tests remain skips. Independent review found no important defects; git diff --check passed. Actual prior red run121pass/1fail, detailed console/TRX confirms the exact reason mismatch.
+
+Logs/TRX retained in the Codex thread workspace under work/issue702-*. No Gateway or deployment behavior changed by this test correction. Separate live gates remain open.

scripts/ci/README.md

@@ -0,0 +1,53 @@
+# Whole-solution CI
+
+The `v2-ci` workflow restores, builds and tests in one job so the test step uses
+that job's Release binaries. Restore disables parallel package work; restore and
+build use at most two MSBuild workers. Testing uses one MSBuild worker, disables
+parallel target-framework runs and loads `ci.runsettings`: one VSTest process and
+two xUnit collection workers with conservative scheduling. Assemblies that disable
+parallelization keep that setting.
+
+The workflow retains the E2E/LiveIntegration exclusions, includes SlowUnit, keeps
+Akka's test time factor at three, and requires at least 12,000 executed tests.
+Runner capacity and TRX files remain available for diagnosing failures. These
+limits reduce contention; they do not prove every timing failure has one cause.
+There are no automatic test retries or changed assertion deadlines.
+
+Gitea 1.26 supports the repository-wide `lmxopcua-v2-ci` concurrency group. Only
+one run is active; an active run is not cancelled, but a newer pending run may
+supersede an older pending run. The group does not serialize other repositories
+using the same host.
+
+To reproduce the bounded test leg after restore and Release build:
+
+```bash
+AKKA_TEST_TIMEFACTOR=3 dotnet test ZB.MOM.WW.OtOpcUa.slnx \
+  --configuration Release --no-build -m:1 -p:TestTfmsInParallel=false \
+  --settings scripts/ci/ci.runsettings \
+  --filter "Category!=E2E&Category!=LiveIntegration" \
+  --logger trx --results-directory artifacts/trx
+MIN_EXECUTED=12000 bash scripts/ci/assert-not-all-skipped.sh artifacts/trx/*.trx
+```
+
+The guard uses the existing TRX `Counters/@executed` value for the aggregate
+floor. Its `skipped` diagnostic counts individual `Results/UnitTestResult`
+records whose outcome is `NotExecuted`, using the document's XML namespace.
+Some VSTest reports set summary `notExecuted` to zero even when these records
+are present. Skip records are counted per file, including repeated runs; they
+are not a count of unique test methods and do not affect the executed floor.
+For malformed XML, the diagnostic retains the old summary-counter fallback;
+this guard does not add XML validation to its existing gate policy.
+
+Run the deterministic script fixtures without building or running the solution:
+
+```bash
+python3 scripts/ci/test-assert-not-all-skipped.py
+```
+
+Verified hosted baseline: [Gitea run 975](https://gitea.dohertylan.com/dohertj2/lmxopcua/actions/runs/975)
+completed successfully on 2026-10-13 for commit
+`76bdca90e94e814e06ef10b78271e2b492b2ba32` (merged by PR #711).
+Restore, Release build, README drift check, the serialized whole-solution test
+step, the 12,000-test floor and TRX upload all passed. The floor counted
+14,630 executed tests across 62 TRX files. This baseline predates the individual
+skip-reporting correction; its printed zero skip diagnostic is not skip evidence.

scripts/ci/assert-not-all-skipped.sh

@@ -33,12 +33,31 @@ for f in "$@"; do
     exit 2
   fi
   seen=$((seen + 1))
-  # The trx <Counters .../> element in <ResultSummary> carries executed + notExecuted counts.
+  # Keep the executed-counter floor unchanged; notExecuted is only a fallback
+  # for malformed XML that this guard historically accepted.
   counters="$(grep -oE '<Counters[^>]*/?>' "$f" | head -1)"
   ex="$(printf '%s' "$counters" | grep -oE 'executed="[0-9]+"' | grep -oE '[0-9]+' || true)"
   ne="$(printf '%s' "$counters" | grep -oE 'notExecuted="[0-9]+"' | grep -oE '[0-9]+' || true)"
   ex="${ex:-0}"
   ne="${ne:-0}"
+  # VSTest can emit notExecuted="0" despite individual NotExecuted results.
+  # Count direct Results/UnitTestResult records in the document's namespace,
+  # excluding foreign elements, nested records and escaped diagnostic text.
+  ne="$(python3 - "$f" "$ne" <<'PY'
+import sys
+import xml.etree.ElementTree as ET
+
+try:
+    root = ET.parse(sys.argv[1]).getroot()
+except ET.ParseError:
+    # Skip reporting must not introduce a new malformed-XML gate failure.
+    print(sys.argv[2])
+else:
+    namespace = root.tag.split("}", 1)[0] + "}" if root.tag.startswith("{") else ""
+    results = root.findall(f"{namespace}Results/{namespace}UnitTestResult")
+    print(sum(result.get("outcome") == "NotExecuted" for result in results))
+PY
+)"
   total_executed=$((total_executed + ex))
   total_skipped=$((total_skipped + ne))
   echo "  $(basename "$f"): executed=${ex} skipped=${ne}"

scripts/ci/ci.runsettings

@@ -0,0 +1,12 @@
+<?xml version="1.0" encoding="utf-8"?>
+<RunSettings>
+  <!-- #707: project scheduling is also bounded by dotnet test -m:1. -->
+  <RunConfiguration>
+    <MaxCpuCount>1</MaxCpuCount>
+  </RunConfiguration>
+  <xUnit>
+    <!-- Preserve each assembly's parallelization opt-out; bound enabled collections. -->
+    <MaxParallelThreads>2</MaxParallelThreads>
+    <ParallelAlgorithm>conservative</ParallelAlgorithm>
+  </xUnit>
+</RunSettings>

scripts/ci/test-assert-not-all-skipped.py

@@ -0,0 +1,108 @@
+#!/usr/bin/env python3
+"""Deterministic subprocess checks for the TRX executed-floor guard."""
+import os
+from pathlib import Path
+import subprocess
+import tempfile
+import unittest
+
+
+SCRIPT = Path(__file__).with_name("assert-not-all-skipped.sh")
+NS = "http://microsoft.com/schemas/VisualStudio/TeamTest/2010"
+
+
+def trx(executed, outcomes, summary_skipped=0, namespace=NS, extra=""):
+    declaration = f' xmlns="{namespace}"' if namespace else ""
+    results = "".join(f'<UnitTestResult outcome="{outcome}"/>' for outcome in outcomes)
+    return (f'<TestRun{declaration}><Results>{results}{extra}</Results>'
+            f'<ResultSummary><Counters total="{len(outcomes)}" executed="{executed}" '
+            f'notExecuted="{summary_skipped}"/></ResultSummary></TestRun>')
+
+
+class SkipGuardTests(unittest.TestCase):
+    def setUp(self):
+        self.directory = tempfile.TemporaryDirectory(prefix="trx-skip-fixtures-")
+        self.addCleanup(self.directory.cleanup)
+        self.root = Path(self.directory.name)
+
+    def write(self, name, content):
+        path = self.root / name
+        path.write_text(content, encoding="utf-8")
+        return path
+
+    def run_guard(self, *paths, minimum=None):
+        env = os.environ.copy()
+        env.pop("MIN_EXECUTED", None)
+        if minimum is not None:
+            env["MIN_EXECUTED"] = str(minimum)
+        return subprocess.run(["bash", str(SCRIPT), *map(str, paths)], env=env,
+                              text=True, capture_output=True, check=False)
+
+    def assert_report(self, result, status, executed, skipped, files=1):
+        self.assertEqual(status, result.returncode, result.stdout + result.stderr)
+        self.assertIn(f"files={files} executed={executed} skipped={skipped} ", result.stdout)
+
+    def test_mixed_results_ignore_zero_summary_skip_counter(self):
+        # A summary-only skip count misses the three actual NotExecuted records.
+        path = self.write("mixed.trx", trx(827, ["Passed"] * 827 + ["NotExecuted"] * 3))
+        result = self.run_guard(path)
+        self.assert_report(result, 0, 827, 3)
+        self.assertIn("mixed.trx: executed=827 skipped=3", result.stdout)
+
+    def test_all_skipped_fails_default_floor_but_report_only_passes(self):
+        path = self.write("all-skipped.trx", trx(0, ["NotExecuted"] * 188))
+        self.assert_report(self.run_guard(path), 1, 0, 188)
+        self.assert_report(self.run_guard(path, minimum=0), 0, 0, 188)
+
+    def test_zero_tests_fails_default_floor_but_report_only_passes(self):
+        path = self.write("zero.trx", trx(0, []))
+        self.assert_report(self.run_guard(path), 1, 0, 0)
+        self.assert_report(self.run_guard(path, minimum=0), 0, 0, 0)
+
+    def test_agreeing_summary_and_outcomes_are_not_double_counted(self):
+        path = self.write("agreeing.trx", trx(1, ["Passed", "NotExecuted"], 1))
+        self.assert_report(self.run_guard(path), 0, 1, 1)
+
+    def test_outcomes_are_authoritative_when_summary_overstates_skips(self):
+        path = self.write("overstated.trx", trx(1, ["Passed", "NotExecuted"], 9))
+        self.assert_report(self.run_guard(path), 0, 1, 1)
+
+    def test_namespace_path_and_exact_outcome_exclude_decoys(self):
+        extra = ('<Other><UnitTestResult outcome="NotExecuted"/></Other>'
+                 '<foreign:UnitTestResult xmlns:foreign="urn:other" outcome="NotExecuted"/>'
+                 '<Output><Message>&lt;UnitTestResult outcome="NotExecuted"/&gt;</Message></Output>')
+        content = trx(2, ["Passed", "Failed", "NotExecuted", "notexecuted"], extra=extra)
+        # Prefixes on Results/UnitTestResult must work just like the default namespace.
+        content = content.replace('<TestRun ', f'<TestRun xmlns:t="{NS}" ')
+        content = content.replace('<Results>', '<t:Results>').replace('</Results>', '</t:Results>')
+        content = content.replace('<UnitTestResult ', '<t:UnitTestResult ')
+        path = self.write("namespaced.trx", content)
+        self.assert_report(self.run_guard(path), 0, 2, 1)
+
+    def test_unnamespaced_results_remain_supported(self):
+        path = self.write("legacy.trx", trx(1, ["Passed", "NotExecuted"], namespace=""))
+        self.assert_report(self.run_guard(path), 0, 1, 1)
+
+    def test_executed_counter_controls_aggregate_floor_at_boundary(self):
+        # Counters retain authority even when result records disagree with executed.
+        mixed = self.write("floor.trx", trx(12000, ["Passed", "NotExecuted"]))
+        skipped = self.write("skipped.trx", trx(0, ["NotExecuted"]))
+        self.assert_report(self.run_guard(mixed, skipped, minimum=12000), 0, 12000, 2, 2)
+        self.assert_report(self.run_guard(mixed, skipped, minimum=12001), 1, 12000, 2, 2)
+
+    def test_missing_arguments_and_file_keep_exit_two(self):
+        self.assertEqual(2, self.run_guard().returncode)
+        self.assertEqual(2, self.run_guard(self.root / "missing.trx").returncode)
+
+    def test_missing_counters_keeps_existing_exit_one(self):
+        path = self.write("no-counters.trx", '<TestRun><Results/></TestRun>')
+        self.assertEqual(1, self.run_guard(path, minimum=0).returncode)
+
+    def test_malformed_xml_keeps_summary_fallback_and_gate_status(self):
+        path = self.write("malformed.trx", '<TestRun><Counters executed="2" notExecuted="3"/>')
+        self.assert_report(self.run_guard(path, minimum=2), 0, 2, 3)
+        self.assert_report(self.run_guard(path, minimum=3), 1, 2, 3)
+
+
+if __name__ == "__main__":
+    unittest.main(verbosity=2)

src/Core/ZB.MOM.WW.OtOpcUa.Commons/OpcUa/DeferredServiceLevelPublisher.cs

@@ -1,21 +1,38 @@
 namespace ZB.MOM.WW.OtOpcUa.Commons.OpcUa;
 
 /// <summary>
-/// Late-binding adapter that holds an inner <see cref="IServiceLevelPublisher"/> reference
-/// swappable at runtime. Mirrors <see cref="DeferredAddressSpaceSink"/>: Akka actors resolve
-/// the publisher at DI time, but the production <c>SdkServiceLevelPublisher</c> only exists
-/// after <c>StandardServer.Start</c>. The Host's hosted service swaps the inner once the SDK
-/// is up; until then writes route through <see cref="NullServiceLevelPublisher"/>.
+/// Late-binding adapter that accepts and retains the latest requested ServiceLevel until
+/// the SDK publisher is available. Binding synchronously replays that request, including zero.
 /// </summary>
+/// <remarks>
+/// Publishing and binding are serialized through the synchronous inner call. Inner publishers
+/// must not call back into this adapter. Acceptance while unbound is not an SDK write confirmation.
+/// A failed call retains both the requested level and the installed publisher for later recovery.
+/// </remarks>
 public sealed class DeferredServiceLevelPublisher : IServiceLevelPublisher
 {
-    private volatile IServiceLevelPublisher _inner = NullServiceLevelPublisher.Instance;
+    private readonly object _gate = new();
+    private IServiceLevelPublisher? _inner;
+    private byte? _latest;
 
-    /// <summary>Swap the underlying publisher. Pass null to revert to the Null no-op.</summary>
-    /// <param name="inner">The publisher implementation to use, or null to use the null publisher.</param>
-    public void SetInner(IServiceLevelPublisher? inner) =>
-        _inner = inner ?? NullServiceLevelPublisher.Instance;
+    /// <summary>Bind and replay the latest request, or unbind while retaining that request.</summary>
+    /// <param name="inner">The publisher implementation to use, or null to unbind.</param>
+    public void SetInner(IServiceLevelPublisher? inner)
+    {
+        lock (_gate)
+        {
+            _inner = inner;
+            if (_latest is { } latest) _inner?.Publish(latest);
+        }
+    }
 
     /// <inheritdoc />
-    public void Publish(byte serviceLevel) => _inner.Publish(serviceLevel);
+    public void Publish(byte serviceLevel)
+    {
+        lock (_gate)
+        {
+            _latest = serviceLevel;
+            _inner?.Publish(serviceLevel);
+        }
+    }
 }

src/Core/ZB.MOM.WW.OtOpcUa.Core.AlarmHistorian/AlarmDeliveryIds.cs

@@ -0,0 +1,25 @@
+using System.Security.Cryptography;
+using System.Text;
+
+namespace ZB.MOM.WW.OtOpcUa.Core.AlarmHistorian;
+
+/// <summary>Stable delivery GUIDs for persisted alarm queue rows.</summary>
+public static class AlarmDeliveryIds
+{
+    /// <summary>
+    /// Converts an existing durable row key into a nonzero version-8 GUID. Version 1 of this encoding hashes
+    /// UTF-8 "OtOpcUa.AlarmDelivery/v1\0" followed by the exact persisted key with SHA-256, takes the first
+    /// 16 bytes, sets the UUID version/variant bits, and interprets the bytes in network order.
+    /// This encoding must remain stable across retries, upgrades and replicated failover.
+    /// </summary>
+    /// <param name="queueId">The persisted row key, never a hash recomputed from the current payload.</param>
+    /// <returns>The row's stable delivery identity.</returns>
+    public static Guid FromQueueId(string queueId)
+    {
+        ArgumentException.ThrowIfNullOrWhiteSpace(queueId);
+        var digest = SHA256.HashData(Encoding.UTF8.GetBytes("OtOpcUa.AlarmDelivery/v1\0" + queueId));
+        digest[6] = (byte)((digest[6] & 0x0F) | 0x80);
+        digest[8] = (byte)((digest[8] & 0x3F) | 0x80);
+        return new Guid(digest.AsSpan(0, 16), bigEndian: true);
+    }
+}

src/Core/ZB.MOM.WW.OtOpcUa.Core.AlarmHistorian/AlarmHistorianEvent.cs

@@ -1,3 +1,4 @@
+using System.Text.Json.Serialization;
 using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
 
 namespace ZB.MOM.WW.OtOpcUa.Core.AlarmHistorian;
@@ -34,4 +35,31 @@ public sealed record AlarmHistorianEvent(
     string Message,
     string User,
     string? Comment,
-    DateTime TimestampUtc);
+    DateTime TimestampUtc)
+{
+    /// <summary>
+    /// Delivery identity, minted once for direct sends and replaced by the durable queue's identity when drained.
+    /// Excluded from payload serialization and business equality so redundant enqueues still converge.
+    /// A record copy preserves this identity; an explicitly empty identity is invalid for delivery.
+    /// </summary>
+    [JsonIgnore]
+    public Guid DeliveryId { get; init; } = Guid.NewGuid();
+
+    /// <summary>Compares the ten business fields, excluding transport delivery identity.</summary>
+    /// <param name="other">The event to compare.</param>
+    /// <returns>Whether the business fields are equal.</returns>
+    public bool Equals(AlarmHistorianEvent? other) => ReferenceEquals(this, other) || other is not null
+        && AlarmId == other.AlarmId && EquipmentPath == other.EquipmentPath && AlarmName == other.AlarmName
+        && AlarmTypeName == other.AlarmTypeName && Severity == other.Severity && EventKind == other.EventKind
+        && Message == other.Message && User == other.User && Comment == other.Comment && TimestampUtc == other.TimestampUtc;
+
+    /// <summary>Hashes the same business fields used by equality, excluding delivery identity.</summary>
+    /// <returns>The business-field hash code.</returns>
+    public override int GetHashCode()
+    {
+        var hash = new HashCode();
+        hash.Add(AlarmId); hash.Add(EquipmentPath); hash.Add(AlarmName); hash.Add(AlarmTypeName); hash.Add(Severity);
+        hash.Add(EventKind); hash.Add(Message); hash.Add(User); hash.Add(Comment); hash.Add(TimestampUtc);
+        return hash.ToHashCode();
+    }
+}

src/Core/ZB.MOM.WW.OtOpcUa.Core.AlarmHistorian/LocalDbStoreAndForwardSink.cs

@@ -603,6 +603,9 @@ public sealed class LocalDbStoreAndForwardSink : IAlarmHistorianSink, IDisposabl
             try
             {
                 evt = JsonSerializer.Deserialize<AlarmHistorianEvent>(payload);
+                // The durable key survives retry, payload edits, legacy migration and replicated failover.
+                // Never derive delivery identity from a reserialized payload or an individual RPC attempt.
+                if (evt is not null) evt = evt with { DeliveryId = AlarmDeliveryIds.FromQueueId(id) };
             }
             catch (JsonException)
             {

src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway/Mapping/AlarmEventMapper.cs

@@ -35,16 +35,16 @@ internal static class AlarmEventMapper
     /// <returns>The gateway wire event ready for <c>SendEvent</c>.</returns>
     public static HistorianEvent ToHistorianEvent(AlarmHistorianEvent alarm)
     {
+        if (alarm.DeliveryId == Guid.Empty)
+            throw new ArgumentException("Alarm delivery identity must be nonzero.", nameof(alarm));
         // Timestamp.FromDateTime requires a Utc-kind DateTime; coerce defensively (TimestampUtc is
         // already Utc by contract, but a caller could pass Unspecified).
         var eventTime = Timestamp.FromDateTime(DateTime.SpecifyKind(alarm.TimestampUtc, DateTimeKind.Utc));
 
         var historianEvent = new HistorianEvent
         {
-            // Deliberately DO NOT set HistorianEvent.Id: the gateway's SendEvent path rejects a
-            // client-supplied event id (the server handler throws and the call fails permanently —
-            // confirmed live). The historian assigns event identity server-side; the alarm's own id
-            // is preserved below as a property for read-back correlation/traceability.
+            // Identity belongs to the event/durable queue, never to an individual RPC attempt.
+            Id = alarm.DeliveryId.ToString("D"),
             SourceName = alarm.EquipmentPath,
             Type = alarm.AlarmTypeName,
             EventTime = eventTime,

src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.S7/S7Driver.cs

@@ -1465,7 +1465,7 @@ public sealed class S7Driver
     {
         // AREA precedence first (before the DataType switch): a Timer (T{n}) decodes its 2-byte
         // S5TIME word to a duration in SECONDS via S7.Net's Timer.FromByteArray (returns double);
-        // a Counter (C{n}) decodes its 2-byte word to a COUNT via Counter.FromByteArray (returns
+        // a Counter (C{n}) exposes its 2-byte RAW WORD via Counter.FromByteArray (returns
         // ushort, surfaced as a non-negative int to match the tag's Int32 type). Routing on Area
         // here is essential: a Timer tag is typed Float64, so letting it fall to the Float64 arm
         // below would ReadDoubleBigEndian over a 2-byte block = garbage/throw. Boxed explicitly so
@@ -1474,11 +1474,12 @@ public sealed class S7Driver
             return (object)global::S7.Net.Types.Timer.FromByteArray(block);
         if (addr.Area is S7Area.Counter)
         {
-            // NOTE: S7.Net Counter.FromByteArray returns the RAW big-endian word ((bytes[0]<<8)|bytes[1]),
-            // NOT a BCD decode. On classic S7-300/400 the C-area word is BCD (0-999), so on that hardware
-            // this raw value can differ from the displayed count; S7-1200/1500 use IEC/DB counters (plain
-            // ints), where it is correct. Surfacing S7.Net's value verbatim is the faithful choice; BCD
-            // reinterpretation for legacy C-area counters is a live-hardware-gated follow-up.
+            // S7.Net Counter.FromByteArray returns the raw big-endian word ((bytes[0]<<8)|bytes[1]),
+            // without BCD reinterpretation. This preserves the raw value, not a verified PLC count:
+            // S7comm C-area response encoding still needs independent raw bytes and PLC-side
+            // count/binary/BCD references (#616). CPU family alone does not prove that encoding.
+            // IEC counter integers in DB memory take the DB type codec, not this C-area path;
+            // S7-1500 also has legacy SIMATIC counters. See docs/drivers/S7.md for the evidence gate.
             return (object)(int)global::S7.Net.Types.Counter.FromByteArray(block);
         }
 

src/Server/ZB.MOM.WW.OtOpcUa.Host/OpcUa/FailedServiceLevelBindingCleanup.cs

@@ -0,0 +1,38 @@
+using Microsoft.Extensions.Logging;
+
+namespace ZB.MOM.WW.OtOpcUa.Host.OpcUa;
+
+/// <summary>Coordinates endpoint cleanup after ServiceLevel binding fails during startup.</summary>
+internal static class FailedServiceLevelBindingCleanup
+{
+    /// <summary>Stops and disposes the endpoint, returning whether ownership can be released.</summary>
+    /// <param name="stop">The SDK stop operation.</param>
+    /// <param name="dispose">The application-host disposal operation.</param>
+    /// <param name="logger">The cleanup logger.</param>
+    /// <returns>Whether cleanup completed successfully.</returns>
+    internal static async Task<bool> StopAndDisposeAsync(
+        Func<Task> stop, Func<ValueTask> dispose, ILogger logger)
+    {
+        var completed = true;
+        try
+        {
+            await stop().ConfigureAwait(false);
+        }
+        catch (Exception ex)
+        {
+            completed = false;
+            logger.LogError(ex, "OtOpcUaServerHostedService: SDK stop failed after binding failure; retaining endpoint ownership for shutdown retry");
+        }
+        // Attempt application cleanup even when stopping faults; keep ownership on either failure.
+        try
+        {
+            await dispose().ConfigureAwait(false);
+        }
+        catch (Exception ex)
+        {
+            completed = false;
+            logger.LogError(ex, "OtOpcUaServerHostedService: endpoint disposal failed after binding failure; retaining endpoint ownership for shutdown retry");
+        }
+        return completed;
+    }
+}

src/Server/ZB.MOM.WW.OtOpcUa.Host/OpcUa/OtOpcUaServerHostedService.cs

@@ -339,11 +339,36 @@ public sealed class OtOpcUaServerHostedService : IHostedService, IAsyncDisposabl
         _server.NodeManager.Clock = historyUpdate.Clock;
 
         // ServiceLevel publisher needs IServerInternal — only available after Start.
-        if (_server.CurrentInstance is { } serverInternal)
+        try
         {
-            _deferredServiceLevel.SetInner(new SdkServiceLevelPublisher(
-                serverInternal,
-                _loggerFactory.CreateLogger<SdkServiceLevelPublisher>()));
+            if (_server.CurrentInstance is { } serverInternal)
+            {
+                _deferredServiceLevel.SetInner(new SdkServiceLevelPublisher(
+                    serverInternal,
+                    _loggerFactory.CreateLogger<SdkServiceLevelPublisher>()));
+            }
+        }
+        catch (Exception ex)
+        {
+            _logger.LogError(ex,
+                "OtOpcUaServerHostedService: ServiceLevel binding failed; stopping the OPC UA endpoint while the admin host continues");
+            try
+            {
+                await StopAsync(cancellationToken).ConfigureAwait(false);
+            }
+            catch (Exception cleanupError)
+            {
+                _logger.LogError(cleanupError, "OtOpcUaServerHostedService: failed to detach OPC UA bindings after startup failure");
+            }
+            var released = await FailedServiceLevelBindingCleanup.StopAndDisposeAsync(
+                async () => await _server.StopAsync(CancellationToken.None).ConfigureAwait(false),
+                () => _appHost.DisposeAsync(), _logger).ConfigureAwait(false);
+            if (released)
+            {
+                _appHost = null;
+                _server = null;
+            }
+            return;
         }
 
         // arch-review 03/S19 — LAST, after the node manager is known non-null and after the ACL gate is

src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/OpcUaApplicationHost.cs

@@ -69,8 +69,10 @@ public sealed class OpcUaApplicationHostOptions
     ///         Note this controls <i>admission</i>, not authority. An admitted anonymous session is a
     ///         normal principal (<c>anonymous</c>) whose grants come from <c>NodeAcl</c> rows naming it,
     ///         exactly like a key's principal — there is no anonymous bypass branch in the evaluator.
-    ///         Until per-node enforcement lands, an admitted session of ANY kind can still read the whole
-    ///         address space, because reads carry no authorization check yet.
+    ///         When a wired data-plane authorizer is enabled, per-node ACL checks apply to anonymous
+    ///         sessions, including reads. Reads denied by the node manager return <c>BadUserAccessDenied</c>.
+    ///         With the authorizer disabled or not wired, reads retain the legacy behaviour without
+    ///         per-node ACL enforcement.
     ///     </para>
     /// </remarks>
     public bool AllowAnonymous { get; set; } = true;
@@ -558,10 +560,15 @@ public sealed class OpcUaApplicationHost : IAsyncDisposable
     /// <returns>A task that represents the asynchronous operation.</returns>
     public async ValueTask DisposeAsync()
     {
-        if (_impersonateHandler is not null && _server?.CurrentInstance?.SessionManager is { } sessionManager)
+        try
         {
-            try { sessionManager.ImpersonateUser -= _impersonateHandler; }
-            catch (Exception ex) { _logger.LogWarning(ex, "OpcUaApplicationHost: detaching ImpersonateUser threw"); }
+            // An explicit SDK stop may already have halted CurrentInstance; still continue disposal.
+            if (_impersonateHandler is not null && _server?.CurrentInstance?.SessionManager is { } sessionManager)
+                sessionManager.ImpersonateUser -= _impersonateHandler;
+        }
+        catch (Exception ex)
+        {
+            _logger.LogWarning(ex, "OpcUaApplicationHost: detaching ImpersonateUser threw");
         }
         _impersonateHandler = null;
 

src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/SdkServiceLevelPublisher.cs

@@ -38,8 +38,8 @@ public sealed class SdkServiceLevelPublisher : IServiceLevelPublisher
         var node = _serverInternal.ServerObject?.ServiceLevel;
         if (node is null)
         {
-            _logger.LogWarning("SdkServiceLevelPublisher: ServerObject.ServiceLevel unavailable; skipping write");
-            return;
+            _logger.LogWarning("SdkServiceLevelPublisher: ServerObject.ServiceLevel unavailable");
+            throw new InvalidOperationException("Server.ServiceLevel is unavailable.");
         }
 
         try
@@ -55,6 +55,7 @@ public sealed class SdkServiceLevelPublisher : IServiceLevelPublisher
         catch (Exception ex)
         {
             _logger.LogWarning(ex, "SdkServiceLevelPublisher: write to Server.ServiceLevel threw");
+            throw;
         }
     }
 }

src/Server/ZB.MOM.WW.OtOpcUa.Runtime/OpcUa/OpcUaPublishActor.cs

@@ -147,6 +147,7 @@ public sealed class OpcUaPublishActor : ReceiveActor, IWithTimers
     private readonly TimeSpan _staleWindow;
     private readonly TimeSpan _probeFreshnessWindow;
     private readonly TimeSpan _healthTickInterval;
+    private readonly Func<DateTime> _utcNow;
     private readonly Akka.Cluster.Cluster _cluster = Akka.Cluster.Cluster.Get(Context.System);
     private readonly ILoggingAdapter _log = Context.GetLogger();
 
@@ -200,9 +201,10 @@ public sealed class OpcUaPublishActor : ReceiveActor, IWithTimers
     /// <param name="probeFreshnessWindow">The window beyond which a peer's OPC UA probe verdict about
     /// this node is considered stale (and thus given the benefit of the doubt rather than demoting);
     /// defaults to 30 seconds.</param>
-    /// <param name="healthTickInterval">The period between self-driven DB-health refresh ticks (each
-    /// Asks <paramref name="dbHealthProbe"/> for its cached status); defaults to 5 seconds. No timer is
-    /// started when <paramref name="dbHealthProbe"/> is null.</param>
+    /// <param name="healthTickInterval">The period between freshness ticks; defaults to 5 seconds.
+    /// DB-backed nodes Ask the probe for cached status; DB-less nodes recompute snapshot freshness.
+    /// Legacy nodes without a probe do not start a timer.</param>
+    /// <param name="utcNow">The UTC clock used for freshness; defaults to <see cref="DateTime.UtcNow"/>.</param>
     /// <param name="dbLess">Per-cluster mesh Phase 4: this is a driver-only node with no ConfigDb, so the
     /// ServiceLevel DB-reachability input is retired (<c>DbReachable</c> treated constant-true, only
     /// staleness demotes). No <paramref name="dbHealthProbe"/> is wired on such a node.</param>
@@ -219,7 +221,8 @@ public sealed class OpcUaPublishActor : ReceiveActor, IWithTimers
         TimeSpan? probeFreshnessWindow = null,
         TimeSpan? healthTickInterval = null,
         bool dbLess = false,
-        IDataPlaneAuthoritySink? dataPlaneAuthority = null) =>
+        IDataPlaneAuthoritySink? dataPlaneAuthority = null,
+        Func<DateTime>? utcNow = null) =>
         Akka.Actor.Props.Create(() => new OpcUaPublishActor(
             sink ?? NullOpcUaAddressSpaceSink.Instance,
             serviceLevel ?? NullServiceLevelPublisher.Instance,
@@ -232,7 +235,8 @@ public sealed class OpcUaPublishActor : ReceiveActor, IWithTimers
             probeFreshnessWindow,
             healthTickInterval,
             dbLess,
-            dataPlaneAuthority)).WithDispatcher(DispatcherId);
+            dataPlaneAuthority,
+            utcNow)).WithDispatcher(DispatcherId);
 
     /// <summary>Test-only Props that omits the pinned-dispatcher requirement and skips the
     /// DPS subscribe so unit tests can spin up the actor on a vanilla TestKit cluster.</summary>
@@ -249,9 +253,10 @@ public sealed class OpcUaPublishActor : ReceiveActor, IWithTimers
     /// <param name="probeFreshnessWindow">The window beyond which a peer's OPC UA probe verdict about
     /// this node is considered stale (and thus given the benefit of the doubt rather than demoting);
     /// defaults to 30 seconds.</param>
-    /// <param name="healthTickInterval">The period between self-driven DB-health refresh ticks (each
-    /// Asks <paramref name="dbHealthProbe"/> for its cached status); defaults to 5 seconds. No timer is
-    /// started when <paramref name="dbHealthProbe"/> is null.</param>
+    /// <param name="healthTickInterval">The period between freshness ticks; defaults to 5 seconds.
+    /// DB-backed nodes Ask the probe for cached status; DB-less nodes recompute snapshot freshness.
+    /// Legacy nodes without a probe do not start a timer.</param>
+    /// <param name="utcNow">The UTC clock used for freshness; defaults to <see cref="DateTime.UtcNow"/>.</param>
     /// <param name="dbLess">Per-cluster mesh Phase 4: this is a driver-only node with no ConfigDb, so the
     /// ServiceLevel DB-reachability input is retired (<c>DbReachable</c> treated constant-true, only
     /// staleness demotes). No <paramref name="dbHealthProbe"/> is wired on such a node.</param>
@@ -269,7 +274,8 @@ public sealed class OpcUaPublishActor : ReceiveActor, IWithTimers
         TimeSpan? probeFreshnessWindow = null,
         TimeSpan? healthTickInterval = null,
         bool dbLess = false,
-        IDataPlaneAuthoritySink? dataPlaneAuthority = null) =>
+        IDataPlaneAuthoritySink? dataPlaneAuthority = null,
+        Func<DateTime>? utcNow = null) =>
         Akka.Actor.Props.Create(() => new OpcUaPublishActor(
             sink ?? NullOpcUaAddressSpaceSink.Instance,
             serviceLevel ?? NullServiceLevelPublisher.Instance,
@@ -282,7 +288,8 @@ public sealed class OpcUaPublishActor : ReceiveActor, IWithTimers
             probeFreshnessWindow,
             healthTickInterval,
             dbLess,
-            dataPlaneAuthority));
+            dataPlaneAuthority,
+            utcNow));
 
     /// <summary>Initializes a new instance of the <see cref="OpcUaPublishActor"/> class.</summary>
     /// <param name="sink">The OPC UA address space sink.</param>
@@ -298,9 +305,10 @@ public sealed class OpcUaPublishActor : ReceiveActor, IWithTimers
     /// <param name="probeFreshnessWindow">The window beyond which a peer's OPC UA probe verdict about
     /// this node is considered stale (and thus given the benefit of the doubt rather than demoting);
     /// defaults to 30 seconds.</param>
-    /// <param name="healthTickInterval">The period between self-driven DB-health refresh ticks (each
-    /// Asks <paramref name="dbHealthProbe"/> for its cached status); defaults to 5 seconds. No timer is
-    /// started when <paramref name="dbHealthProbe"/> is null.</param>
+    /// <param name="healthTickInterval">The period between freshness ticks; defaults to 5 seconds.
+    /// DB-backed nodes Ask the probe for cached status; DB-less nodes recompute snapshot freshness.
+    /// Legacy nodes without a probe do not start a timer.</param>
+    /// <param name="utcNow">The UTC clock used for freshness; defaults to <see cref="DateTime.UtcNow"/>.</param>
     /// <param name="dbLess">Per-cluster mesh Phase 4: this is a driver-only node with no ConfigDb, so the
     /// ServiceLevel DB-reachability input is retired (<c>DbReachable</c> treated constant-true, only
     /// staleness demotes). No <paramref name="dbHealthProbe"/> is wired on such a node.</param>
@@ -316,7 +324,8 @@ public sealed class OpcUaPublishActor : ReceiveActor, IWithTimers
         TimeSpan? probeFreshnessWindow = null,
         TimeSpan? healthTickInterval = null,
         bool dbLess = false,
-        IDataPlaneAuthoritySink? dataPlaneAuthority = null)
+        IDataPlaneAuthoritySink? dataPlaneAuthority = null,
+        Func<DateTime>? utcNow = null)
     {
         _sink = sink;
         _dataPlaneAuthority = dataPlaneAuthority;
@@ -330,6 +339,7 @@ public sealed class OpcUaPublishActor : ReceiveActor, IWithTimers
         _staleWindow = staleWindow ?? TimeSpan.FromSeconds(30);
         _probeFreshnessWindow = probeFreshnessWindow ?? TimeSpan.FromSeconds(30);
         _healthTickInterval = healthTickInterval ?? TimeSpan.FromSeconds(5);
+        _utcNow = utcNow ?? (() => DateTime.UtcNow);
 
         Receive<AttributeValueUpdate>(HandleAttributeUpdate);
         Receive<AttributeValueBatchUpdate>(HandleAttributeBatchUpdate);
@@ -353,11 +363,10 @@ public sealed class OpcUaPublishActor : ReceiveActor, IWithTimers
             DistributedPubSub.Get(Context.System).Mediator.Tell(new Subscribe(RedundancyStateTopic, Self));
         }
 
-        // Production delivery of DB-health: when a probe is wired, kick an immediate refresh + start the
-        // periodic tick so _lastDbHealth gets populated and kept fresh without any external pump. Gated
-        // ONLY on the probe being present (independent of the DPS subscribe) so the calculator path works
-        // in tests too. No probe → stay on the legacy role-only seam (no timer).
-        if (_dbHealthProbe is not null)
+        // DB-backed nodes refresh their cached probe status; DB-less nodes still need periodic
+        // snapshot/probe-verdict age checks even when no new messages arrive. Legacy nodes keep
+        // their role-only behavior without a timer.
+        if (_dbLess || _dbHealthProbe is not null)
         {
             Self.Tell(HealthTick.Instance);                                        // immediate first refresh
             Timers.StartPeriodicTimer("db-health", HealthTick.Instance, _healthTickInterval);
@@ -970,10 +979,10 @@ public sealed class OpcUaPublishActor : ReceiveActor, IWithTimers
         // node starting Detached/role-less (first level = 0) would be dedup'd away, leaving the SDK's
         // built-in default (255 = full service) standing — a degraded node wrongly advertising 255.
         if (_publishedAtLeastOnce && msg.ServiceLevel == _lastServiceLevel) return;
-        _lastServiceLevel = msg.ServiceLevel;
         try
         {
             _serviceLevel.Publish(msg.ServiceLevel);
+            _lastServiceLevel = msg.ServiceLevel;
             _publishedAtLeastOnce = true;
             OtOpcUaTelemetry.ServiceLevelChange.Add(1,
                 new KeyValuePair<string, object?>("level", msg.ServiceLevel));
@@ -981,6 +990,8 @@ public sealed class OpcUaPublishActor : ReceiveActor, IWithTimers
         }
         catch (Exception ex)
         {
+            // A failed write may be partially applied; even a previous successful value must retry.
+            _publishedAtLeastOnce = false;
             _log.Warning(ex, "OpcUaPublish: ServiceLevel publisher threw at level {Level}", msg.ServiceLevel);
         }
     }
@@ -1004,7 +1015,8 @@ public sealed class OpcUaPublishActor : ReceiveActor, IWithTimers
         RecomputeServiceLevel();
     }
 
-    /// <summary>Periodic self-tick: Asks the local <see cref="DbHealthProbeActor"/> for its cached
+    /// <summary>Periodic self-tick: recomputes freshness directly when no DB probe exists; otherwise
+    /// Asks the local <see cref="DbHealthProbeActor"/> for its cached
     /// status and pipes the reply back to <see cref="Self"/>, where <see cref="HandleDbHealthStatus"/>
     /// caches it + recomputes. A hung/late probe (Ask timeout) pipes a <c>Reachable=false</c> status so
     /// the node FAIL-SAFE-DEMOTES rather than freezing the last-known-good. The continuation only
@@ -1013,11 +1025,17 @@ public sealed class OpcUaPublishActor : ReceiveActor, IWithTimers
     /// touched here.</summary>
     private void OnHealthTick()
     {
-        _dbHealthProbe!.Ask<DbHealthProbeActor.DbHealthStatus>(
+        if (_dbHealthProbe is null)
+        {
+            RecomputeServiceLevel();
+            return;
+        }
+
+        _dbHealthProbe.Ask<DbHealthProbeActor.DbHealthStatus>(
                 DbHealthProbeActor.GetStatus.Instance, TimeSpan.FromSeconds(1))
             .ContinueWith(t => t.IsCompletedSuccessfully
                 ? t.Result
-                : new DbHealthProbeActor.DbHealthStatus(false, DateTime.UtcNow, "db-health ask timeout"))
+                : new DbHealthProbeActor.DbHealthStatus(false, _utcNow(), "db-health ask timeout"))
             .PipeTo(Self);
     }
 
@@ -1030,7 +1048,7 @@ public sealed class OpcUaPublishActor : ReceiveActor, IWithTimers
     {
         // The result targets the probed node. If it isn't me, it's about a peer — ignore it.
         if (_localNode is null || r.NodeId != _localNode.Value) return;
-        _probeAboutMe = (r.Ok, DateTime.UtcNow);
+        _probeAboutMe = (r.Ok, _utcNow());
         RecomputeServiceLevel();
     }
 
@@ -1042,7 +1060,7 @@ public sealed class OpcUaPublishActor : ReceiveActor, IWithTimers
     private bool OpcUaProbeOk()
     {
         if (_probeAboutMe is not { } verdict) return true;
-        if (DateTime.UtcNow - verdict.At > _probeFreshnessWindow) return true;
+        if (_utcNow() - verdict.At > _probeFreshnessWindow) return true;
         return verdict.Ok;
     }
 
@@ -1079,7 +1097,7 @@ public sealed class OpcUaPublishActor : ReceiveActor, IWithTimers
         // wins, so a DB-less Detached node stays 0.
         if (_dbLess)
         {
-            var utcNow = DateTime.UtcNow;
+            var utcNow = _utcNow();
             var dbLessInputs = new NodeHealthInputs(
                 MemberState: SafeSelfStatus(),
                 DbReachable: true,
@@ -1099,7 +1117,7 @@ public sealed class OpcUaPublishActor : ReceiveActor, IWithTimers
             return;
         }
 
-        var now = DateTime.UtcNow;
+        var now = _utcNow();
         var inputs = new NodeHealthInputs(
             MemberState: SafeSelfStatus(),
             DbReachable: _lastDbHealth.Reachable,

tests/Core/ZB.MOM.WW.OtOpcUa.Commons.Tests/OpcUa/DeferredServiceLevelPublisherTests.cs

@@ -10,10 +10,10 @@ namespace ZB.MOM.WW.OtOpcUa.Commons.Tests.OpcUa;
 public class DeferredServiceLevelPublisherTests
 {
     [Fact]
-    public void Before_SetInner_Publish_is_a_noop()
+    public void Before_SetInner_Publish_accepts_a_cached_request()
     {
         var publisher = new DeferredServiceLevelPublisher();
-        // Must not throw; the NullServiceLevelPublisher absorbs it.
+        // Accept the request before an SDK publisher exists.
         publisher.Publish(200);
     }
 
@@ -30,7 +30,7 @@ public class DeferredServiceLevelPublisherTests
     }
 
     [Fact]
-    public void SetInner_null_reverts_to_null_publisher()
+    public void SetInner_null_detaches_the_previous_publisher()
     {
         var spy = new SpyPublisher();
         var publisher = new DeferredServiceLevelPublisher();
@@ -39,8 +39,139 @@ public class DeferredServiceLevelPublisherTests
 
         publisher.Publish(100);
 
-        // The spy should NOT have received the publish after revert.
-        spy.LastPublished.ShouldBe((byte)0, "spy should not have been updated after revert to null");
+        // The previous publisher must not receive an unbound request.
+        spy.LastPublished.ShouldBe((byte)0, "the previous publisher should not be updated after unbinding");
+    }
+
+    [Fact]
+    public void Binding_without_a_request_does_not_invent_a_level()
+    {
+        var received = new List<byte>();
+        new DeferredServiceLevelPublisher().SetInner(new CallbackPublisher(received.Add));
+        received.ShouldBeEmpty();
+    }
+
+    [Theory]
+    [InlineData((byte)100)]
+    [InlineData((byte)0)]
+    public void Binding_replays_the_latest_request_including_zero(byte latest)
+    {
+        var publisher = new DeferredServiceLevelPublisher();
+        publisher.Publish(240);
+        publisher.Publish(latest);
+        var received = new List<byte>();
+        publisher.SetInner(new CallbackPublisher(received.Add));
+        received.ShouldBe(new[] { latest });
+    }
+
+    [Fact]
+    public void Unbinding_retains_latest_for_rebinding()
+    {
+        var publisher = new DeferredServiceLevelPublisher();
+        var old = new List<byte>();
+        publisher.SetInner(new CallbackPublisher(old.Add));
+        publisher.Publish(240);
+        publisher.SetInner(null);
+        publisher.Publish(0);
+        var replacement = new List<byte>();
+        publisher.SetInner(new CallbackPublisher(replacement.Add));
+        old.ShouldBe(new byte[] { 240 });
+        replacement.ShouldBe(new byte[] { 0 });
+    }
+
+    [Fact]
+    public void Failed_publish_retains_the_requested_value_for_replay()
+    {
+        var publisher = new DeferredServiceLevelPublisher();
+        publisher.SetInner(new CallbackPublisher(_ => throw new InvalidOperationException()));
+        Should.Throw<InvalidOperationException>(() => publisher.Publish(0));
+        var received = new List<byte>();
+        publisher.SetInner(new CallbackPublisher(received.Add));
+        received.ShouldBe(new byte[] { 0 });
+    }
+
+    [Fact]
+    public void Failed_replay_keeps_candidate_bound_for_later_publish()
+    {
+        var publisher = new DeferredServiceLevelPublisher();
+        publisher.Publish(100);
+        var fail = true;
+        var received = new List<byte>();
+        var candidate = new CallbackPublisher(value =>
+        {
+            if (fail) throw new InvalidOperationException();
+            received.Add(value);
+        });
+        Should.Throw<InvalidOperationException>(() => publisher.SetInner(candidate));
+        fail = false;
+        publisher.Publish(0);
+        publisher.SetInner(null);
+        publisher.SetInner(candidate);
+        received.ShouldBe(new byte[] { 0, 0 });
+    }
+
+    [Fact]
+    public void Failed_replay_retains_latest_for_a_replacement_publisher()
+    {
+        var publisher = new DeferredServiceLevelPublisher();
+        publisher.Publish(100);
+        Should.Throw<InvalidOperationException>(() => publisher.SetInner(
+            new CallbackPublisher(_ => throw new InvalidOperationException())));
+        var received = new List<byte>();
+        publisher.SetInner(new CallbackPublisher(received.Add));
+        received.ShouldBe(new byte[] { 100 });
+    }
+
+    [Fact]
+    public async Task Binding_waits_for_inflight_publish_then_replays_latest()
+    {
+        var publisher = new DeferredServiceLevelPublisher();
+        using var entered = new ManualResetEventSlim();
+        using var release = new ManualResetEventSlim();
+        publisher.SetInner(new CallbackPublisher(_ =>
+        {
+            entered.Set();
+            release.Wait(TimeSpan.FromSeconds(5), TestContext.Current.CancellationToken).ShouldBeTrue();
+        }));
+        var publish = Task.Run(() => publisher.Publish(100), TestContext.Current.CancellationToken);
+        var received = new List<byte>();
+        using var bindingStarted = new ManualResetEventSlim();
+        Exception? bindingError = null;
+        var binding = new Thread(() =>
+        {
+            bindingStarted.Set();
+            try { publisher.SetInner(new CallbackPublisher(received.Add)); }
+            catch (Exception ex) { bindingError = ex; }
+        });
+        var bindingWasStarted = false;
+        try
+        {
+            entered.Wait(TimeSpan.FromSeconds(5), TestContext.Current.CancellationToken).ShouldBeTrue();
+            binding.Start();
+            bindingWasStarted = true;
+            bindingStarted.Wait(TimeSpan.FromSeconds(5), TestContext.Current.CancellationToken).ShouldBeTrue();
+            // Past the marker, SetInner is the thread's only potentially blocking operation.
+            // Observe actual contention rather than assuming Task.Run has entered the adapter.
+            SpinWait.SpinUntil(() => !binding.IsAlive ||
+                (binding.ThreadState & System.Threading.ThreadState.WaitSleepJoin) != 0,
+                TimeSpan.FromSeconds(3)).ShouldBeTrue();
+            (binding.ThreadState & System.Threading.ThreadState.WaitSleepJoin).ShouldNotBe(
+                (System.Threading.ThreadState)0, "binding must wait for the in-flight delivery");
+        }
+        finally
+        {
+            release.Set();
+            if (bindingWasStarted) binding.Join(TimeSpan.FromSeconds(5)).ShouldBeTrue();
+            await publish.WaitAsync(TimeSpan.FromSeconds(5), TestContext.Current.CancellationToken);
+        }
+        bindingError.ShouldBeNull();
+        publisher.Publish(0);
+        received.ShouldBe(new byte[] { 100, 0 });
+    }
+
+    private sealed class CallbackPublisher(Action<byte> publish) : IServiceLevelPublisher
+    {
+        public void Publish(byte serviceLevel) => publish(serviceLevel);
     }
 
     // ---- test double ----

tests/Core/ZB.MOM.WW.OtOpcUa.Core.AlarmHistorian.Tests/AlarmHistorianEventIdentityTests.cs

@@ -0,0 +1,78 @@
+using System.Text.Json;
+using Xunit;
+using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
+
+namespace ZB.MOM.WW.OtOpcUa.Core.AlarmHistorian.Tests;
+
+public sealed class AlarmHistorianEventIdentityTests
+{
+    internal const string Payload = """{"AlarmId":"A","EquipmentPath":"S","AlarmName":"N","AlarmTypeName":"LimitAlarm","Severity":2,"EventKind":"Activated","Message":"m","User":"system","Comment":null,"TimestampUtc":"2026-01-01T00:00:00Z"}""";
+    internal const string QueueId = "52C8292271C629CCC2BFE4EC8C16A5E70CA7456C7D88BC282ABCEDF636F05901";
+    private static AlarmHistorianEvent Event() => new("A", "S", "N", "LimitAlarm", AlarmSeverity.High,
+        "Activated", "m", "system", null, new DateTime(2026, 1, 1, 0, 0, 0, DateTimeKind.Utc));
+
+    [Theory]
+    [InlineData(QueueId, "48bca092-b99d-8ea4-a2de-8b30d80a481a")]
+    [InlineData("persisted-row-v1", "39473040-002b-876c-a1e6-9a9ad4a60ded")]
+    public void Persisted_keys_have_frozen_versioned_delivery_GUID_vectors(string key, string expected)
+    {
+        Assert.Equal(Guid.Parse(expected), AlarmDeliveryIds.FromQueueId(key));
+    }
+
+    [Fact]
+    public void Delivery_identity_is_nonzero_and_a_copy_preserves_it()
+    {
+        var original = Event();
+        Assert.NotEqual(Guid.Empty, original.DeliveryId);
+        Assert.Equal(original.DeliveryId, (original with { }).DeliveryId);
+        Assert.Equal(original.DeliveryId, (original with { Comment = "copy" }).DeliveryId);
+    }
+
+    [Fact]
+    public void Business_equality_hash_and_historical_JSON_ignore_delivery_identity()
+    {
+        var first = Event();
+        var second = Event();
+        Assert.NotEqual(first.DeliveryId, second.DeliveryId);
+        Assert.Equal(first, second);
+        Assert.True(first == second);
+        Assert.Equal(first.GetHashCode(), second.GetHashCode());
+        Assert.Single(new HashSet<AlarmHistorianEvent> { first, second });
+        Assert.Equal(Payload, JsonSerializer.Serialize(first));
+        Assert.Equal(Payload, JsonSerializer.Serialize(second));
+        Assert.Equal(QueueId, AlarmSfSchema.DeriveId(Payload));
+        Assert.Equal(first, JsonSerializer.Deserialize<AlarmHistorianEvent>(Payload));
+    }
+
+    [Theory]
+    [InlineData("AlarmId")]
+    [InlineData("EquipmentPath")]
+    [InlineData("AlarmName")]
+    [InlineData("AlarmTypeName")]
+    [InlineData("Severity")]
+    [InlineData("EventKind")]
+    [InlineData("Message")]
+    [InlineData("User")]
+    [InlineData("Comment")]
+    [InlineData("TimestampUtc")]
+    public void Each_business_field_still_participates_in_equality(string field)
+    {
+        var first = Event();
+        var changed = field switch
+        {
+            "AlarmId" => first with { AlarmId = "B" },
+            "EquipmentPath" => first with { EquipmentPath = "T" },
+            "AlarmName" => first with { AlarmName = "Other" },
+            "AlarmTypeName" => first with { AlarmTypeName = "DiscreteAlarm" },
+            "Severity" => first with { Severity = AlarmSeverity.Low },
+            "EventKind" => first with { EventKind = "Cleared" },
+            "Message" => first with { Message = "other" },
+            "User" => first with { User = "operator" },
+            "Comment" => first with { Comment = "note" },
+            "TimestampUtc" => first with { TimestampUtc = first.TimestampUtc.AddTicks(1) },
+            _ => throw new ArgumentOutOfRangeException(nameof(field)),
+        };
+        Assert.NotEqual(first, changed);
+        Assert.False(first == changed);
+    }
+}

tests/Core/ZB.MOM.WW.OtOpcUa.Core.AlarmHistorian.Tests/LocalDbStoreAndForwardSinkTests.cs

@@ -123,6 +123,69 @@ public sealed class LocalDbStoreAndForwardSinkTests : IDisposable
         Comment: null,
         TimestampUtc: ts ?? DateTime.UtcNow);
 
+    [Fact]
+    public async Task Queued_delivery_identity_survives_retry_reopen_and_payload_changes()
+    {
+        // The persisted row key is authoritative: reserializing today's payload must never change a retry's identity.
+        var expected = Guid.Parse("48bca092-b99d-8ea4-a2de-8b30d80a481a");
+        var evt = System.Text.Json.JsonSerializer.Deserialize<AlarmHistorianEvent>(AlarmHistorianEventIdentityTests.Payload)!;
+        var writer = new FakeWriter { DefaultOutcome = HistorianWriteOutcome.RetryPlease };
+        using (var sink = new LocalDbStoreAndForwardSink(Db, writer, _log))
+        {
+            await sink.EnqueueAsync(evt, TestContext.Current.CancellationToken);
+            var row = (await Db.QueryAsync("SELECT id, payload_json FROM alarm_sf_events",
+                r => (Id: r.GetString(0), Payload: r.GetString(1)), null, TestContext.Current.CancellationToken)).Single();
+            row.Id.ShouldBe(AlarmHistorianEventIdentityTests.QueueId);
+            row.Payload.ShouldBe(AlarmHistorianEventIdentityTests.Payload);
+            await sink.DrainOnceAsync(TestContext.Current.CancellationToken);
+            await sink.DrainOnceAsync(TestContext.Current.CancellationToken);
+            writer.Batches.ShouldAllBe(batch => batch.Single().DeliveryId == expected);
+            await Db.ExecuteAsync("UPDATE alarm_sf_events SET payload_json = @Payload",
+                new { Payload = System.Text.Json.JsonSerializer.Serialize(evt with { Comment = "edited" }) },
+                TestContext.Current.CancellationToken);
+        }
+        _provider!.Dispose();
+        _provider = null;
+        _db = null;
+        var reopenedWriter = new FakeWriter();
+        using var reopened = new LocalDbStoreAndForwardSink(Db, reopenedWriter, _log);
+        await reopened.DrainOnceAsync(TestContext.Current.CancellationToken);
+        reopenedWriter.Batches.Single().Single().DeliveryId.ShouldBe(expected);
+        reopenedWriter.Batches.Single().Single().Comment.ShouldBe("edited");
+        reopened.GetStatus().QueueDepth.ShouldBe(0);
+    }
+
+    [Fact]
+    public async Task Independently_created_identical_events_keep_the_historical_queue_key()
+    {
+        var first = System.Text.Json.JsonSerializer.Deserialize<AlarmHistorianEvent>(AlarmHistorianEventIdentityTests.Payload)!;
+        var second = System.Text.Json.JsonSerializer.Deserialize<AlarmHistorianEvent>(AlarmHistorianEventIdentityTests.Payload)!;
+        first.DeliveryId.ShouldNotBe(second.DeliveryId);
+        using var sink = new LocalDbStoreAndForwardSink(Db, new FakeWriter(), _log);
+        await sink.EnqueueAsync(first, TestContext.Current.CancellationToken);
+        await sink.EnqueueAsync(second, TestContext.Current.CancellationToken);
+        sink.GetStatus().QueueDepth.ShouldBe(1);
+        (await Db.QueryAsync("SELECT id FROM alarm_sf_events", r => r.GetString(0), null, TestContext.Current.CancellationToken))
+            .ShouldBe([AlarmHistorianEventIdentityTests.QueueId]);
+    }
+
+    [Fact]
+    public async Task Distinct_business_events_at_the_same_time_have_distinct_queued_delivery_ids()
+    {
+        var first = Event("A", new DateTime(2026, 1, 1, 0, 0, 0, DateTimeKind.Utc));
+        var second = first with { EventKind = "Cleared", Comment = "different transition" };
+        var writer = new FakeWriter();
+        using var sink = new LocalDbStoreAndForwardSink(Db, writer, _log);
+        await sink.EnqueueAsync(first, TestContext.Current.CancellationToken);
+        await sink.EnqueueAsync(second, TestContext.Current.CancellationToken);
+        sink.GetStatus().QueueDepth.ShouldBe(2);
+        await sink.DrainOnceAsync(TestContext.Current.CancellationToken);
+        var delivered = writer.Batches.Single();
+        delivered.Count.ShouldBe(2);
+        delivered.ShouldAllBe(evt => evt.DeliveryId != Guid.Empty);
+        delivered.Select(evt => evt.DeliveryId).Distinct().Count().ShouldBe(2);
+    }
+
     /// <summary>Verifies that acknowledged events are removed from the queue.</summary>
     [Fact]
     public async Task EnqueueThenDrain_Ack_removes_row()
@@ -297,6 +360,11 @@ public sealed class LocalDbStoreAndForwardSinkTests : IDisposable
         var status = sink.GetStatus();
         status.QueueDepth.ShouldBe(1);
         status.DeadLetterDepth.ShouldBe(0);
+        var originalDeliveryId = writer.Batches.Single().Single().DeliveryId;
+        await sink.DrainOnceAsync(TestContext.Current.CancellationToken);
+        writer.Batches.Count.ShouldBe(2);
+        writer.Batches.Last().Single().DeliveryId.ShouldBe(originalDeliveryId);
+        sink.GetStatus().QueueDepth.ShouldBe(0);
     }
 
     /// <summary>Verifies that the exponential backoff ladder caps at 60 seconds.</summary>

tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.AbCip.Tests/AbCipReinitializeReleasesStateTests.cs

@@ -1,4 +1,5 @@
 using Shouldly;
+using ZB.MOM.WW.OtOpcUa.Tests;
 using Xunit;
 
 namespace ZB.MOM.WW.OtOpcUa.Driver.AbCip.Tests;
@@ -8,7 +9,7 @@ namespace ZB.MOM.WW.OtOpcUa.Driver.AbCip.Tests;
 ///     orphaned the previous state's probe loop + handles. It now releases them first.
 /// </summary>
 [Trait("Category", "Unit")]
-public sealed class AbCipReinitializeReleasesStateTests
+public sealed class AbCipReinitializeReleasesStateTests(ITestOutputHelper output)
 {
     /// <summary>Initialize twice: the first state's probe loop has stopped and its probe handle is disposed.</summary>
     [Fact]
@@ -27,28 +28,94 @@ public sealed class AbCipReinitializeReleasesStateTests
                 ProbeTagPath = "@raw_cpu_type",
             },
         }, "drv-reinit", factory);
-        await drv.InitializeAsync("{}", CancellationToken.None);
+        await FixtureCleanup.RunAsync(async () =>
+        {
+            await drv.InitializeAsync("{}", CancellationToken.None);
 
-        var first = drv.GetDeviceState(host)!;
-        // Positive control: the first probe loop is running and has initialised its runtime.
+            var first = drv.GetDeviceState(host)!;
+            // Positive control: the first probe loop is running and has initialised its runtime.
+            var deadline = DateTime.UtcNow.AddSeconds(5);
+            while (!(factory.Tags.TryGetValue("@raw_cpu_type", out var t) && t.InitializeCount > 0))
+            {
+                if (DateTime.UtcNow > deadline) throw new TimeoutException("probe never started");
+                await Task.Delay(10);
+            }
+            var firstTag = factory.Tags["@raw_cpu_type"];
+            var firstLoop = first.ProbeTask;
+            _ = firstLoop.ShouldNotBeNull();
+            firstLoop!.IsCompleted.ShouldBeFalse();
+            firstTag.Disposed.ShouldBeFalse();
+
+            await drv.InitializeAsync("{}", CancellationToken.None);
+
+            firstLoop.IsCompleted.ShouldBeTrue(); // joined
+            firstTag.Disposed.ShouldBeTrue();
+            drv.GetDeviceState(host).ShouldNotBeSameAs(first);
+        }, () => drv.ShutdownAsync(CancellationToken.None),
+            failure => output.WriteLine($"AbCip SecondInitialize cleanup failed: {failure}"));
+    }
+
+    /// <summary>Failure after a live first or replacement state must release both owned resources.</summary>
+    [Theory]
+    [InlineData(false)]
+    [InlineData(true)]
+    public async Task Body_failure_releases_live_probe_resources(bool afterReplacement)
+    {
+        const string host = "ab://10.0.0.5/1,0";
+        var factory = new FakeAbCipTagFactory();
+        var drv = new AbCipDriver(new AbCipDriverOptions
+        {
+            Devices = [new AbCipDeviceOptions(host)],
+            Probe = new AbCipProbeOptions
+            {
+                Enabled = true,
+                Interval = TimeSpan.FromMilliseconds(20),
+                Timeout = TimeSpan.FromMilliseconds(200),
+                ProbeTagPath = "@raw_cpu_type",
+            },
+        }, "drv-reinit", factory);
+        Task? firstLoop = null, secondLoop = null;
+        FakeAbCipTag? firstTag = null, secondTag = null;
+        var sentinel = new InvalidOperationException("body sentinel");
+        var actual = await Should.ThrowAsync<InvalidOperationException>(() => FixtureCleanup.RunAsync(async () =>
+        {
+            await drv.InitializeAsync("{}", CancellationToken.None);
+            await WaitForProbeAsync(() => factory.Tags.TryGetValue("@raw_cpu_type", out var tag) && tag.InitializeCount > 0);
+            firstTag = factory.Tags["@raw_cpu_type"];
+            firstLoop = drv.GetDeviceState(host)!.ProbeTask;
+            _ = firstLoop.ShouldNotBeNull();
+            firstLoop!.IsCompleted.ShouldBeFalse();
+            firstTag.Disposed.ShouldBeFalse();
+            if (afterReplacement)
+            {
+                await drv.InitializeAsync("{}", CancellationToken.None);
+                await WaitForProbeAsync(() => factory.Tags.TryGetValue("@raw_cpu_type", out var tag) && !ReferenceEquals(tag, firstTag) && tag.InitializeCount > 0);
+                secondTag = factory.Tags["@raw_cpu_type"];
+                secondLoop = drv.GetDeviceState(host)!.ProbeTask;
+                _ = secondLoop.ShouldNotBeNull();
+                secondLoop!.IsCompleted.ShouldBeFalse();
+            }
+            throw sentinel;
+        }, () => drv.ShutdownAsync(CancellationToken.None),
+            failure => output.WriteLine($"AbCip failure fixture cleanup failed: {failure}")));
+        actual.ShouldBeSameAs(sentinel);
+        firstLoop!.IsCompleted.ShouldBeTrue();
+        firstTag!.Disposed.ShouldBeTrue();
+        if (afterReplacement)
+        {
+            secondLoop!.IsCompleted.ShouldBeTrue();
+            secondTag!.Disposed.ShouldBeTrue();
+        }
+    }
+
+    private static async Task WaitForProbeAsync(Func<bool> condition)
+    {
         var deadline = DateTime.UtcNow.AddSeconds(5);
-        while (!(factory.Tags.TryGetValue("@raw_cpu_type", out var t) && t.InitializeCount > 0))
+        while (!condition())
         {
             if (DateTime.UtcNow > deadline) throw new TimeoutException("probe never started");
             await Task.Delay(10);
         }
-        var firstTag = factory.Tags["@raw_cpu_type"];
-        var firstLoop = first.ProbeTask;
-        firstLoop.ShouldNotBeNull();
-        firstLoop!.IsCompleted.ShouldBeFalse();
-        firstTag.Disposed.ShouldBeFalse();
-
-        await drv.InitializeAsync("{}", CancellationToken.None);
-
-        firstLoop.IsCompleted.ShouldBeTrue(); // joined
-        firstTag.Disposed.ShouldBeTrue();
-        drv.GetDeviceState(host).ShouldNotBeSameAs(first);
-
-        await drv.ShutdownAsync(CancellationToken.None);
     }
+
 }

tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.AbCip.Tests/FixtureCleanupTests.cs

@@ -0,0 +1,92 @@
+using Shouldly;
+using Xunit;
+using ZB.MOM.WW.OtOpcUa.Tests;
+
+namespace ZB.MOM.WW.OtOpcUa.Driver.AbCip.Tests;
+
+[Trait("Category", "Unit")]
+public sealed class FixtureCleanupTests
+{
+    [Fact]
+    public async Task Successful_body_runs_one_awaited_cleanup()
+    {
+        var calls = 0;
+        var release = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
+        var run = FixtureCleanup.RunAsync(() => Task.CompletedTask, async () =>
+        {
+            calls++;
+            await release.Task;
+        }, _ => throw new Exception("unexpected report"));
+
+        run.IsCompleted.ShouldBeFalse();
+        release.SetResult();
+        await run;
+        calls.ShouldBe(1);
+    }
+
+    [Fact]
+    public async Task Body_fault_remains_same_exception_and_original_throwing_stack()
+    {
+        var primary = new InvalidOperationException("body");
+        var cleanupCalls = 0;
+        var actual = await Should.ThrowAsync<InvalidOperationException>(() => FixtureCleanup.RunAsync(
+            () => ThrowFromBody(primary),
+            () => { cleanupCalls++; return Task.CompletedTask; },
+            _ => throw new Exception("unexpected report")));
+
+        actual.ShouldBeSameAs(primary);
+        actual.StackTrace.ShouldNotBeNull().ShouldContain(nameof(ThrowFromBody));
+        cleanupCalls.ShouldBe(1);
+    }
+
+    [Fact]
+    public async Task Cleanup_only_fault_is_thrown_exactly()
+    {
+        var cleanup = new ApplicationException("cleanup");
+        var actual = await Should.ThrowAsync<ApplicationException>(() => FixtureCleanup.RunAsync(
+            () => Task.CompletedTask,
+            () => Task.FromException(cleanup),
+            _ => throw new Exception("unexpected report")));
+        actual.ShouldBeSameAs(cleanup);
+    }
+
+    [Fact]
+    public async Task Both_faults_preserve_primary_and_record_secondary_once()
+    {
+        var primary = new InvalidOperationException("body");
+        var cleanup = new ApplicationException("cleanup");
+        var reported = new List<Exception>();
+        var actual = await Should.ThrowAsync<InvalidOperationException>(() => FixtureCleanup.RunAsync(
+            () => ThrowFromBody(primary),
+            () => Task.FromException(cleanup),
+            reported.Add));
+
+        actual.ShouldBeSameAs(primary);
+        actual.StackTrace.ShouldNotBeNull().ShouldContain(nameof(ThrowFromBody));
+        actual.Data[FixtureCleanup.SecondaryExceptionKey].ShouldBeSameAs(cleanup);
+        reported.Count.ShouldBe(1);
+        reported[0].ShouldBeSameAs(cleanup);
+    }
+
+    [Fact]
+    public async Task Reporting_fault_does_not_replace_primary()
+    {
+        var primary = new InvalidOperationException("body");
+        var cleanup = new ApplicationException("cleanup");
+        var reportCalls = 0;
+        var actual = await Should.ThrowAsync<InvalidOperationException>(() => FixtureCleanup.RunAsync(
+            () => ThrowFromBody(primary),
+            () => Task.FromException(cleanup),
+            _ => { reportCalls++; throw new Exception("report"); }));
+
+        actual.ShouldBeSameAs(primary);
+        actual.Data[FixtureCleanup.SecondaryExceptionKey].ShouldBeSameAs(cleanup);
+        reportCalls.ShouldBe(1);
+    }
+
+    private static async Task ThrowFromBody(Exception failure)
+    {
+        await Task.Yield();
+        throw failure;
+    }
+}

tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.AbCip.Tests/ZB.MOM.WW.OtOpcUa.Driver.AbCip.Tests.csproj

@@ -38,4 +38,7 @@
         <Compile Include="..\ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport.Tests\Fakes\FakeCipTarget.cs" Link="Fakes\FakeCipTarget.cs" />
     </ItemGroup>
 
+    <ItemGroup>
+        <Compile Include="..\..\TestSupport\FixtureCleanup.cs" Link="TestSupport\FixtureCleanup.cs" />
+    </ItemGroup>
 </Project>

tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.AbLegacy.Tests/AbLegacyReinitializeReleasesStateTests.cs

@@ -1,4 +1,5 @@
 using Shouldly;
+using ZB.MOM.WW.OtOpcUa.Tests;
 using Xunit;
 using ZB.MOM.WW.OtOpcUa.Driver.AbLegacy.PlcFamilies;
 
@@ -10,7 +11,7 @@ namespace ZB.MOM.WW.OtOpcUa.Driver.AbLegacy.Tests;
 ///     probe loop + runtime. It now releases them first.
 /// </summary>
 [Trait("Category", "Unit")]
-public sealed class AbLegacyReinitializeReleasesStateTests
+public sealed class AbLegacyReinitializeReleasesStateTests(ITestOutputHelper output)
 {
     /// <summary>Initialize twice: the first state's probe loop has stopped and its probe runtime is disposed.</summary>
     [Fact]
@@ -23,28 +24,88 @@ public sealed class AbLegacyReinitializeReleasesStateTests
             Devices = [new AbLegacyDeviceOptions(host, AbLegacyPlcFamily.Slc500)],
             Probe = new AbLegacyProbeOptions { Enabled = true, Interval = TimeSpan.FromMilliseconds(20), ProbeAddress = "S:0" },
         }, "drv-reinit", factory);
-        await drv.InitializeAsync("{}", CancellationToken.None);
+        await FixtureCleanup.RunAsync(async () =>
+        {
+            await drv.InitializeAsync("{}", CancellationToken.None);
 
-        var first = drv.GetDeviceState(host)!;
-        // Positive control: the first state's probe loop is running and has created + initialised its runtime.
+            var first = drv.GetDeviceState(host)!;
+            // Positive control: the first state's probe loop is running and has created + initialised its runtime.
+            var deadline = DateTime.UtcNow.AddSeconds(5);
+            while (!(factory.Tags.TryGetValue("S:0", out var t) && t.InitializeCount > 0))
+            {
+                if (DateTime.UtcNow > deadline) throw new TimeoutException("probe never started");
+                await Task.Delay(10);
+            }
+            var firstTag = factory.Tags["S:0"];
+            var firstLoop = first.ProbeTask;
+            _ = firstLoop.ShouldNotBeNull();
+            firstLoop!.IsCompleted.ShouldBeFalse();
+            firstTag.Disposed.ShouldBeFalse();
+
+            await drv.InitializeAsync("{}", CancellationToken.None);
+
+            firstLoop.IsCompleted.ShouldBeTrue(); // joined
+            firstTag.Disposed.ShouldBeTrue();
+            drv.GetDeviceState(host).ShouldNotBeSameAs(first);
+        }, () => drv.ShutdownAsync(CancellationToken.None),
+            failure => output.WriteLine($"AbLegacy SecondInitialize cleanup failed: {failure}"));
+    }
+
+    /// <summary>Failure after a live first or replacement state must release both owned resources.</summary>
+    [Theory]
+    [InlineData(false)]
+    [InlineData(true)]
+    public async Task Body_failure_releases_live_probe_resources(bool afterReplacement)
+    {
+        const string host = "ab://10.0.0.5/1,0";
+        var factory = new FakeAbLegacyTagFactory();
+        var drv = new AbLegacyDriver(new AbLegacyDriverOptions
+        {
+            Devices = [new AbLegacyDeviceOptions(host, AbLegacyPlcFamily.Slc500)],
+            Probe = new AbLegacyProbeOptions { Enabled = true, Interval = TimeSpan.FromMilliseconds(20), ProbeAddress = "S:0" },
+        }, "drv-reinit", factory);
+        Task? firstLoop = null, secondLoop = null;
+        FakeAbLegacyTag? firstTag = null, secondTag = null;
+        var sentinel = new InvalidOperationException("body sentinel");
+        var actual = await Should.ThrowAsync<InvalidOperationException>(() => FixtureCleanup.RunAsync(async () =>
+        {
+            await drv.InitializeAsync("{}", CancellationToken.None);
+            await WaitForProbeAsync(() => factory.Tags.TryGetValue("S:0", out var tag) && tag.InitializeCount > 0);
+            firstTag = factory.Tags["S:0"];
+            firstLoop = drv.GetDeviceState(host)!.ProbeTask;
+            _ = firstLoop.ShouldNotBeNull();
+            firstLoop!.IsCompleted.ShouldBeFalse();
+            firstTag.Disposed.ShouldBeFalse();
+            if (afterReplacement)
+            {
+                await drv.InitializeAsync("{}", CancellationToken.None);
+                await WaitForProbeAsync(() => factory.Tags.TryGetValue("S:0", out var tag) && !ReferenceEquals(tag, firstTag) && tag.InitializeCount > 0);
+                secondTag = factory.Tags["S:0"];
+                secondLoop = drv.GetDeviceState(host)!.ProbeTask;
+                _ = secondLoop.ShouldNotBeNull();
+                secondLoop!.IsCompleted.ShouldBeFalse();
+            }
+            throw sentinel;
+        }, () => drv.ShutdownAsync(CancellationToken.None),
+            failure => output.WriteLine($"AbLegacy failure fixture cleanup failed: {failure}")));
+        actual.ShouldBeSameAs(sentinel);
+        firstLoop!.IsCompleted.ShouldBeTrue();
+        firstTag!.Disposed.ShouldBeTrue();
+        if (afterReplacement)
+        {
+            secondLoop!.IsCompleted.ShouldBeTrue();
+            secondTag!.Disposed.ShouldBeTrue();
+        }
+    }
+
+    private static async Task WaitForProbeAsync(Func<bool> condition)
+    {
         var deadline = DateTime.UtcNow.AddSeconds(5);
-        while (!(factory.Tags.TryGetValue("S:0", out var t) && t.InitializeCount > 0))
+        while (!condition())
         {
             if (DateTime.UtcNow > deadline) throw new TimeoutException("probe never started");
             await Task.Delay(10);
         }
-        var firstTag = factory.Tags["S:0"];
-        var firstLoop = first.ProbeTask;
-        firstLoop.ShouldNotBeNull();
-        firstLoop!.IsCompleted.ShouldBeFalse();
-        firstTag.Disposed.ShouldBeFalse();
-
-        await drv.InitializeAsync("{}", CancellationToken.None);
-
-        firstLoop.IsCompleted.ShouldBeTrue(); // joined
-        firstTag.Disposed.ShouldBeTrue();
-        drv.GetDeviceState(host).ShouldNotBeSameAs(first);
-
-        await drv.ShutdownAsync(CancellationToken.None);
     }
+
 }

tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.AbLegacy.Tests/ZB.MOM.WW.OtOpcUa.Driver.AbLegacy.Tests.csproj

@@ -38,4 +38,7 @@
         <None Include="..\ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport.Tests\Captures\*.hex" Link="Captures\%(Filename)%(Extension)" CopyToOutputDirectory="PreserveNewest"/>
     </ItemGroup>
 
+    <ItemGroup>
+        <Compile Include="..\..\TestSupport\FixtureCleanup.cs" Link="TestSupport\FixtureCleanup.cs" />
+    </ItemGroup>
 </Project>

tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.FOCAS.Tests/FocasReinitializeReleasesStateTests.cs

@@ -1,4 +1,5 @@
 using Shouldly;
+using ZB.MOM.WW.OtOpcUa.Tests;
 using Xunit;
 using ZB.MOM.WW.OtOpcUa.Driver.FOCAS;
 
@@ -10,7 +11,7 @@ namespace ZB.MOM.WW.OtOpcUa.Driver.FOCAS.Tests;
 ///     probe loop + client. It now releases them first.
 /// </summary>
 [Trait("Category", "Unit")]
-public sealed class FocasReinitializeReleasesStateTests
+public sealed class FocasReinitializeReleasesStateTests(ITestOutputHelper output)
 {
     /// <summary>Initialize twice: the first state's probe loop has finished and its client is disposed.</summary>
     [Fact]
@@ -25,25 +26,78 @@ public sealed class FocasReinitializeReleasesStateTests
             Probe = new FocasProbeOptions { Enabled = true, Interval = TimeSpan.FromMilliseconds(20), Timeout = TimeSpan.FromSeconds(2) },
         };
         var drv = new FocasDriver(options, "focas-reinit", factory);
-        await drv.InitializeAsync("{}", CancellationToken.None);
+        await FixtureCleanup.RunAsync(async () =>
+        {
+            await drv.InitializeAsync("{}", CancellationToken.None);
 
-        var first = drv.GetDeviceState(host)!;
-        // Positive control: the first state's probe loop is demonstrably running and holds a live client.
-        await WaitUntilAsync(() => factory.Clients.Count > 0 && factory.Clients[0].IsConnected);
-        var firstLoop = first.ProbeTask;
-        firstLoop.ShouldNotBeNull();
-        firstLoop!.IsCompleted.ShouldBeFalse();
-        var firstClient = factory.Clients[0];
+            var first = drv.GetDeviceState(host)!;
+            // Positive control: the first state's probe loop is demonstrably running and holds a live client.
+            await WaitUntilAsync(() => factory.Clients.Count > 0 && factory.Clients[0].IsConnected);
+            var firstLoop = first.ProbeTask;
+            _ = firstLoop.ShouldNotBeNull();
+            firstLoop!.IsCompleted.ShouldBeFalse();
+            var firstClient = factory.Clients[0];
 
-        await drv.InitializeAsync("{}", CancellationToken.None);
+            await drv.InitializeAsync("{}", CancellationToken.None);
 
-        firstLoop.IsCompleted.ShouldBeTrue(); // joined
-        firstClient.DisposeCount.ShouldBeGreaterThanOrEqualTo(1);
+            firstLoop.IsCompleted.ShouldBeTrue(); // joined
+            firstClient.DisposeCount.ShouldBeGreaterThanOrEqualTo(1);
+            firstClient.IsConnected.ShouldBeFalse();
+            drv.GetDeviceState(host).ShouldNotBeSameAs(first);
+            _ = drv.GetDeviceState(host)!.ProbeTask.ShouldNotBeNull(); // the new state is running
+        }, () => drv.ShutdownAsync(CancellationToken.None),
+            failure => output.WriteLine($"FOCAS SecondInitialize cleanup failed: {failure}"));
+    }
+
+
+    /// <summary>Failure after a live first or replacement state must release both owned resources.</summary>
+    [Theory]
+    [InlineData(false)]
+    [InlineData(true)]
+    public async Task Body_failure_releases_live_probe_resources(bool afterReplacement)
+    {
+        const string host = "focas://10.0.0.5:8193";
+        var factory = new FakeFocasClientFactory();
+        var options = new FocasDriverOptions
+        {
+            Devices = [new FocasDeviceOptions(host)],
+            RawTags = [FocasRawTags.Entry(new FocasTagDefinition("R", host, "R100", FocasDataType.Byte))],
+            Probe = new FocasProbeOptions { Enabled = true, Interval = TimeSpan.FromMilliseconds(20), Timeout = TimeSpan.FromSeconds(2) },
+        };
+        var drv = new FocasDriver(options, "focas-reinit", factory);
+        Task? firstLoop = null, secondLoop = null;
+        FakeFocasClient? firstClient = null, secondClient = null;
+        var sentinel = new InvalidOperationException("body sentinel");
+        var actual = await Should.ThrowAsync<InvalidOperationException>(() => FixtureCleanup.RunAsync(async () =>
+        {
+            await drv.InitializeAsync("{}", CancellationToken.None);
+            await WaitUntilAsync(() => factory.Clients.Count > 0 && factory.Clients[0].IsConnected);
+            firstClient = factory.Clients[0];
+            firstLoop = drv.GetDeviceState(host)!.ProbeTask;
+            _ = firstLoop.ShouldNotBeNull();
+            firstLoop!.IsCompleted.ShouldBeFalse();
+            if (afterReplacement)
+            {
+                await drv.InitializeAsync("{}", CancellationToken.None);
+                await WaitUntilAsync(() => factory.Clients.Count > 1 && factory.Clients[1].IsConnected);
+                secondClient = factory.Clients[1];
+                secondLoop = drv.GetDeviceState(host)!.ProbeTask;
+                _ = secondLoop.ShouldNotBeNull();
+                secondLoop!.IsCompleted.ShouldBeFalse();
+            }
+            throw sentinel;
+        }, () => drv.ShutdownAsync(CancellationToken.None),
+            failure => output.WriteLine($"FOCAS failure fixture cleanup failed: {failure}")));
+        actual.ShouldBeSameAs(sentinel);
+        firstLoop!.IsCompleted.ShouldBeTrue();
+        firstClient!.DisposeCount.ShouldBeGreaterThanOrEqualTo(1);
         firstClient.IsConnected.ShouldBeFalse();
-        drv.GetDeviceState(host).ShouldNotBeSameAs(first);
-        drv.GetDeviceState(host)!.ProbeTask.ShouldNotBeNull(); // the new state is running
-
-        await drv.ShutdownAsync(CancellationToken.None);
+        if (afterReplacement)
+        {
+            secondLoop!.IsCompleted.ShouldBeTrue();
+            secondClient!.DisposeCount.ShouldBeGreaterThanOrEqualTo(1);
+            secondClient.IsConnected.ShouldBeFalse();
+        }
     }
 
     private static async Task WaitUntilAsync(Func<bool> cond)

tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.FOCAS.Tests/ZB.MOM.WW.OtOpcUa.Driver.FOCAS.Tests.csproj

@@ -34,4 +34,7 @@
         <ProjectReference Include="..\..\..\src\Drivers\ZB.MOM.WW.OtOpcUa.Driver.FOCAS\ZB.MOM.WW.OtOpcUa.Driver.FOCAS.csproj"/>
     </ItemGroup>
 
+    <ItemGroup>
+        <Compile Include="..\..\TestSupport\FixtureCleanup.cs" Link="TestSupport\FixtureCleanup.cs" />
+    </ItemGroup>
 </Project>

tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Tests/FakeHistorianGatewayClient.cs

@@ -15,10 +15,13 @@ public sealed class FakeHistorianGatewayClient : IHistorianGatewayClient
     public bool TagExistsResult = true;
     public int TagExistsCallCount;
     public Exception? TagExistsThrows;
+    /// <summary>Optional asynchronous catalog response for overlapping presence-cache tests.</summary>
+    public Func<string, CancellationToken, Task<bool>>? TagExistsSource;
     public Task<bool> TagExistsAsync(string tag, CancellationToken ct)
     {
         ct.ThrowIfCancellationRequested();
-        TagExistsCallCount++;
+        Interlocked.Increment(ref TagExistsCallCount);
+        if (TagExistsSource is { } source) return source(tag, ct);
         if (TagExistsThrows is { } error) throw error;
         return Task.FromResult(TagExistsResult);
     }

tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Tests/GatewayAlarmHistorianWriterTests.cs

@@ -18,6 +18,33 @@ public sealed class GatewayAlarmHistorianWriterTests
     private static GatewayAlarmHistorianWriter Writer(FakeHistorianGatewayClient fake) =>
         new(fake, NullLogger<GatewayAlarmHistorianWriter>.Instance);
 
+    [Fact]
+    public async Task A_direct_retry_preserves_the_original_delivery_id()
+    {
+        var evt = Evt("A");
+        var fake = new FakeHistorianGatewayClient { SendEventThrows = new RpcException(new Status(StatusCode.Unavailable, "down")) };
+        using var writer = Writer(fake);
+        Assert.Equal(HistorianWriteOutcome.RetryPlease,
+            (await writer.WriteBatchAsync([evt], TestContext.Current.CancellationToken)).Single());
+        var first = fake.LastSendEvent!.Id;
+        fake.SendEventThrows = null;
+        fake.SendEventResult = new WriteAck { Success = true };
+        Assert.Equal(HistorianWriteOutcome.Ack,
+            (await writer.WriteBatchAsync([evt with { }], TestContext.Current.CancellationToken)).Single());
+        Assert.Equal(evt.DeliveryId.ToString("D"), first);
+        Assert.Equal(first, fake.LastSendEvent!.Id);
+    }
+
+    [Fact]
+    public async Task An_empty_delivery_id_is_permanent_failure_without_an_RPC()
+    {
+        var fake = new FakeHistorianGatewayClient();
+        using var writer = Writer(fake);
+        Assert.Equal(HistorianWriteOutcome.PermanentFail,
+            (await writer.WriteBatchAsync([Evt("A") with { DeliveryId = Guid.Empty }], TestContext.Current.CancellationToken)).Single());
+        Assert.Equal(0, fake.SendEventCallCount);
+    }
+
     [Fact]
     public async Task All_acked_when_SendEvent_succeeds()
     {

tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Tests/GatewayCeilingPagingTests.cs

@@ -197,8 +197,6 @@ public sealed class GatewayCeilingPagingTests : IDisposable
     [InlineData(L - 1, 5000u)]
     [InlineData(L, 5000u)]
     [InlineData(L + 1, 5000u)]
-    [InlineData(Ceiling - 17, 5000u)]
-    [InlineData(Ceiling - 16, 5000u)]
     [InlineData(Ceiling + 1, 5000u)]
     [InlineData(L - 2, 0u)]
     [InlineData(L - 1, 0u)]

tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Tests/GatewayTagPresenceCacheConcurrencyTests.cs

@@ -0,0 +1,209 @@
+using Grpc.Core;
+using Shouldly;
+using Xunit;
+
+namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Tests;
+
+public sealed class GatewayTagPresenceCacheConcurrencyTests(ITestOutputHelper output)
+{
+    private static CancellationToken Ct => TestContext.Current.CancellationToken;
+    private const int BurstSize = 16;
+
+    private sealed class Clock : TimeProvider
+    {
+        public DateTimeOffset Current = new(2026, 10, 13, 8, 0, 0, TimeSpan.Zero);
+        public override DateTimeOffset GetUtcNow() => Current;
+    }
+
+    private static TaskCompletionSource<bool> Gate() => new(TaskCreationOptions.RunContinuationsAsynchronously);
+
+    [Theory]
+    [InlineData(false)]
+    [InlineData(true)]
+    public async Task Overlapping_cold_lookups_return_the_catalog_result_and_populate_the_cache(bool exists)
+    {
+        // Catches dropped/corrupted concurrent writes or a result cached with the wrong presence value.
+        var gate = Gate();
+        var fake = new FakeHistorianGatewayClient { TagExistsSource = (_, _) => gate.Task };
+        var cache = new GatewayTagPresenceCache(fake, new Clock());
+        var reads = Enumerable.Range(0, BurstSize).Select(_ => cache.ExistsAsync("T", Ct)).ToArray();
+        reads.ShouldAllBe(read => !read.IsCompleted);
+        var rpcCount = fake.TagExistsCallCount;
+        // Measure amplification without requiring either coalescing or one RPC per caller.
+        rpcCount.ShouldBeInRange(1, BurstSize);
+        output.WriteLine($"Cold {exists}: {BurstSize} overlapping callers, {rpcCount} catalog RPCs.");
+        gate.SetResult(exists);
+        (await Task.WhenAll(reads).WaitAsync(Ct)).ShouldAllBe(value => value == exists);
+
+        fake.TagExistsSource = null;
+        fake.TagExistsThrows = new InvalidOperationException("A warm lookup must not contact the catalog.");
+        (await cache.ExistsAsync("T", Ct)).ShouldBe(exists);
+    }
+
+    [Theory]
+    [InlineData(false, 5)]
+    [InlineData(true, 60)]
+    public async Task Overlapping_expired_lookups_refresh_presence_at_the_TTL_boundary(bool initial, int ttlSeconds)
+    {
+        // Catches an inclusive expiry comparison or failure to replace an expired positive/negative entry.
+        var clock = new Clock();
+        var fake = new FakeHistorianGatewayClient { TagExistsResult = initial };
+        var cache = new GatewayTagPresenceCache(fake, clock);
+        (await cache.ExistsAsync("T", Ct)).ShouldBe(initial);
+        var expires = clock.Current.AddSeconds(ttlSeconds);
+        clock.Current = expires.AddTicks(-1);
+        fake.TagExistsThrows = new InvalidOperationException("The entry must still be warm before expiry.");
+        (await cache.ExistsAsync("T", Ct)).ShouldBe(initial);
+
+        clock.Current = expires;
+        var gate = Gate();
+        fake.TagExistsThrows = null;
+        fake.TagExistsSource = (_, _) => gate.Task;
+        var reads = Enumerable.Range(0, BurstSize).Select(_ => cache.ExistsAsync("T", Ct)).ToArray();
+        reads.ShouldAllBe(read => !read.IsCompleted);
+        var rpcCount = fake.TagExistsCallCount - 1;
+        rpcCount.ShouldBeInRange(1, BurstSize);
+        output.WriteLine($"Expired {initial}: {BurstSize} overlapping callers, {rpcCount} catalog RPCs.");
+        gate.SetResult(!initial);
+        (await Task.WhenAll(reads).WaitAsync(Ct)).ShouldAllBe(value => value == !initial);
+
+        fake.TagExistsSource = null;
+        fake.TagExistsThrows = new InvalidOperationException("A refreshed lookup must stay cached.");
+        (await cache.ExistsAsync("T", Ct)).ShouldBe(!initial);
+    }
+
+    [Theory]
+    [InlineData(false)]
+    [InlineData(true)]
+    public async Task Overlapping_distinct_tag_completions_keep_the_cache_bounded(bool exists)
+    {
+        // Catches racing capacity checks that allow more than Capacity resident entries.
+        var fake = new FakeHistorianGatewayClient { TagExistsResult = exists };
+        var cache = new GatewayTagPresenceCache(fake, new Clock());
+        var tags = Enumerable.Range(0, GatewayTagPresenceCache.Capacity + 31).Select(i => $"T{i}").ToArray();
+        foreach (var tag in tags.Take(GatewayTagPresenceCache.Capacity - 1))
+            (await cache.ExistsAsync(tag, Ct)).ShouldBe(exists);
+
+        var gate = Gate();
+        fake.TagExistsSource = (_, _) => gate.Task;
+        var reads = tags.Skip(GatewayTagPresenceCache.Capacity - 1).Select(tag => cache.ExistsAsync(tag, Ct)).ToArray();
+        reads.ShouldAllBe(read => !read.IsCompleted);
+        gate.SetResult(exists);
+        (await Task.WhenAll(reads).WaitAsync(Ct)).ShouldAllBe(value => value == exists);
+
+        // Failed probes cannot insert or evict entries, so count resident hits through the public behavior.
+        fake.TagExistsSource = null;
+        var uncached = new InvalidOperationException("Uncached catalog probe.");
+        fake.TagExistsThrows = uncached;
+        var hits = 0;
+        foreach (var tag in tags)
+        {
+            try { (await cache.ExistsAsync(tag, Ct)).ShouldBe(exists); hits++; }
+            catch (InvalidOperationException error) when (ReferenceEquals(error, uncached)) { }
+        }
+        hits.ShouldBe(GatewayTagPresenceCache.Capacity);
+    }
+
+    [Theory]
+    [InlineData(StatusCode.Unavailable)]
+    [InlineData(StatusCode.PermissionDenied)]
+    [InlineData(StatusCode.Internal)]
+    [InlineData(StatusCode.Cancelled)]
+    public async Task A_faulting_lookup_cannot_overwrite_an_overlapping_positive_result(StatusCode code)
+    {
+        // Catches converting metadata faults to false, including a fault finishing after a successful lookup.
+        var success = Gate();
+        var failure = Gate();
+        var calls = 0;
+        var fake = new FakeHistorianGatewayClient
+        {
+            TagExistsSource = (_, _) => Interlocked.Increment(ref calls) == 1 ? success.Task : failure.Task,
+        };
+        var cache = new GatewayTagPresenceCache(fake, new Clock());
+        var goodRead = cache.ExistsAsync("T", Ct);
+        var badRead = cache.ExistsAsync("T", Ct);
+        goodRead.IsCompleted.ShouldBeFalse();
+        badRead.IsCompleted.ShouldBeFalse();
+        success.SetResult(true);
+        (await goodRead.WaitAsync(Ct)).ShouldBeTrue();
+        failure.SetException(new RpcException(new Status(code, "Catalog fault.")));
+        (await Should.ThrowAsync<RpcException>(() => badRead.WaitAsync(Ct))).StatusCode.ShouldBe(code);
+        fake.TagExistsSource = null;
+        fake.TagExistsThrows = new InvalidOperationException("The positive entry must survive the failed lookup.");
+        (await cache.ExistsAsync("T", Ct)).ShouldBeTrue();
+    }
+
+    [Fact]
+    public async Task Cancellation_after_a_catalog_result_cannot_overwrite_an_overlapping_positive_result()
+    {
+        // Catches removal of the post-await token check when the client ignores cancellation and returns false.
+        var success = Gate();
+        var cancelled = Gate();
+        using var caller = CancellationTokenSource.CreateLinkedTokenSource(Ct);
+        var fake = new FakeHistorianGatewayClient
+        {
+            TagExistsSource = (_, token) => token == caller.Token ? cancelled.Task : success.Task,
+        };
+        var cache = new GatewayTagPresenceCache(fake, new Clock());
+        var goodRead = cache.ExistsAsync("T", Ct);
+        var cancelledRead = cache.ExistsAsync("T", caller.Token);
+        success.SetResult(true);
+        (await goodRead.WaitAsync(Ct)).ShouldBeTrue();
+        caller.Cancel();
+        cancelled.SetResult(false);
+        await Should.ThrowAsync<OperationCanceledException>(() => cancelledRead.WaitAsync(Ct));
+        fake.TagExistsSource = null;
+        fake.TagExistsThrows = new InvalidOperationException("The positive entry must survive cancellation.");
+        (await cache.ExistsAsync("T", Ct)).ShouldBeTrue();
+        // Cached results still respect an already-cancelled caller.
+        await Should.ThrowAsync<OperationCanceledException>(() => cache.ExistsAsync("T", caller.Token));
+    }
+
+    [Theory]
+    [InlineData(false, 5)]
+    [InlineData(true, 60)]
+    public async Task An_expired_entry_is_rechecked_after_a_faulting_refresh(bool initial, int ttlSeconds)
+    {
+        // Catches faults renewing an expired negative or replacing an expired positive with a negative.
+        var clock = new Clock();
+        var fake = new FakeHistorianGatewayClient { TagExistsResult = initial };
+        var cache = new GatewayTagPresenceCache(fake, clock);
+        (await cache.ExistsAsync("T", Ct)).ShouldBe(initial);
+        clock.Current = clock.Current.AddSeconds(ttlSeconds);
+        var gate = Gate();
+        fake.TagExistsSource = (_, _) => gate.Task;
+        var read = cache.ExistsAsync("T", Ct);
+        read.IsCompleted.ShouldBeFalse();
+        gate.SetException(new RpcException(new Status(StatusCode.Unavailable, "Catalog unavailable.")));
+        await Should.ThrowAsync<RpcException>(() => read.WaitAsync(Ct));
+        fake.TagExistsSource = null;
+        fake.TagExistsResult = true;
+        (await cache.ExistsAsync("T", Ct)).ShouldBeTrue();
+        fake.TagExistsCallCount.ShouldBe(3);
+    }
+
+    [Theory]
+    [InlineData(false, 5)]
+    [InlineData(true, 60)]
+    public async Task An_expired_entry_is_rechecked_after_a_cancelled_refresh(bool initial, int ttlSeconds)
+    {
+        // Catches cancellation renewing expired entries when the catalog returns an absence result anyway.
+        var clock = new Clock();
+        var fake = new FakeHistorianGatewayClient { TagExistsResult = initial };
+        var cache = new GatewayTagPresenceCache(fake, clock);
+        (await cache.ExistsAsync("T", Ct)).ShouldBe(initial);
+        clock.Current = clock.Current.AddSeconds(ttlSeconds);
+        var gate = Gate();
+        fake.TagExistsSource = (_, _) => gate.Task;
+        using var caller = CancellationTokenSource.CreateLinkedTokenSource(Ct);
+        var read = cache.ExistsAsync("T", caller.Token);
+        read.IsCompleted.ShouldBeFalse();
+        caller.Cancel();
+        gate.SetResult(false);
+        await Should.ThrowAsync<OperationCanceledException>(() => read.WaitAsync(Ct));
+        fake.TagExistsSource = null;
+        fake.TagExistsResult = true;
+        (await cache.ExistsAsync("T", Ct)).ShouldBeTrue();
+        fake.TagExistsCallCount.ShouldBe(3);
+    }
+}

tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Tests/Live/GatewayLiveIntegrationTests.cs

@@ -149,9 +149,8 @@ public sealed class GatewayLiveIntegrationTests(GatewayLiveFixture fixture) : IC
     ///     <see cref="GatewayAlarmHistorianWriter"/>, then <c>ReadEvents</c> the recent window for that
     ///     source through the real <see cref="GatewayHistorianDataSource"/> and assert the event is
     ///     present. Requires the gateway running <c>RuntimeDb:EventReadsEnabled=true</c> (the SQL
-    ///     alarm-history read path). Presence is asserted as "at least one event for the source surfaced
-    ///     in the post-send window" (the data source filters by source); the exact AlarmId / message
-    ///     match is surfaced as a diagnostic, since the SQL event store may re-key the row.
+    ///     alarm-history read path). The exact delivery GUID must return; an unrelated event for the
+    ///     same source cannot satisfy the assertion.
     /// </summary>
     [Fact]
     [Trait("Category", "LiveIntegration")]
@@ -196,7 +195,7 @@ public sealed class GatewayLiveIntegrationTests(GatewayLiveFixture fixture) : IC
             var read = await dataSource.ReadEventsAsync(
                 source, eventUtc - TimeSpan.FromMinutes(5), DateTime.UtcNow + TimeSpan.FromMinutes(1), maxEvents: 0, ct);
             events = read.Events;
-            if (events.Count > 0) break;
+            if (events.Any(e => Guid.TryParse(e.EventId, out var id) && id == alarm.DeliveryId)) break;
             await Task.Delay(TimeSpan.FromSeconds(2), ct);
         }
         while (DateTime.UtcNow < deadline);
@@ -211,18 +210,18 @@ public sealed class GatewayLiveIntegrationTests(GatewayLiveFixture fixture) : IC
             $"a source-filtered ReadEvents must return the just-sent alarm for source '{source}': the gateway now " +
             "populates Source_Object from the event SourceName (C4 fix), so OtOpcUa's own alarm sends round-trip.");
 
-        var exactMatch = events.Any(e => string.Equals(e.EventId, alarmId, StringComparison.Ordinal));
+        var exactMatch = events.Any(e => Guid.TryParse(e.EventId, out var id) && id == alarm.DeliveryId);
+        exactMatch.ShouldBeTrue("the just-sent event must retain its nonzero delivery GUID through persistence and readback.");
         TestContext.Current.SendDiagnosticMessage(
-            $"alarm round-trip: SendEvent source='{source}' id={alarmId} → ReadEvents returned {events.Count} event(s); exact-id match={exactMatch}.");
+            $"alarm round-trip: SendEvent source='{source}' delivery-id={alarm.DeliveryId:D} → exact ID returned among {events.Count} event(s).");
     }
 
     /// <summary>
     ///     Alarm send contract — the OtOpcUa responsibility in isolation: an <see cref="AlarmHistorianEvent"/>
     ///     maps to a wire event the gateway accepts and <c>SendEvent</c> returns <see cref="HistorianWriteOutcome.Ack"/>.
     ///     This is the half that must always hold (the readback half is historian-gated; see
-    ///     <see cref="Alarm_SendEvent_then_ReadEvents"/>). Regression guard for the event-id mapping —
-    ///     a client-supplied wire <c>Id</c> makes the gateway's SendEvent handler throw, so the mapper
-    ///     must leave it unset.
+    ///     <see cref="Alarm_SendEvent_then_ReadEvents"/>). The mapper sends the stable nonzero delivery GUID;
+    ///     the earlier assumption that the gateway rejected every client-supplied ID was incorrect (#705).
     /// </summary>
     [Fact]
     [Trait("Category", "LiveIntegration")]
@@ -251,7 +250,7 @@ public sealed class GatewayLiveIntegrationTests(GatewayLiveFixture fixture) : IC
 
         outcomes.ShouldHaveSingleItem().ShouldBe(
             HistorianWriteOutcome.Ack,
-            "the alarm SendEvent must be acked (the AlarmEventMapper must NOT set the wire event Id — the gateway rejects a client-supplied id).");
+            "the alarm SendEvent carrying its nonzero delivery GUID must be acked.");
     }
 
     /// <summary>

tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Tests/Mapping/AlarmEventMapperTests.cs

@@ -7,6 +7,17 @@ namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Tests.Mapping;
 
 public sealed class AlarmEventMapperTests
 {
+    [Fact]
+    public void A_direct_event_has_a_nonzero_wire_id_that_survives_repeated_mapping()
+    {
+        var alarm = new AlarmHistorianEvent("A", "S", "N", "LimitAlarm", AlarmSeverity.High,
+            "Activated", "m", "system", null, new DateTime(2026, 1, 1, 0, 0, 0, DateTimeKind.Utc));
+        var first = AlarmEventMapper.ToHistorianEvent(alarm).Id;
+        Assert.True(Guid.TryParse(first, out var id), $"Expected a delivery GUID; received '{first}'.");
+        Assert.NotEqual(Guid.Empty, id);
+        Assert.Equal(first, AlarmEventMapper.ToHistorianEvent(alarm).Id);
+    }
+
     [Fact]
     public void Maps_source_time_type_and_rich_properties()
     {
@@ -17,9 +28,7 @@ public sealed class AlarmEventMapperTests
         Assert.Equal("Area/Line/Pump1", e.SourceName);
         Assert.Equal("LimitAlarm", e.Type);
         Assert.Equal(new DateTime(2026, 1, 1, 0, 0, 0, DateTimeKind.Utc), e.EventTime.ToDateTime());
-        // The wire Id is deliberately left unset — the gateway's SendEvent rejects a client-supplied
-        // event id. The alarm's own id is carried as a property instead.
-        Assert.Equal(string.Empty, e.Id);
+        Assert.Equal(a.DeliveryId.ToString("D"), e.Id);
         Assert.Equal("A1", e.Properties["AlarmId"]);
         Assert.Equal("HiHi", e.Properties["AlarmName"]);
         Assert.Equal("Activated", e.Properties["EventKind"]);
@@ -67,14 +76,21 @@ public sealed class AlarmEventMapperTests
     }
 
     [Fact]
-    public void Wire_id_is_never_set_and_blank_alarm_id_adds_no_property()
+    public void A_blank_condition_id_still_has_delivery_identity_and_adds_no_condition_property()
     {
-        // A blank AlarmId must NOT fall back to a generated wire Id (the gateway rejects any id) and
-        // must not add an empty AlarmId property.
+        // Condition identity and event delivery identity serve different purposes.
         var a = new AlarmHistorianEvent("", "S", "N", "DiscreteAlarm", AlarmSeverity.Low, "Cleared", "m", "system", null,
             new DateTime(2026, 1, 1, 0, 0, 0, DateTimeKind.Utc));
         var e = AlarmEventMapper.ToHistorianEvent(a);
-        Assert.Equal(string.Empty, e.Id);
+        Assert.Equal(a.DeliveryId.ToString("D"), e.Id);
         Assert.False(e.Properties.ContainsKey("AlarmId"));
     }
+
+    [Fact]
+    public void An_explicitly_empty_delivery_id_is_rejected_instead_of_reminted()
+    {
+        var alarm = new AlarmHistorianEvent("A", "S", "N", "LimitAlarm", AlarmSeverity.High,
+            "Activated", "m", "system", null, DateTime.UtcNow) { DeliveryId = Guid.Empty };
+        Assert.Throws<ArgumentException>(() => AlarmEventMapper.ToHistorianEvent(alarm));
+    }
 }

tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.MTConnect.Tests/MTConnectDriverLifecycleTests.cs

@@ -1,4 +1,5 @@
 using Shouldly;
+using ZB.MOM.WW.OtOpcUa.Tests;
 using Xunit;
 using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
 
@@ -16,7 +17,7 @@ namespace ZB.MOM.WW.OtOpcUa.Driver.MTConnect.Tests;
 ///     keeps talking to the old Agent, and a cache flush that wedges browse. Each of those is a
 ///     defect the happy-path assertions alone would pass straight over.
 /// </remarks>
-public sealed class MTConnectDriverLifecycleTests
+public sealed class MTConnectDriverLifecycleTests(ITestOutputHelper output)
 {
     private const string AgentUri = "http://fixture-agent:5000";
 
@@ -598,18 +599,79 @@ public sealed class MTConnectDriverLifecycleTests
     // same "browse must not be left permanently empty" obligation.
 
     [Fact]
-    public void Dropping_the_probe_cache_keeps_the_observation_index()
+    public async Task Dropping_the_probe_cache_keeps_the_observation_index()
     {
-        var (driver, _) = InitializedDriverAsync().GetAwaiter().GetResult();
+        var (driver, _) = NewDriver();
+        await FixtureCleanup.RunAsync(async () =>
+        {
+            await driver.InitializeAsync("{}", CancellationToken.None);
 
-        driver.InvalidateProbeModel();
+            driver.InvalidateProbeModel();
 
-        driver.CachedProbeModel.ShouldBeNull();
+            driver.CachedProbeModel.ShouldBeNull();
 
-        // The index is required for correctness (it IS the read surface) — never flushable.
-        driver.ObservationIndex.Count.ShouldBe(FixtureDataItemCount);
-        driver.ObservationIndex.Get("dev1_pos").StatusCode.ShouldBe(0u);
-        driver.GetHealth().State.ShouldBe(DriverState.Healthy);
+            // The index is required for correctness (it IS the read surface) — never flushable.
+            driver.ObservationIndex.Count.ShouldBe(FixtureDataItemCount);
+            driver.ObservationIndex.Get("dev1_pos").StatusCode.ShouldBe(0u);
+            driver.GetHealth().State.ShouldBe(DriverState.Healthy);
+        }, () => driver.ShutdownAsync(CancellationToken.None),
+            failure => output.WriteLine($"MTConnect cache drop cleanup failed: {failure}"));
+    }
+
+    [Fact]
+    public async Task Body_failure_after_probe_scheduler_enters_stops_probe_and_disposes_client()
+    {
+        var client = CannedAgentClient.FromFixtures();
+        var entered = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
+        var released = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
+        var cancelled = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
+        var driver = new MTConnectDriver(Opts(), "mt1", _ => client,
+            probeScheduler: async (_, ct) =>
+            {
+                entered.TrySetResult();
+                try { await released.Task.WaitAsync(ct); }
+                catch (OperationCanceledException) when (ct.IsCancellationRequested)
+                {
+                    cancelled.TrySetResult();
+                    throw;
+                }
+            });
+        Task? probe = null;
+        var sentinel = new InvalidOperationException("body sentinel");
+        try
+        {
+            var actual = await Should.ThrowAsync<InvalidOperationException>(() => FixtureCleanup.RunAsync(async () =>
+            {
+                await driver.InitializeAsync("{}", CancellationToken.None);
+                await entered.Task.WaitAsync(TimeSpan.FromSeconds(5), TestContext.Current.CancellationToken);
+                probe = driver.ProbeLoopTask;
+                _ = probe.ShouldNotBeNull();
+                probe!.IsCompleted.ShouldBeFalse();
+                client.IsDisposed.ShouldBeFalse();
+                throw sentinel;
+            }, () => driver.ShutdownAsync(CancellationToken.None),
+                failure => output.WriteLine($"MTConnect probe fixture cleanup failed: {failure}")));
+            actual.ShouldBeSameAs(sentinel);
+            await cancelled.Task.WaitAsync(TimeSpan.FromSeconds(5), TestContext.Current.CancellationToken);
+            probe!.IsCompleted.ShouldBeTrue();
+            client.DisposeCount.ShouldBe(1);
+        }
+        finally { released.TrySetResult(); }
+    }
+
+    [Fact]
+    public async Task Initialization_failure_disposes_the_client_inside_fixture_boundary()
+    {
+        var client = CannedAgentClient.FromFixtures();
+        var failure = new HttpRequestException("current unavailable");
+        client.CurrentFailure = failure;
+        var driver = new MTConnectDriver(Opts(), "mt1", _ => client);
+        var actual = await Should.ThrowAsync<HttpRequestException>(() => FixtureCleanup.RunAsync(
+            () => driver.InitializeAsync("{}", CancellationToken.None),
+            () => driver.ShutdownAsync(CancellationToken.None),
+            secondary => output.WriteLine($"MTConnect init fixture cleanup failed: {secondary}")));
+        actual.ShouldBeSameAs(failure);
+        client.DisposeCount.ShouldBe(1);
     }
 
     [Fact]

tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.MTConnect.Tests/ZB.MOM.WW.OtOpcUa.Driver.MTConnect.Tests.csproj

@@ -39,4 +39,7 @@
         </Content>
     </ItemGroup>
 
+    <ItemGroup>
+        <Compile Include="..\..\TestSupport\FixtureCleanup.cs" Link="TestSupport\FixtureCleanup.cs" />
+    </ItemGroup>
 </Project>

tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Modbus.Tests/ModbusLifecycleHygieneTests.cs

@@ -1,6 +1,7 @@
 using System.Collections.Concurrent;
 using System.Reflection;
 using Shouldly;
+using ZB.MOM.WW.OtOpcUa.Tests;
 using Xunit;
 using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
 using ZB.MOM.WW.OtOpcUa.Driver.Modbus;
@@ -14,7 +15,7 @@ namespace ZB.MOM.WW.OtOpcUa.Driver.Modbus.Tests;
 ///     unit test alongside them per Driver.Modbus-012.
 /// </summary>
 [Trait("Category", "Unit")]
-public sealed class ModbusLifecycleHygieneTests
+public sealed class ModbusLifecycleHygieneTests(ITestOutputHelper output)
 {
     private sealed class FakeTransport : IModbusTransport
     {
@@ -528,7 +529,7 @@ public sealed class ModbusLifecycleHygieneTests
 
     /// <summary>Verifies that DecodeBitArray rejects an empty bitmap with InvalidDataException.</summary>
     [Fact]
-    public void DecodeBitArray_rejects_an_empty_bitmap_with_InvalidDataException()
+    public async Task DecodeBitArray_rejects_an_empty_bitmap_with_InvalidDataException()
     {
         var decode = typeof(ModbusDriver).GetMethod(
             "DecodeBitArray", BindingFlags.NonPublic | BindingFlags.Static)!;
@@ -542,11 +543,49 @@ public sealed class ModbusLifecycleHygieneTests
             RawTags = ModbusRawTags.Entries([new ModbusTagDefinition("Coil", ModbusRegion.Coils, 0, ModbusDataType.Bool)]),
         };
         var drv = new ModbusDriver(opts, "modbus-1", _ => fake);
-        drv.InitializeAsync("{}", CancellationToken.None).GetAwaiter().GetResult();
+        await FixtureCleanup.RunAsync(async () =>
+        {
+            await drv.InitializeAsync("{}", CancellationToken.None);
 
-        var r = drv.ReadAsync(["Coil"], CancellationToken.None).GetAwaiter().GetResult();
-        // The empty-bitmap guard surfaces via the BadCommunicationError catch-all.
-        r[0].StatusCode.ShouldBe(0x80050000u);
+            var r = await drv.ReadAsync(["Coil"], CancellationToken.None);
+            // The empty-bitmap guard surfaces via the BadCommunicationError catch-all.
+            r[0].StatusCode.ShouldBe(0x80050000u);
+        }, () => drv.ShutdownAsync(CancellationToken.None),
+            failure => output.WriteLine($"Modbus empty bitmap cleanup failed: {failure}"));
+    }
+
+    [Fact]
+    public async Task Body_failure_after_probe_request_cancels_it_and_disposes_transport()
+    {
+        var fake = new EmptyBitTransport();
+        var opts = new ModbusDriverOptions
+        {
+            Host = "fake",
+            RawTags = ModbusRawTags.Entries([new ModbusTagDefinition("Coil", ModbusRegion.Coils, 0, ModbusDataType.Bool)]),
+        };
+        var drv = new ModbusDriver(opts, "modbus-1", _ => fake);
+        var sentinel = new InvalidOperationException("body sentinel");
+        try
+        {
+            var actual = await Should.ThrowAsync<InvalidOperationException>(() => FixtureCleanup.RunAsync(async () =>
+            {
+                await drv.InitializeAsync("{}", CancellationToken.None);
+                var result = await drv.ReadAsync(["Coil"], CancellationToken.None);
+                result[0].StatusCode.ShouldBe(0x80050000u);
+                fake.CoilRequestCount.ShouldBe(1);
+                await fake.ProbeEntered.Task.WaitAsync(TimeSpan.FromSeconds(5), TestContext.Current.CancellationToken);
+                fake.ProbeToken.IsCancellationRequested.ShouldBeFalse();
+                fake.DisposeCount.ShouldBe(0);
+                throw sentinel;
+            }, () => drv.ShutdownAsync(CancellationToken.None),
+                failure => output.WriteLine($"Modbus probe fixture cleanup failed: {failure}")));
+            actual.ShouldBeSameAs(sentinel);
+            fake.ProbeToken.IsCancellationRequested.ShouldBeTrue();
+            await fake.ProbeCancelled.Task.WaitAsync(TimeSpan.FromSeconds(5), TestContext.Current.CancellationToken);
+            await fake.ProbeRequestCompleted.Task.WaitAsync(TimeSpan.FromSeconds(5), TestContext.Current.CancellationToken);
+            fake.DisposeCount.ShouldBe(1);
+        }
+        finally { fake.ReleaseProbe.TrySetResult(); }
     }
 
     /// <summary>
@@ -556,6 +595,14 @@ public sealed class ModbusLifecycleHygieneTests
     /// </summary>
     private sealed class EmptyBitTransport : IModbusTransport
     {
+        public TaskCompletionSource ProbeEntered { get; } = new(TaskCreationOptions.RunContinuationsAsynchronously);
+        public TaskCompletionSource ProbeCancelled { get; } = new(TaskCreationOptions.RunContinuationsAsynchronously);
+        public TaskCompletionSource ProbeRequestCompleted { get; } = new(TaskCreationOptions.RunContinuationsAsynchronously);
+        public TaskCompletionSource ReleaseProbe { get; } = new(TaskCreationOptions.RunContinuationsAsynchronously);
+        public CancellationToken ProbeToken { get; private set; }
+        public int DisposeCount { get; private set; }
+        public int CoilRequestCount { get; private set; }
+
         /// <summary>Returns a completed task without performing any connection.</summary>
         /// <param name="ct">The cancellation token for the operation.</param>
         public Task ConnectAsync(CancellationToken ct) => Task.CompletedTask;
@@ -563,10 +610,29 @@ public sealed class ModbusLifecycleHygieneTests
         /// <param name="unitId">The Modbus unit ID.</param>
         /// <param name="pdu">The protocol data unit being sent.</param>
         /// <param name="ct">The cancellation token for the operation.</param>
-        public Task<byte[]> SendAsync(byte unitId, byte[] pdu, CancellationToken ct)
-            => Task.FromResult(new byte[] { pdu[0], 0 });
-        /// <summary>Completes the disposal without doing any work.</summary>
-        public ValueTask DisposeAsync() => ValueTask.CompletedTask;
+        public async Task<byte[]> SendAsync(byte unitId, byte[] pdu, CancellationToken ct)
+        {
+            if (pdu[0] == 0x03)
+            {
+                ProbeToken = ct;
+                ProbeEntered.TrySetResult();
+                try { await ReleaseProbe.Task.WaitAsync(ct); }
+                catch (OperationCanceledException) when (ct.IsCancellationRequested)
+                {
+                    ProbeCancelled.TrySetResult();
+                    throw;
+                }
+                finally { ProbeRequestCompleted.TrySetResult(); }
+            }
+            if (pdu[0] == 0x01) CoilRequestCount++;
+            return new byte[] { pdu[0], 0 };
+        }
+        /// <summary>Records transport disposal.</summary>
+        public ValueTask DisposeAsync()
+        {
+            DisposeCount++;
+            return ValueTask.CompletedTask;
+        }
     }
 
     // -------------------- Finding -003 (volatile _health) --------------------

tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Modbus.Tests/ZB.MOM.WW.OtOpcUa.Driver.Modbus.Tests.csproj

@@ -29,4 +29,7 @@
         <ProjectReference Include="..\..\..\src\Drivers\ZB.MOM.WW.OtOpcUa.Driver.Modbus\ZB.MOM.WW.OtOpcUa.Driver.Modbus.csproj"/>
     </ItemGroup>
 
+    <ItemGroup>
+        <Compile Include="..\..\TestSupport\FixtureCleanup.cs" Link="TestSupport\FixtureCleanup.cs" />
+    </ItemGroup>
 </Project>

tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.S7.IntegrationTests/S7_1500/S7_1500SpanReadTests.cs

@@ -27,7 +27,15 @@ namespace ZB.MOM.WW.OtOpcUa.Driver.S7.IntegrationTests.S7_1500;
 ///         areas: <b>PASSED</b> — 4 passed, 0 failed, 1 skipped (the device number-addressing test,
 ///         see below). The Timer/Counter batch was served by 4 requests for 7 tags: one DB1 span,
 ///         one 3-item multi-var carrying only the scalars, and one 2-byte area read apiece for
-///         <c>T0</c> (decoded 456.0 s from S5TIME <c>0x2456</c>) and <c>C0</c> (decoded count 4711).
+///         <c>T0</c> (decoded 456.0 s from S5TIME <c>0x2456</c>) and <c>C0</c> (surfaced raw word 4711).
+///     </para>
+///     <para>
+///         <b>Counter evidence limit.</b> The profile's C0=4711 and C7=31337 are synthetic raw
+///         words, outside classic SIMATIC counts 0..999. The device-0 record proves the fixture's
+///         raw-word decode and batching exclusion, not classic count semantics or nonzero
+///         addressing. BuildOptions retains S71500/rack 0/slot 0; a target-specific semantic gate
+///         still needs independent PLC counts and binary/BCD references paired with captured
+///         C-area bytes, CPU model/firmware and connection settings (docs/drivers/S7.md, #616).
 ///     </para>
 ///     <para>
 ///         <b>What the layout buys.</b> The ten tags are positioned so the planner's rules produce a
@@ -73,11 +81,12 @@ namespace ZB.MOM.WW.OtOpcUa.Driver.S7.IntegrationTests.S7_1500;
 ///         returns TM bytes 0..1, and it takes <c>start 80</c> (= 10 × 8) to reach the timer seeded
 ///         at TM byte 10. Device <b>0</b> is the single fixed point of the two conventions
 ///         (<c>0 &gt;&gt; 3 == 2 × 0 == 0</c>), so it is the only device this fixture can pin a
-///         VALUE for, and that is what the mixed batch uses. The spec-correct <c>2 × N</c> cells
+///         VALUE for, and that is what the mixed batch uses. The profile's <c>2 × N</c> cells
 ///         are seeded too and
 ///         <see cref="Timer_and_counter_decode_the_seeded_values_at_their_device_numbers"/> asserts
 ///         them — skipped unless <c>S7_TIMER_NUMBER_ADDRESSING=1</c> declares the endpoint a real
-///         CPU. The device-number offset arithmetic itself stays unit-covered
+///         CPU. This flag is an operator declaration, not hardware evidence.
+///         The device-number offset arithmetic itself stays unit-covered
 ///         (<c>S7ScalarBlockTests</c>, <c>S7SpanReadParityTests</c>).
 ///     </para>
 ///     <para>
@@ -240,7 +249,7 @@ public sealed class S7_1500SpanReadTests(Snap7ServerFixture sim)
             snapshot.StatusCode.ShouldBe(0u, $"'{name}' must be Good");
 
         // (a) Correct decode off real wire. A Timer surfaces as the S5TIME word's duration in
-        // seconds (double) and a Counter as the raw count (int) — two different areas, two
+        // seconds (double) and a Counter as the raw word (int) — two different areas, two
         // different values, two different CLR types, so a read that reached the wrong area or
         // decoded through the wrong arm cannot pass.
         byName["TC_Timer"].Value.ShouldBe(TimerCounterLayout.Device0TimerSeconds);
@@ -280,15 +289,19 @@ public sealed class S7_1500SpanReadTests(Snap7ServerFixture sim)
 
     /// <summary>
     ///     Verifies a Timer and a Counter at NON-zero device numbers decode their seeded values —
-    ///     the assertion the shared fixture cannot serve.
+    ///     the raw-word assertion the shared fixture cannot serve.
     /// </summary>
     /// <remarks>
     ///     Skipped unless <c>S7_TIMER_NUMBER_ADDRESSING=1</c> declares the endpoint a device
     ///     number-addressing CPU, because python-snap7's server is not one: it resolves the TM/CT
     ///     request address as a bit address, so <c>T5</c> reads TM byte 0 rather than TM byte 10
-    ///     (measured 2026-08-17 — see the type remarks). Set the variable alongside
-    ///     <c>S7_SIM_ENDPOINT</c> when pointing this suite at a real S7 CPU seeded to match
-    ///     <c>Docker/profiles/s7_1500.json</c>'s TM/CT areas.
+    ///     (measured 2026-08-17 — see the type remarks). The flag only enables this test;
+    ///     BuildOptions still selects S71500, rack 0, slot 0 and the profile's raw C7 word 31337.
+    ///     It does not configure a classic CPU or prove classic 0..999 count semantics. A separate
+    ///     target-specific gate needs valid counts, independent PLC binary/BCD references and
+    ///     captured C-area bytes, plus CPU model/firmware and connection settings (#616).
+    ///     See docs/drivers/S7.md for the evidence checklist. The skip message below retains the
+    ///     historical fixture instructions; enabling the flag alone does not satisfy this gate.
     /// </remarks>
     [Fact]
     public async Task Timer_and_counter_decode_the_seeded_values_at_their_device_numbers()
@@ -408,12 +421,13 @@ public sealed class S7_1500SpanReadTests(Snap7ServerFixture sim)
     /// <summary>
     ///     The Timer/Counter layouts. Two sets: a mixed batch that provokes all three read shapes at
     ///     once (block span, multi-var, per-tag Timer/Counter), and a device-numbered pair only a
-    ///     real CPU can serve. Addresses map 1:1 to the TM0 / CT0 seeds in
+    ///     device number-addressing endpoint can serve. These synthetic raw-word expectations
+    ///     do not prove classic counter semantics. Addresses map 1:1 to the TM0 / CT0 seeds in
     ///     <c>Docker/profiles/s7_1500.json</c>.
     /// </summary>
     private static class TimerCounterLayout
     {
-        /// <summary>Declares the endpoint a real S7 CPU, which addresses TM/CT by device number.</summary>
+        /// <summary>Operator opt-in for nonzero TM/CT addressing; does not change CPU options or prove encoding.</summary>
         public const string RealCpuEnvVar = "S7_TIMER_NUMBER_ADDRESSING";
 
         /// <summary>
@@ -425,7 +439,7 @@ public sealed class S7_1500SpanReadTests(Snap7ServerFixture sim)
         /// <summary>T0's seed 0x2456 — S5TIME base 2 (×1 s) over BCD 456.</summary>
         public const double Device0TimerSeconds = 456.0d;
 
-        /// <summary>C0's seed — the raw big-endian count word.</summary>
+        /// <summary>C0's synthetic raw big-endian word 4711; not a classic 0..999 count reference.</summary>
         public const int Device0CounterCount = 4711;
 
         /// <summary>
@@ -441,7 +455,7 @@ public sealed class S7_1500SpanReadTests(Snap7ServerFixture sim)
             new("TC_Lone_I16",   "DB1.DBW600", S7DataType.Int16),
             new("TC_Lone_Mk",    "MW10",       S7DataType.UInt16),
             // Timer tags are typed Float64 (decoded to seconds) and Counter tags Int32 (decoded to
-            // a count) — the init guard rejects any other typing. Writable:false is mandatory, not
+            // a raw word) — the init guard rejects any other typing. Writable:false is mandatory, not
             // stylistic: S7TagDefinition defaults Writable to TRUE and the same guard rejects a
             // writable Timer/Counter outright (they are read-only this phase).
             new("TC_Timer",      "T0",         S7DataType.Float64, Writable: false),

tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.TwinCAT.Tests/TwinCATReinitializeReleasesStateTests.cs

@@ -1,4 +1,5 @@
 using Shouldly;
+using ZB.MOM.WW.OtOpcUa.Tests;
 using Xunit;
 
 namespace ZB.MOM.WW.OtOpcUa.Driver.TwinCAT.Tests;
@@ -8,7 +9,7 @@ namespace ZB.MOM.WW.OtOpcUa.Driver.TwinCAT.Tests;
 ///     orphaned the previous state's probe loop + client. It now releases them first.
 /// </summary>
 [Trait("Category", "Unit")]
-public sealed class TwinCATReinitializeReleasesStateTests
+public sealed class TwinCATReinitializeReleasesStateTests(ITestOutputHelper output)
 {
     /// <summary>Initialize twice: the first state's probe loop has finished and its client is disposed.</summary>
     [Fact]
@@ -21,28 +22,88 @@ public sealed class TwinCATReinitializeReleasesStateTests
             Devices = [new TwinCATDeviceOptions(host)],
             Probe = new TwinCATProbeOptions { Enabled = true, Interval = TimeSpan.FromMilliseconds(20), Timeout = TimeSpan.FromSeconds(1) },
         }, "drv-reinit", factory);
-        await drv.InitializeAsync("{}", CancellationToken.None);
+        await FixtureCleanup.RunAsync(async () =>
+        {
+            await drv.InitializeAsync("{}", CancellationToken.None);
 
-        var first = drv.GetDeviceState(host)!;
-        // Positive control: the first probe loop is running and holds a connected client.
+            var first = drv.GetDeviceState(host)!;
+            // Positive control: the first probe loop is running and holds a connected client.
+            var deadline = DateTime.UtcNow.AddSeconds(5);
+            while (!(factory.Clients.Count > 0 && factory.Clients[0].IsConnected))
+            {
+                if (DateTime.UtcNow > deadline) throw new TimeoutException("probe never connected");
+                await Task.Delay(10);
+            }
+            var firstClient = factory.Clients[0];
+            var firstLoop = first.ProbeTask;
+            _ = firstLoop.ShouldNotBeNull();
+            firstLoop!.IsCompleted.ShouldBeFalse();
+
+            await drv.InitializeAsync("{}", CancellationToken.None);
+
+            firstLoop.IsCompleted.ShouldBeTrue(); // joined
+            firstClient.DisposeCount.ShouldBeGreaterThanOrEqualTo(1);
+            firstClient.IsConnected.ShouldBeFalse();
+            drv.GetDeviceState(host).ShouldNotBeSameAs(first);
+        }, () => drv.ShutdownAsync(CancellationToken.None),
+            failure => output.WriteLine($"TwinCAT SecondInitialize cleanup failed: {failure}"));
+    }
+
+    /// <summary>Failure after a live first or replacement state must release both owned resources.</summary>
+    [Theory]
+    [InlineData(false)]
+    [InlineData(true)]
+    public async Task Body_failure_releases_live_probe_resources(bool afterReplacement)
+    {
+        const string host = "ads://5.23.91.23.1.1:851";
+        var factory = new FakeTwinCATClientFactory();
+        var drv = new TwinCATDriver(new TwinCATDriverOptions
+        {
+            Devices = [new TwinCATDeviceOptions(host)],
+            Probe = new TwinCATProbeOptions { Enabled = true, Interval = TimeSpan.FromMilliseconds(20), Timeout = TimeSpan.FromSeconds(1) },
+        }, "drv-reinit", factory);
+        Task? firstLoop = null, secondLoop = null;
+        FakeTwinCATClient? firstClient = null, secondClient = null;
+        var sentinel = new InvalidOperationException("body sentinel");
+        var actual = await Should.ThrowAsync<InvalidOperationException>(() => FixtureCleanup.RunAsync(async () =>
+        {
+            await drv.InitializeAsync("{}", CancellationToken.None);
+            await WaitUntilAsync(() => factory.Clients.Count > 0 && factory.Clients[0].IsConnected);
+            firstClient = factory.Clients[0];
+            firstLoop = drv.GetDeviceState(host)!.ProbeTask;
+            _ = firstLoop.ShouldNotBeNull();
+            firstLoop!.IsCompleted.ShouldBeFalse();
+            if (afterReplacement)
+            {
+                await drv.InitializeAsync("{}", CancellationToken.None);
+                await WaitUntilAsync(() => factory.Clients.Count > 1 && factory.Clients[1].IsConnected);
+                secondClient = factory.Clients[1];
+                secondLoop = drv.GetDeviceState(host)!.ProbeTask;
+                _ = secondLoop.ShouldNotBeNull();
+                secondLoop!.IsCompleted.ShouldBeFalse();
+            }
+            throw sentinel;
+        }, () => drv.ShutdownAsync(CancellationToken.None),
+            failure => output.WriteLine($"TwinCAT failure fixture cleanup failed: {failure}")));
+        actual.ShouldBeSameAs(sentinel);
+        firstLoop!.IsCompleted.ShouldBeTrue();
+        firstClient!.DisposeCount.ShouldBeGreaterThanOrEqualTo(1);
+        firstClient.IsConnected.ShouldBeFalse();
+        if (afterReplacement)
+        {
+            secondLoop!.IsCompleted.ShouldBeTrue();
+            secondClient!.DisposeCount.ShouldBeGreaterThanOrEqualTo(1);
+            secondClient.IsConnected.ShouldBeFalse();
+        }
+    }
+
+    private static async Task WaitUntilAsync(Func<bool> condition)
+    {
         var deadline = DateTime.UtcNow.AddSeconds(5);
-        while (!(factory.Clients.Count > 0 && factory.Clients[0].IsConnected))
+        while (!condition())
         {
             if (DateTime.UtcNow > deadline) throw new TimeoutException("probe never connected");
             await Task.Delay(10);
         }
-        var firstClient = factory.Clients[0];
-        var firstLoop = first.ProbeTask;
-        firstLoop.ShouldNotBeNull();
-        firstLoop!.IsCompleted.ShouldBeFalse();
-
-        await drv.InitializeAsync("{}", CancellationToken.None);
-
-        firstLoop.IsCompleted.ShouldBeTrue(); // joined
-        firstClient.DisposeCount.ShouldBeGreaterThanOrEqualTo(1);
-        firstClient.IsConnected.ShouldBeFalse();
-        drv.GetDeviceState(host).ShouldNotBeSameAs(first);
-
-        await drv.ShutdownAsync(CancellationToken.None);
     }
 }

tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.TwinCAT.Tests/ZB.MOM.WW.OtOpcUa.Driver.TwinCAT.Tests.csproj

@@ -29,4 +29,7 @@
         <ProjectReference Include="..\..\..\src\Drivers\ZB.MOM.WW.OtOpcUa.Driver.TwinCAT\ZB.MOM.WW.OtOpcUa.Driver.TwinCAT.csproj"/>
     </ItemGroup>
 
+    <ItemGroup>
+        <Compile Include="..\..\TestSupport\FixtureCleanup.cs" Link="TestSupport\FixtureCleanup.cs" />
+    </ItemGroup>
 </Project>

tests/Server/ZB.MOM.WW.OtOpcUa.ControlPlane.Tests/Communication/CentralCommunicationActorTests.cs

@@ -1,6 +1,9 @@
 using System.Diagnostics.CodeAnalysis;
 using Akka.Actor;
 using Akka.Cluster.Tools.PublishSubscribe;
+using Akka.Event;
+using Akka.TestKit;
+using Akka.TestKit.Xunit2;
 using Microsoft.EntityFrameworkCore;
 using Shouldly;
 using Xunit;
@@ -680,7 +683,7 @@ public class CentralCommunicationActorTests : ControlPlaneActorTestBase
 
     /// <summary>Counts contact loads and fails the first <c>failFirst</c> of them.</summary>
     private sealed class CountingDbFactory(
-        IDbContextFactory<OtOpcUaConfigDbContext> inner, int failFirst = 0)
+        IDbContextFactory<OtOpcUaConfigDbContext> inner, int failFirst = 0, Action<int>? onLoad = null)
         : IDbContextFactory<OtOpcUaConfigDbContext>
     {
         private int _calls;
@@ -689,7 +692,9 @@ public class CentralCommunicationActorTests : ControlPlaneActorTestBase
 
         public OtOpcUaConfigDbContext CreateDbContext()
         {
-            if (Interlocked.Increment(ref _calls) <= failFirst)
+            var call = Interlocked.Increment(ref _calls);
+            onLoad?.Invoke(call);
+            if (call <= failFirst)
                 throw new InvalidOperationException("ConfigDb unreachable at boot");
             return inner.CreateDbContext();
         }
@@ -747,31 +752,51 @@ public class CentralCommunicationActorTests : ControlPlaneActorTestBase
     [Fact]
     public void Bootstrap_retries_stop_once_the_gate_is_published()
     {
-        // The provider throws on PreStart (call 1) and on the first load's resolve (call 2), which
-        // schedules a 1 s retry. A manual refresh then publishes while that retry is still PENDING;
-        // stop-on-publish must cancel it. Without the cancel the timer fires, loads again and
-        // republishes (loads == 3, publications == 2).
+        // Freeze time only for this fact: the cluster harness and every other retry test retain
+        // their real scheduler. gRPC mode needs neither cluster membership nor pub-sub here.
+        using var local = new TestKit(@"akka {
+            loglevel = INFO
+            scheduler.implementation = ""Akka.TestKit.TestScheduler, Akka.TestKit""
+        }");
+        var scheduler = (TestScheduler)local.Sys.Scheduler;
+        var loads = local.CreateTestProbe();
+        var barrier = local.CreateTestProbe();
+        local.Sys.EventStream.Subscribe(barrier.Ref, typeof(UnhandledMessage));
         var inner = NewInMemoryDbFactory();
         SeedGrpcCluster(inner, "C1", ("host-a:4053", "host-a", 4056, true, false));
-        var factory = new CountingDbFactory(inner);
+        var factory = new CountingDbFactory(inner, onLoad: n => loads.Ref.Tell(n));
         var psk = new RecordingPskProvider();
         var calls = 0;
 
+        // PreStart and the first contact load fail resolution. The latter registers a retry.
         IMeshPskProvider? PskFactory() =>
             Interlocked.Increment(ref calls) <= 2 ? throw new InvalidOperationException("race") : psk;
 
-        var actor = Sys.ActorOf(CentralCommunicationActor.Props(
+        var actor = local.Sys.ActorOf(CentralCommunicationActor.Props(
             factory, GrpcMode(), () => null, () => new RecordingDialer(), PskFactory,
             bootstrapRetryInterval: TimeSpan.FromSeconds(1)));
+        // A mailbox barrier also waits for PreStart to register the zero-delay periodic tick.
+        var marker = new object();
+        actor.Tell(marker);
+        barrier.ExpectMsg<UnhandledMessage>(TimeSpan.FromSeconds(5)).Message.ShouldBeSameAs(marker);
+        scheduler.Advance(TimeSpan.Zero);
+        loads.ExpectMsg(1, TimeSpan.FromSeconds(5));
+        local.AwaitCondition(() => Volatile.Read(ref calls) == 2, TimeSpan.FromSeconds(5));
 
-        // Positive control: the failed first resolve happened, so a retry is pending.
-        AwaitCondition(() => Volatile.Read(ref calls) >= 2, TimeSpan.FromSeconds(5));
-        Thread.Sleep(100);
-        actor.Tell(new CentralCommunicationActor.RefreshContacts());
-        AwaitCondition(() => psk.KnownClusterPublications.Count >= 1, TimeSpan.FromSeconds(5));
+        // This mailbox marker is handled after the failed ContactsLoaded handler returns.
+        // Unlike observing the provider call alone, it proves retry registration is complete.
+        actor.Tell(marker);
+        barrier.ExpectMsg<UnhandledMessage>(TimeSpan.FromSeconds(5)).Message.ShouldBeSameAs(marker);
 
-        Thread.Sleep(2500);
+        local.EventFilter.Info(contains: "Known-cluster gate published").ExpectOne(
+            TimeSpan.FromSeconds(5), () => actor.Tell(new CentralCommunicationActor.RefreshContacts()));
+        loads.ExpectMsg(2, TimeSpan.FromSeconds(5));
 
+        // Cross the pending retry deadline, leaving the 60-second periodic tick untouched.
+        // LoadContactsFromDb uses Task.Run, so observe its boundary for the original absence
+        // budget rather than asserting before a mistakenly retained retry can start its read.
+        scheduler.Advance(TimeSpan.FromMilliseconds(2500));
+        loads.ExpectNoMsg(TimeSpan.FromMilliseconds(2500));
         factory.Calls.ShouldBe(2);
         psk.KnownClusterPublications.Count.ShouldBe(1);
     }

tests/Server/ZB.MOM.WW.OtOpcUa.Host.IntegrationTests/LocalDb/AlarmSfConvergenceTests.cs

@@ -41,10 +41,14 @@ public sealed class AlarmSfConvergenceTests
     /// <summary>A writer that never acknowledges — a historian outage, which is when the buffer matters.</summary>
     private sealed class OutageWriter : IAlarmHistorianWriter
     {
+        public List<AlarmHistorianEvent> Events { get; } = [];
         public Task<IReadOnlyList<HistorianWriteOutcome>> WriteBatchAsync(
-            IReadOnlyList<AlarmHistorianEvent> batch, CancellationToken cancellationToken) =>
-            Task.FromResult<IReadOnlyList<HistorianWriteOutcome>>(
+            IReadOnlyList<AlarmHistorianEvent> batch, CancellationToken cancellationToken)
+        {
+            Events.AddRange(batch);
+            return Task.FromResult<IReadOnlyList<HistorianWriteOutcome>>(
                 batch.Select(_ => HistorianWriteOutcome.RetryPlease).ToList());
+        }
     }
 
     private static AlarmHistorianEvent Event(string alarmId, int second) => new(
@@ -170,11 +174,15 @@ public sealed class AlarmSfConvergenceTests
         await pair.StartAsync();
 
         var shared = Event("eq/shared", 1);
-        using var sinkA = new LocalDbStoreAndForwardSink(pair.A, new OutageWriter(), Log);
-        using var sinkB = new LocalDbStoreAndForwardSink(pair.B, new OutageWriter(), Log);
+        var sharedOnB = Event("eq/shared", 1);
+        shared.DeliveryId.ShouldNotBe(sharedOnB.DeliveryId);
+        var writerA = new OutageWriter();
+        var writerB = new OutageWriter();
+        using var sinkA = new LocalDbStoreAndForwardSink(pair.A, writerA, Log);
+        using var sinkB = new LocalDbStoreAndForwardSink(pair.B, writerB, Log);
 
         await sinkA.EnqueueAsync(shared, TestContext.Current.CancellationToken);
-        await sinkB.EnqueueAsync(shared, TestContext.Current.CancellationToken);
+        await sinkB.EnqueueAsync(sharedOnB, TestContext.Current.CancellationToken);
 
         // A second, DISTINCT event on B. Without it the scenario would pass with replication
         // switched off entirely — each node would hold its own single copy and "count == 1 on
@@ -193,5 +201,13 @@ public sealed class AlarmSfConvergenceTests
                 async () => await CountAsync(pair.A) > 2 || await CountAsync(pair.B) > 2,
                 TimeSpan.FromSeconds(3)))
             .ShouldBeFalse("the duplicate must stay collapsed");
+
+        // Either node can retry after promotion, with the same GUID despite independent input identities.
+        await sinkA.DrainOnceAsync(TestContext.Current.CancellationToken);
+        await sinkB.DrainOnceAsync(TestContext.Current.CancellationToken);
+        var deliveryA = writerA.Events.Single(e => e.AlarmId == "eq/shared").DeliveryId;
+        var deliveryB = writerB.Events.Single(e => e.AlarmId == "eq/shared").DeliveryId;
+        deliveryA.ShouldNotBe(Guid.Empty);
+        deliveryB.ShouldBe(deliveryA);
     }
 }

tests/Server/ZB.MOM.WW.OtOpcUa.Host.IntegrationTests/LocalDb/AlarmSfLegacyMigratorTests.cs

@@ -36,6 +36,38 @@ public sealed class AlarmSfLegacyMigratorTests : IDisposable
 
     private ServiceProvider? _provider;
 
+    private sealed class CapturingRetryWriter : IAlarmHistorianWriter
+    {
+        public List<AlarmHistorianEvent> Events { get; } = [];
+        public Task<IReadOnlyList<HistorianWriteOutcome>> WriteBatchAsync(
+            IReadOnlyList<AlarmHistorianEvent> batch, CancellationToken cancellationToken)
+        {
+            Events.AddRange(batch);
+            return Task.FromResult<IReadOnlyList<HistorianWriteOutcome>>(batch.Select(_ => HistorianWriteOutcome.RetryPlease).ToArray());
+        }
+    }
+
+    [Fact]
+    public async Task A_legacy_payload_keeps_its_key_and_stable_delivery_identity_across_drains()
+    {
+        var payload = Payload("eq/a");
+        SeedLegacy((1, "eq/a", payload, 0, dead: 0));
+        var db = BuildDb();
+        AlarmSfLegacyMigrator.Migrate(db, ConfigWithLegacyPath());
+        var row = (await db.QueryAsync("SELECT id, payload_json FROM alarm_sf_events",
+            r => (Id: r.GetString(0), Payload: r.GetString(1)), null, TestContext.Current.CancellationToken)).Single();
+        row.Id.ShouldBe("B1D80B7207C3F3B6CB31105DBF1A6E0D6FE300DA1B3F877AB746B36535310A60");
+        row.Payload.ShouldBe(payload);
+        var writer = new CapturingRetryWriter();
+        using var log = new Serilog.LoggerConfiguration().CreateLogger();
+        using (var sink = new LocalDbStoreAndForwardSink(db, writer, log))
+            await sink.DrainOnceAsync(TestContext.Current.CancellationToken);
+        using (var sink = new LocalDbStoreAndForwardSink(db, writer, log))
+            await sink.DrainOnceAsync(TestContext.Current.CancellationToken);
+        writer.Events.Count.ShouldBe(2);
+        writer.Events.ShouldAllBe(evt => evt.DeliveryId == Guid.Parse("55372883-eaec-8dd7-a1a4-6f98f416a272"));
+    }
+
     // ---- the cases ---------------------------------------------------------------------------
 
     [Fact]

tests/Server/ZB.MOM.WW.OtOpcUa.Host.Tests/OpcUa/FailedServiceLevelBindingCleanupTests.cs

@@ -0,0 +1,45 @@
+using Microsoft.Extensions.Logging.Abstractions;
+using Shouldly;
+using Xunit;
+using ZB.MOM.WW.OtOpcUa.Host.OpcUa;
+
+namespace ZB.MOM.WW.OtOpcUa.Host.Tests.OpcUa;
+
+public sealed class FailedServiceLevelBindingCleanupTests
+{
+    [Fact]
+    public async Task Successful_cleanup_observes_stop_before_disposal_and_releases_ownership()
+    {
+        var events = new List<string>();
+        var released = await FailedServiceLevelBindingCleanup.StopAndDisposeAsync(
+            () => { events.Add("stop"); return Task.CompletedTask; },
+            () => { events.Add("dispose"); return ValueTask.CompletedTask; },
+            NullLogger.Instance);
+        events.ShouldBe(new[] { "stop", "dispose" });
+        released.ShouldBeTrue();
+    }
+
+    [Fact]
+    public async Task Failed_stop_still_disposes_and_retains_ownership_for_retry()
+    {
+        var events = new List<string>();
+        var released = await FailedServiceLevelBindingCleanup.StopAndDisposeAsync(
+            () => { events.Add("stop"); throw new InvalidOperationException("controlled stop failure"); },
+            () => { events.Add("dispose"); return ValueTask.CompletedTask; },
+            NullLogger.Instance);
+        events.ShouldBe(new[] { "stop", "dispose" });
+        released.ShouldBeFalse();
+    }
+
+    [Fact]
+    public async Task Failed_disposal_retains_ownership_without_escaping_the_binding_failure_handler()
+    {
+        var events = new List<string>();
+        var released = await FailedServiceLevelBindingCleanup.StopAndDisposeAsync(
+            () => { events.Add("stop"); return Task.CompletedTask; },
+            () => { events.Add("dispose"); throw new InvalidOperationException("controlled disposal failure"); },
+            NullLogger.Instance);
+        events.ShouldBe(new[] { "stop", "dispose" });
+        released.ShouldBeFalse();
+    }
+}

tests/Server/ZB.MOM.WW.OtOpcUa.Host.Tests/OpcUa/OtOpcUaServerHostedServiceBindingFailureTests.cs

@@ -0,0 +1,103 @@
+using System.Net;
+using System.Net.Sockets;
+using Akka.Hosting;
+using Microsoft.Extensions.Configuration;
+using Microsoft.Extensions.DependencyInjection;
+using Microsoft.Extensions.Hosting;
+using Microsoft.Extensions.Logging;
+using Microsoft.Extensions.Logging.Abstractions;
+using Microsoft.Extensions.Options;
+using Shouldly;
+using Xunit;
+using ZB.MOM.WW.OtOpcUa.Commons.OpcUa;
+using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
+using ZB.MOM.WW.OtOpcUa.Core.Authorization;
+using ZB.MOM.WW.OtOpcUa.Host.OpcUa;
+using ZB.MOM.WW.OtOpcUa.OpcUaServer;
+using ZB.MOM.WW.OtOpcUa.OpcUaServer.Security;
+
+namespace ZB.MOM.WW.OtOpcUa.Host.Tests.OpcUa;
+
+public sealed class OtOpcUaServerHostedServiceBindingFailureTests
+{
+    [Fact]
+    public async Task Binding_failure_closes_started_endpoint_and_allows_other_host_services_to_start()
+    {
+        using var deadline = new CancellationTokenSource(TimeSpan.FromSeconds(15));
+        using var probe = new TcpListener(IPAddress.Loopback, 0);
+        probe.Start();
+        var port = ((IPEndPoint)probe.LocalEndpoint).Port;
+        probe.Stop();
+        var pki = Path.Combine(Path.GetTempPath(), $"otopcua-binding-{Guid.NewGuid():N}");
+        var deferred = new DeferredServiceLevelPublisher();
+        deferred.Publish(0);
+        var bindingReached = false;
+        var listenerObserved = false;
+        OtOpcUaServerHostedService? service = null;
+        var factory = new FaultInjectingLoggerFactory(() =>
+        {
+            using var connected = new TcpClient();
+            connected.Connect(IPAddress.Loopback, port);
+            listenerObserved = connected.Connected;
+            bindingReached = true;
+            throw new InvalidOperationException("controlled binding factory failure");
+        });
+        service = new OtOpcUaServerHostedService(
+            Options.Create(new OpcUaApplicationHostOptions
+            {
+                ApplicationName = "OtOpcUa.BindingFailure",
+                ApplicationUri = $"urn:OtOpcUa.BindingFailure:{Guid.NewGuid():N}",
+                OpcUaPort = port, PublicHostname = "localhost", PkiStoreRoot = pki,
+            }), new DeferredAddressSpaceSink(), deferred,
+            NullOpcUaUserAuthenticator.Instance, NullHistorianDataSource.Instance,
+            EmptyDataPlaneAuthoritySource.Instance, new ConfigurationBuilder().Build(),
+            () => throw new NotSupportedException("no actor system required"), new ActorRegistry(),
+            null, Options.Create(new AclOptions()), TimeProvider.System, factory);
+        var admin = new AdminBootSentinel();
+        using var host = new HostBuilder().ConfigureServices(services =>
+        {
+            services.AddSingleton<IHostedService>(service);
+            services.AddSingleton<IHostedService>(admin);
+        }).Build();
+        try
+        {
+            await host.StartAsync(deadline.Token);
+            listenerObserved.ShouldBeTrue("the SDK endpoint was actually listening before replay");
+            bindingReached.ShouldBeTrue();
+            admin.Started.ShouldBeTrue();
+            using var client = new TcpClient();
+            await Should.ThrowAsync<SocketException>(async () =>
+                await client.ConnectAsync(IPAddress.Loopback, port, deadline.Token));
+            // After construction failure, subsequent requests are accepted into the unbound cache.
+            deferred.Publish(100);
+            await host.StopAsync(deadline.Token);
+        }
+        finally
+        {
+            await service.DisposeAsync();
+            if (Directory.Exists(pki)) Directory.Delete(pki, recursive: true);
+        }
+    }
+
+    private sealed class FaultInjectingLoggerFactory(Action installFault) : ILoggerFactory
+    {
+        public ILogger CreateLogger(string categoryName)
+        {
+            if (categoryName == typeof(SdkServiceLevelPublisher).FullName) installFault();
+            return NullLogger.Instance;
+        }
+        public void AddProvider(ILoggerProvider provider) { }
+        public void Dispose() { }
+    }
+
+    private sealed class AdminBootSentinel : IHostedService
+    {
+        public bool Started { get; private set; }
+        public Task StartAsync(CancellationToken cancellationToken)
+        {
+            Started = true;
+            return Task.CompletedTask;
+        }
+        public Task StopAsync(CancellationToken cancellationToken) => Task.CompletedTask;
+    }
+}

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

@@ -13,8 +13,8 @@ namespace ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests;
 ///     comms-loss watchdog trips on). Recovery is the watchdog's sequence, replayed here as <c>DriverInstanceActor</c>
 ///     runs it: ShutdownAsync, InitializeAsync, re-subscribe — after which the value comes back Good. There is no
 ///     in-driver reconnect any more (#670: the SDK reconnect handler kept everything Good and Healthy forever).
-///     Since #676 a gracefully stopped server is a DEFINITIVE loss (<c>BadServerHalted</c>), so the driver also raises
-///     <c>ConnectionLost</c> and the host runs that sequence at once instead of after the watchdog's silence window.
+///     Since #676 definitive session loss also raises <c>ConnectionLost</c>, so the host runs that sequence at once
+///     instead of after the watchdog's silence window. Shutdown timing determines the SDK's loss status.
 /// </summary>
 [Trait("Category", "Integration")]
 public sealed class OpcUaClientDriverReconnectWireTests
@@ -72,12 +72,22 @@ public sealed class OpcUaClientDriverReconnectWireTests
             await Task.Delay(TimeSpan.FromSeconds(3), ct); // three keep-alive intervals
             drv.GetHealth().LastSuccessfulRead.ShouldBe(frozen);
 
-            // #676: a server that HALTED (graceful stop) is a definitive loss — measured here, the SDK reports
-            // BadServerHalted on every keep-alive and never re-opens the channel, so the session can never recover in
-            // place. The driver says so at once, so the host re-initialises it instead of waiting out the ~90 s watchdog.
+            // #676/#702: shutdown can report a halted server, an invalid/closed session, or a sustained closed
+            // channel. Each proves definitive loss; #702 reproduced BadSessionIdInvalid during this graceful stop.
+            // Deterministic ConnectionLossSignal tests cover the individual statuses and the sustained-channel gate.
             await WaitAsync(() => { lock (lost) return lost.Count > 0; },
                 "a halted upstream must raise ConnectionLost", TimeSpan.FromSeconds(10));
-            lock (lost) lost.Single().ShouldContain("BadServerHalted");
+            lock (lost)
+            {
+                var reason = lost.Single();
+                string[] definitiveStatuses =
+                [
+                    "BadSessionIdInvalid", "BadSessionClosed", "BadSessionNotActivated", "BadServerHalted",
+                    "BadSecureChannelClosed", "BadConnectionClosed",
+                ];
+                definitiveStatuses.ShouldContain(status => reason.EndsWith(status, StringComparison.Ordinal),
+                    $"the restart must report definitive loss, but reported: {reason}");
+            }
         });
 
         // The restarted server knows nothing of the old session, so the driver stays down on its own …

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

@@ -0,0 +1,71 @@
+using Microsoft.Extensions.Logging.Abstractions;
+using Opc.Ua;
+using Opc.Ua.Client;
+using Opc.Ua.Server;
+using Shouldly;
+using Xunit;
+using ZB.MOM.WW.OtOpcUa.Commons.OpcUa;
+
+namespace ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests;
+
+/// <summary>Local loopback evidence that late binding replaces the SDK default on the wire.</summary>
+public sealed class ServiceLevelBindingReplayTests
+{
+    [Theory]
+    [InlineData((byte)100)]
+    [InlineData((byte)0)]
+    public async Task Late_binding_replays_latest_request_to_standard_ServiceLevel_read(byte latest)
+    {
+        using var deadline = CancellationTokenSource.CreateLinkedTokenSource(TestContext.Current.CancellationToken);
+        deadline.CancelAfter(TimeSpan.FromSeconds(15));
+        var ct = deadline.Token;
+        var pki = Path.Combine(Path.GetTempPath(), $"otopcua-replay-{Guid.NewGuid():N}");
+        var clientPki = TestClientSecurity.AllocatePkiRoot();
+        var port = 0;
+        try
+        {
+            var deferred = new DeferredServiceLevelPublisher();
+            deferred.Publish(240);
+            deferred.Publish(latest);
+            var started = await OpcUaHostStart.StartAsync(p =>
+            {
+                port = p;
+                return new OpcUaApplicationHost(new OpcUaApplicationHostOptions
+                {
+                    ApplicationName = "OtOpcUa.Replay",
+                    ApplicationUri = $"urn:OtOpcUa.Replay:{Path.GetFileName(pki)}",
+                    OpcUaPort = p, PublicHostname = "localhost", PkiStoreRoot = pki,
+                }, NullLogger<OpcUaApplicationHost>.Instance);
+            }, () => new StandardServer(), ct);
+            await using var host = started.Host;
+            deferred.SetInner(new SdkServiceLevelPublisher(started.Server.CurrentInstance,
+                NullLogger<SdkServiceLevelPublisher>.Instance));
+            var config = new ApplicationConfiguration
+            {
+                ApplicationName = "OtOpcUa.ReplayClient",
+                ApplicationUri = $"urn:OtOpcUa.ReplayClient:{Guid.NewGuid():N}",
+                ApplicationType = ApplicationType.Client,
+                SecurityConfiguration = TestClientSecurity.Build(clientPki),
+                ClientConfiguration = new ClientConfiguration { DefaultSessionTimeout = 10_000 },
+                TransportQuotas = new TransportQuotas { OperationTimeout = 5_000 },
+            };
+            await config.ValidateAsync(ApplicationType.Client, ct);
+            config.CertificateValidator.CertificateValidation += (_, e) => e.Accept = true;
+            var endpoint = await CoreClientUtils.SelectEndpointAsync(config,
+                $"opc.tcp://localhost:{port}/OtOpcUa", false, DefaultTelemetry.Create(_ => { }), ct);
+            using var session = await new DefaultSessionFactory(DefaultTelemetry.Create(_ => { })).CreateAsync(
+                config, new ConfiguredEndpoint(null, endpoint, EndpointConfiguration.Create(config)),
+                updateBeforeConnect: false, checkDomain: false, sessionName: "ServiceLevelReplay",
+                sessionTimeout: 10_000, identity: new UserIdentity(new AnonymousIdentityToken()),
+                preferredLocales: null, ct: ct);
+            var read = await session.ReadValueAsync(new Opc.Ua.NodeId(2267u), ct);
+            read.StatusCode.ShouldBe(StatusCodes.Good);
+            read.Value.ShouldBe(latest);
+        }
+        finally
+        {
+            if (Directory.Exists(pki)) Directory.Delete(pki, recursive: true);
+            if (Directory.Exists(clientPki)) Directory.Delete(clientPki, recursive: true);
+        }
+    }
+}

tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests/DeferredServiceLevelPublisherTests.cs

@@ -6,9 +6,9 @@ namespace ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests;
 
 public sealed class DeferredServiceLevelPublisherTests
 {
-    /// <summary>Verifies that publish before SetInner is a safe no-op.</summary>
+    /// <summary>Verifies that publish before SetInner safely accepts a cached request.</summary>
     [Fact]
-    public void Publish_before_SetInner_is_a_safe_noop()
+    public void Publish_before_SetInner_accepts_a_cached_request()
     {
         var deferred = new DeferredServiceLevelPublisher();
 
@@ -28,9 +28,9 @@ public sealed class DeferredServiceLevelPublisherTests
         recording.LastValue.ShouldBe((byte)200);
     }
 
-    /// <summary>Verifies that SetInner with null reverts to the null publisher.</summary>
+    /// <summary>Verifies that SetInner with null detaches the previous publisher.</summary>
     [Fact]
-    public void SetInner_null_reverts_to_Null_publisher()
+    public void SetInner_null_detaches_the_previous_publisher()
     {
         var recording = new RecordingPublisher();
         var deferred = new DeferredServiceLevelPublisher();

tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests/SdkServiceLevelPublisherTests.cs

@@ -19,6 +19,40 @@ public sealed class SdkServiceLevelPublisherTests : IDisposable
         Path.GetTempPath(),
         $"otopcua-pki-{Guid.NewGuid():N}");
 
+    [Fact]
+    public void Missing_ServiceLevel_is_reported_as_failure()
+    {
+        var server = System.Reflection.DispatchProxy.Create<IServerInternal, FaultingServer>();
+        var publisher = new SdkServiceLevelPublisher(server, NullLogger<SdkServiceLevelPublisher>.Instance);
+        Should.Throw<InvalidOperationException>(() => publisher.Publish(0));
+    }
+
+    [Fact]
+    public void Sdk_write_failure_propagates_the_original_exception()
+    {
+        var server = System.Reflection.DispatchProxy.Create<IServerInternal, FaultingServer>();
+        var proxy = (FaultingServer)(object)server;
+        proxy.Node = new Opc.Ua.ServerObjectState(null)
+        {
+            ServiceLevel = new Opc.Ua.PropertyState<byte>(null),
+        };
+        proxy.Fault = new InvalidOperationException("controlled diagnostics failure");
+        var publisher = new SdkServiceLevelPublisher(server, NullLogger<SdkServiceLevelPublisher>.Instance);
+        Should.Throw<InvalidOperationException>(() => publisher.Publish(0)).ShouldBeSameAs(proxy.Fault);
+    }
+
+    public class FaultingServer : System.Reflection.DispatchProxy
+    {
+        public Opc.Ua.ServerObjectState? Node { get; set; }
+        public Exception? Fault { get; set; }
+        protected override object? Invoke(System.Reflection.MethodInfo? method, object?[]? args)
+        {
+            if (method?.Name == "get_ServerObject") return Node;
+            if (method?.Name == "get_DiagnosticsLock") throw Fault!;
+            throw new NotSupportedException(method?.Name);
+        }
+    }
+
     /// <summary>Verifies that the publisher writes values to the standard Server.ServiceLevel variable.</summary>
     [Fact]
     public async Task Publish_writes_value_to_Server_ServiceLevel_variable()

tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/OpcUa/OpcUaPublishActorFreshnessTests.cs

@@ -0,0 +1,240 @@
+using System.Collections.Concurrent;
+using Akka.Actor;
+using Akka.TestKit;
+using Akka.TestKit.Xunit2;
+using Shouldly;
+using Xunit;
+using ZB.MOM.WW.OtOpcUa.Commons.Messages.Redundancy;
+using ZB.MOM.WW.OtOpcUa.Commons.Types;
+using ZB.MOM.WW.OtOpcUa.Commons.OpcUa;
+using ZB.MOM.WW.OtOpcUa.OpcUaServer;
+using ZB.MOM.WW.OtOpcUa.Runtime.OpcUa;
+using ZB.MOM.WW.OtOpcUa.Runtime.Health;
+using ZB.MOM.WW.OtOpcUa.Runtime.Tests.Harness;
+
+namespace ZB.MOM.WW.OtOpcUa.Runtime.Tests.OpcUa;
+
+public sealed class OpcUaPublishActorFreshnessTests : RuntimeActorTestBase
+{
+    [Theory]
+    [InlineData(false, 240, 200)]
+    [InlineData(true, 250, 210)]
+    public void DbLess_timer_demotes_without_incoming_messages(bool primary, byte fresh, byte stale)
+    {
+        var publisher = new RecordingPublisher();
+        var clock = new ControlledClock();
+        var local = NodeId.Parse("freshness-node");
+        var actor = Sys.ActorOf(OpcUaPublishActor.PropsForTests(
+            serviceLevel: publisher, localNode: local, dbLess: true,
+            staleWindow: TimeSpan.FromSeconds(1), healthTickInterval: TimeSpan.FromMilliseconds(25),
+            utcNow: clock.Read));
+        actor.Tell(Snapshot(local, primary, clock.Now));
+        AwaitAssert(() => publisher.Levels.ShouldBe(new[] { fresh }), PresenceBudget);
+        // Equality remains fresh. Observe several real ticks with a fixed clock; no actor Tell.
+        clock.Advance(TimeSpan.FromSeconds(1));
+        var reads = clock.Reads;
+        AwaitCondition(() => clock.Reads >= reads + 4, PresenceBudget);
+        // Drain earlier self-enqueued publications without triggering a health recomputation.
+        actor.Tell(new Identify("freshness-barrier"), TestActor);
+        ExpectMsg<ActorIdentity>().Subject.ShouldBe(actor);
+        publisher.Levels.ShouldBe(new[] { fresh });
+        clock.Advance(TimeSpan.FromTicks(1));
+        // No further Tell: only the production timer can notice age crossing the window.
+        AwaitAssert(() => publisher.Levels.ShouldBe(new[] { fresh, stale }), PresenceBudget);
+        // A new fresh snapshot restores service, after which ticks continue to age that snapshot.
+        actor.Tell(Snapshot(local, primary, clock.Now));
+        AwaitAssert(() => publisher.Levels.ShouldBe(new[] { fresh, stale, fresh }), PresenceBudget);
+        clock.Advance(TimeSpan.FromSeconds(1) + TimeSpan.FromTicks(1));
+        AwaitAssert(() => publisher.Levels.ShouldBe(new[] { fresh, stale, fresh, stale }), PresenceBudget);
+    }
+
+    [Fact]
+    public void DbLess_timer_expires_negative_peer_verdict_only_after_exact_boundary()
+    {
+        var publisher = new RecordingPublisher();
+        var clock = new ControlledClock();
+        var local = NodeId.Parse("peer-verdict-node");
+        var actor = Sys.ActorOf(OpcUaPublishActor.PropsForTests(
+            serviceLevel: publisher, localNode: local, dbLess: true,
+            staleWindow: TimeSpan.FromSeconds(10), probeFreshnessWindow: TimeSpan.FromSeconds(2),
+            healthTickInterval: TimeSpan.FromMilliseconds(25), utcNow: clock.Read));
+        actor.Tell(Snapshot(local, false, clock.Now));
+        AwaitAssert(() => publisher.Levels.ShouldBe(new byte[] { 240 }), PresenceBudget);
+        actor.Tell(new PeerOpcUaProbeActor.OpcUaProbeResult(local, Ok: false));
+        AwaitAssert(() => publisher.Levels.ShouldBe(new byte[] { 240, 0 }), PresenceBudget);
+        clock.Advance(TimeSpan.FromSeconds(2));
+        var reads = clock.Reads;
+        AwaitCondition(() => clock.Reads >= reads + 4, PresenceBudget);
+        // Drain earlier self-enqueued publications without triggering a health recomputation.
+        actor.Tell(new Identify("freshness-barrier"), TestActor);
+        ExpectMsg<ActorIdentity>().Subject.ShouldBe(actor);
+        publisher.Levels.ShouldBe(new byte[] { 240, 0 });
+        clock.Advance(TimeSpan.FromTicks(1));
+        AwaitAssert(() => publisher.Levels.ShouldBe(new byte[] { 240, 0, 240 }), PresenceBudget);
+    }
+
+    [Fact]
+    public void Stopping_DbLess_actor_cancels_freshness_timer_and_publications()
+    {
+        var publisher = new RecordingPublisher();
+        var clock = new ControlledClock();
+        var local = NodeId.Parse("stopped-node");
+        var actor = Sys.ActorOf(OpcUaPublishActor.PropsForTests(
+            serviceLevel: publisher, localNode: local, dbLess: true,
+            staleWindow: TimeSpan.FromSeconds(1), healthTickInterval: TimeSpan.FromMilliseconds(25),
+            utcNow: clock.Read));
+        actor.Tell(Snapshot(local, false, clock.Now));
+        AwaitAssert(() => publisher.Levels.ShouldBe(new byte[] { 240 }), PresenceBudget);
+        var reads = clock.Reads;
+        AwaitCondition(() => clock.Reads >= reads + 4, PresenceBudget);
+        Watch(actor);
+        Sys.Stop(actor);
+        ExpectTerminated(actor);
+        reads = clock.Reads;
+        clock.Advance(TimeSpan.FromSeconds(2));
+        ExpectNoMsg(TimeSpan.FromMilliseconds(150)); // Six timer periods after confirmed termination.
+        clock.Reads.ShouldBe(reads);
+        publisher.Levels.ShouldBe(new byte[] { 240 });
+    }
+
+    [Fact]
+    public void DbBacked_timer_refreshes_probe_status_periodically()
+    {
+        var publisher = new RecordingPublisher();
+        var clock = new ControlledClock();
+        var local = NodeId.Parse("db-backed-periodic-node");
+        var db = CreateTestProbe();
+        var actor = Sys.ActorOf(OpcUaPublishActor.PropsForTests(
+            serviceLevel: publisher, localNode: local, dbHealthProbe: db.Ref,
+            healthTickInterval: TimeSpan.FromMilliseconds(100), utcNow: clock.Read));
+        actor.Tell(Snapshot(local, false, clock.Now));
+        db.ExpectMsg<DbHealthProbeActor.GetStatus>();
+        db.Reply(new DbHealthProbeActor.DbHealthStatus(true, clock.Now, null));
+        AwaitAssert(() => publisher.Levels.ShouldContain((byte)240), PresenceBudget);
+        // A second request comes from the periodic timer, not the immediate startup tick.
+        db.ExpectMsg<DbHealthProbeActor.GetStatus>();
+        db.Reply(new DbHealthProbeActor.DbHealthStatus(false, clock.Now, "unreachable"));
+        AwaitAssert(() => publisher.Levels.Last().ShouldBe((byte)100), PresenceBudget);
+    }
+
+    [Fact]
+    public void DbBacked_timer_ask_timeout_demotes_last_known_healthy_node()
+    {
+        var publisher = new RecordingPublisher();
+        var clock = new ControlledClock();
+        var local = NodeId.Parse("db-backed-timeout-node");
+        var db = CreateTestProbe();
+        var actor = Sys.ActorOf(OpcUaPublishActor.PropsForTests(
+            serviceLevel: publisher, localNode: local, dbHealthProbe: db.Ref,
+            healthTickInterval: TimeSpan.FromMilliseconds(100), utcNow: clock.Read));
+        actor.Tell(Snapshot(local, true, clock.Now));
+        db.ExpectMsg<DbHealthProbeActor.GetStatus>();
+        db.Reply(new DbHealthProbeActor.DbHealthStatus(true, clock.Now, null));
+        AwaitAssert(() => publisher.Levels.ShouldContain((byte)250), PresenceBudget);
+        // Leave subsequent real Ask requests unanswered: the timeout status must reach the actor.
+        db.ExpectMsg<DbHealthProbeActor.GetStatus>();
+        AwaitAssert(() => publisher.Levels.Last().ShouldBe((byte)110), PresenceBudget);
+    }
+
+    [Fact]
+    public void Missing_local_snapshot_entry_retries_zero_on_timer_and_recovers()
+    {
+        var publisher = new RetryZeroPublisher();
+        var clock = new ControlledClock();
+        var local = NodeId.Parse("missing-entry-node");
+        var actor = Sys.ActorOf(OpcUaPublishActor.PropsForTests(
+            serviceLevel: publisher, localNode: local, dbLess: true,
+            healthTickInterval: TimeSpan.FromMilliseconds(25), utcNow: clock.Read));
+        actor.Tell(Snapshot(local, false, clock.Now));
+        AwaitAssert(() => publisher.Levels.ShouldBe(new byte[] { 240 }), PresenceBudget);
+
+        // Only this snapshot Tell requests the first zero. Fail it and the next attempt so
+        // subsequent zero attempts must come from the periodic timer, exercising the null-entry
+        // branch repeatedly rather than mistaking the Detached branch for missing-node coverage.
+        actor.Tell(Snapshot(NodeId.Parse("other-node"), false, clock.Now));
+        AwaitAssert(() => publisher.Levels.ShouldBe(new byte[] { 240, 0 }), PresenceBudget);
+        publisher.ZeroAttempts.ShouldBeGreaterThanOrEqualTo(3);
+
+        actor.Tell(Snapshot(local, false, clock.Now));
+        AwaitAssert(() => publisher.Levels.ShouldBe(new byte[] { 240, 0, 240 }), PresenceBudget);
+        clock.Advance(TimeSpan.FromSeconds(30) + TimeSpan.FromTicks(1));
+        AwaitAssert(() => publisher.Levels.ShouldBe(new byte[] { 240, 0, 240, 200 }), PresenceBudget);
+    }
+
+    [Fact]
+    public void No_snapshot_startup_does_not_publish_across_virtual_timer_periods()
+    {
+        // This actor requires the Cluster extension, but the absent/missing-entry guards do
+        // not inspect membership. Do not self-join: the ordinary harness waits for Up in its
+        // constructor, which cannot progress with a frozen scheduler.
+        using var local = new TestKit(AkkaTestHocon + @"
+            akka.scheduler.implementation = ""Akka.TestKit.TestScheduler, Akka.TestKit""");
+        var scheduler = (TestScheduler)local.Sys.Scheduler;
+        var publisher = new RetryZeroPublisher();
+        var clock = new ControlledClock();
+        var actor = local.ActorOfAsTestActorRef<OpcUaPublishActor>(OpcUaPublishActor.PropsForTests(
+            serviceLevel: publisher, localNode: NodeId.Parse("no-snapshot-node"), dbLess: true,
+            healthTickInterval: TimeSpan.FromSeconds(1), utcNow: clock.Read));
+
+        // TestActorRef uses the calling-thread dispatcher: Advance completes delivery before
+        // these assertions, including the immediate startup tick and three periodic ticks.
+        for (var period = 0; period < 3; period++)
+            scheduler.Advance(TimeSpan.FromSeconds(1));
+        publisher.Levels.ShouldBeEmpty();
+        publisher.ZeroAttempts.ShouldBe(0);
+        clock.Reads.ShouldBe(0);
+
+        // Positive control: the same registered timer must subsequently retry a failed zero.
+        // No whole snapshot at startup means no publication; a received snapshot missing the
+        // local entry means zero. Keep these two policies explicitly distinct.
+        actor.Tell(Snapshot(NodeId.Parse("other-node"), false, clock.Now));
+        publisher.ZeroAttempts.ShouldBe(1);
+        scheduler.Advance(TimeSpan.FromSeconds(1));
+        publisher.ZeroAttempts.ShouldBe(2);
+        publisher.Levels.ShouldBeEmpty();
+        scheduler.Advance(TimeSpan.FromSeconds(1));
+        publisher.ZeroAttempts.ShouldBe(3);
+        publisher.Levels.ShouldBe(new byte[] { 0 });
+        clock.Reads.ShouldBe(0);
+    }
+
+    private sealed class RetryZeroPublisher : IServiceLevelPublisher
+    {
+        private readonly ConcurrentQueue<byte> _levels = new();
+        private int _zeroAttempts;
+        public byte[] Levels => _levels.ToArray();
+        public int ZeroAttempts => Volatile.Read(ref _zeroAttempts);
+        public void Publish(byte serviceLevel)
+        {
+            if (serviceLevel == 0 && Interlocked.Increment(ref _zeroAttempts) <= 2)
+                throw new InvalidOperationException("test zero-publication failure");
+            _levels.Enqueue(serviceLevel);
+        }
+    }
+
+    private sealed class ControlledClock
+    {
+        private long _ticks = new DateTime(2026, 1, 1, 0, 0, 0, DateTimeKind.Utc).Ticks;
+        private int _reads;
+        public DateTime Now => new(Interlocked.Read(ref _ticks), DateTimeKind.Utc);
+        public int Reads => Volatile.Read(ref _reads);
+        public DateTime Read()
+        {
+            Interlocked.Increment(ref _reads);
+            return Now;
+        }
+        public void Advance(TimeSpan age) => Interlocked.Add(ref _ticks, age.Ticks);
+    }
+
+    private static RedundancyStateChanged Snapshot(NodeId local, bool primary, DateTime asOf) =>
+        new(new[] { new NodeRedundancyState(local,
+            primary ? RedundancyRole.Primary : RedundancyRole.Secondary,
+            IsClusterLeader: primary, IsDriverPrimary: primary, asOf) }, CorrelationId.NewId());
+
+    private sealed class RecordingPublisher : IServiceLevelPublisher
+    {
+        private readonly ConcurrentQueue<byte> _levels = new();
+        public byte[] Levels => _levels.ToArray();
+        public void Publish(byte serviceLevel) => _levels.Enqueue(serviceLevel);
+    }
+}

tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/OpcUa/OpcUaPublishActorTests.cs

@@ -15,6 +15,49 @@ namespace ZB.MOM.WW.OtOpcUa.Runtime.Tests.OpcUa;
 
 public sealed class OpcUaPublishActorTests : RuntimeActorTestBase
 {
+    [Fact]
+    public void Failed_value_is_retried_and_last_level_only_tracks_success()
+    {
+        var sink = new RecoveringPublisher { Fail = false };
+        var actor = ActorOfAsTestActorRef<OpcUaPublishActor>(OpcUaPublishActor.PropsForTests(serviceLevel: sink));
+        actor.Tell(new OpcUaPublishActor.ServiceLevelChanged(100));
+        sink.Fail = true;
+        actor.Tell(new OpcUaPublishActor.ServiceLevelChanged(0));
+        actor.UnderlyingActor.LastServiceLevel.ShouldBe((byte)100);
+        sink.Fail = false;
+        actor.Tell(new OpcUaPublishActor.ServiceLevelChanged(0));
+        sink.Delivered.ShouldBe(new byte[] { 100, 0 });
+    }
+
+    [Fact]
+    public void Requesting_prior_success_after_failure_updates_delivery_and_replay()
+    {
+        var sink = new RecoveringPublisher();
+        var deferred = new DeferredServiceLevelPublisher();
+        deferred.SetInner(sink);
+        var actor = ActorOfAsTestActorRef<OpcUaPublishActor>(OpcUaPublishActor.PropsForTests(serviceLevel: deferred));
+        actor.Tell(new OpcUaPublishActor.ServiceLevelChanged(100));
+        sink.Fail = true;
+        actor.Tell(new OpcUaPublishActor.ServiceLevelChanged(0));
+        sink.Fail = false;
+        actor.Tell(new OpcUaPublishActor.ServiceLevelChanged(100));
+        sink.Delivered.ShouldBe(new byte[] { 100, 100 });
+        var rebound = new RecoveringPublisher();
+        deferred.SetInner(rebound);
+        rebound.Delivered.ShouldBe(new byte[] { 100 });
+    }
+
+    private sealed class RecoveringPublisher : IServiceLevelPublisher
+    {
+        public bool Fail { get; set; }
+        public List<byte> Delivered { get; } = new();
+        public void Publish(byte level)
+        {
+            if (Fail) throw new InvalidOperationException("controlled failure");
+            Delivered.Add(level);
+        }
+    }
+
     /// <summary>Verifies that message contracts are accepted without a pinned dispatcher in tests.</summary>
     [Fact]
     public void Accepts_message_contracts_without_pinned_dispatcher_in_tests()

tests/TestSupport/FixtureCleanup.cs

@@ -0,0 +1,32 @@
+namespace ZB.MOM.WW.OtOpcUa.Tests;
+
+internal static class FixtureCleanup
+{
+    internal const string SecondaryExceptionKey = "FixtureCleanup.SecondaryException";
+
+    internal static async Task RunAsync(Func<Task> body, Func<Task> cleanup, Action<Exception> report)
+    {
+        Exception? primary = null;
+        try
+        {
+            await body();
+        }
+        catch (Exception failure)
+        {
+            primary = failure;
+            throw;
+        }
+        finally
+        {
+            try
+            {
+                await cleanup();
+            }
+            catch (Exception secondary) when (primary is not null)
+            {
+                try { primary.Data[SecondaryExceptionKey] = secondary; } catch { }
+                try { report(secondary); } catch { }
+            }
+        }
+    }
+}