test: verify missing-snapshot ServiceLevel timer behavior

.github/workflows/v2-ci.yml

@@ -41,6 +41,8 @@ jobs:
       # #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 --disable-parallel -m:2

scripts/ci/README.md

@@ -28,3 +28,26 @@ AKKA_TEST_TIMEFACTOR=3 dotnet test ZB.MOM.WW.OtOpcUa.slnx \
   --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/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/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;

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);
@@ -1006,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
@@ -1015,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);
     }
 
@@ -1032,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();
     }
 
@@ -1044,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;
     }
 
@@ -1081,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,
@@ -1101,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/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);
+    }
+}