fix(deploy): hold the Applied ack until the address-space apply outcome is known (#601)

DriverHostActor no longer acks Applied the moment the revision advances. It
asks OpcUaPublishActor for the outcome (RebuildAddressSpace.ReportOutcome) and
acks Applied+Degraded(reason) when the apply was degraded; a missing outcome
times out (60 s, inside the coordinator's 2 min deadline) to a degraded ack,
never a clean one. Re-dispatching a revision whose apply was degraded or is
still unconfirmed re-applies instead of taking the clean idempotent
short-circuit - that re-apply is also what repairs the degraded nodes.
Tests with a bare publish probe use a clean-outcome auto-pilot probe.

Claude-Session: https://claude.ai/code/session_01Ef2ozxS7QPanGT8uoATbTe

src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Drivers/DriverHostActor.cs

@@ -84,6 +84,19 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS
     /// </remarks>
     public static readonly TimeSpan DefaultApplyWedgeTimeout = TimeSpan.FromMinutes(5);
 
+    /// <summary>
+    ///     Gitea #601 — how long an Applied ack waits for <c>OpcUaPublishActor</c>'s address-space apply
+    ///     outcome (<see cref="OpcUa.OpcUaPublishActor.RebuildCompleted"/>) before it is sent anyway.
+    /// </summary>
+    /// <remarks>
+    ///     Deliberately well inside the coordinator's apply deadline
+    ///     (<c>ConfigPublishCoordinator.DefaultApplyDeadline</c>, 2 min, which also has to cover the artifact
+    ///     read/fetch): a hung or very slow apply must still produce an ack before central times the
+    ///     deployment out. An ack sent on this timeout is marked <b>degraded</b> ("outcome unknown"), never
+    ///     clean — the revision did advance, but nobody has confirmed the address space matches it.
+    /// </remarks>
+    public static readonly TimeSpan DefaultApplyOutcomeTimeout = TimeSpan.FromSeconds(60);
+
     /// <summary>Minimum interval between two supervision reprovisions of the SAME driver child (archreview
     /// review F3). Akka's default decider restarts a faulting child immediately and without limit, and the
     /// reprovision hook makes every restart do ShutdownAsync + InitializeAsync + SubscribeAsync — so an
@@ -244,6 +257,21 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS
     private IActorRef? _scriptedAlarmHost;
 
     private RevisionHash? _currentRevision;
+
+    /// <summary>
+    ///     Gitea #601 — the last address-space apply of <see cref="_currentRevision"/> reported DEGRADED (or no
+    ///     outcome in time). A re-dispatch of that same revision then RE-APPLIES instead of taking the
+    ///     idempotent "already at this revision" short-circuit: the short-circuit would ack a clean Applied
+    ///     for an address space known to be partial, and the re-apply is what repairs it (a degraded
+    ///     non-rebuild apply leaves the publish actor's composition un-advanced, so it re-diffs the missing
+    ///     nodes). Reset by every clean outcome and by every boot path that restores a revision.
+    /// </summary>
+    private bool _currentRevisionDegraded;
+
+    /// <summary>Gitea #601 — Applied acks held until the address-space apply reports its outcome, keyed by
+    /// the apply's correlation id (echoed back in <see cref="OpcUa.OpcUaPublishActor.RebuildCompleted"/>).</summary>
+    private readonly Dictionary<CorrelationId, PendingAck> _pendingAcks = new();
+
     private DeploymentId? _applyingDeploymentId;
 
     private readonly Dictionary<string, ChildEntry> _children = new(StringComparer.Ordinal);
@@ -431,6 +459,10 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS
     /// </summary>
     private readonly Func<CommonsNodeId, bool>? _sharesHistorizationScopeWithLocal;
 
+    /// <summary>Gitea #601 — how long an Applied ack waits for the address-space outcome (see
+    /// <see cref="DefaultApplyOutcomeTimeout"/>).</summary>
+    private readonly TimeSpan _applyOutcomeTimeout;
+
     /// <summary>This host's <see cref="Applying"/>-wedge watchdog delay (see
     /// <see cref="DefaultApplyWedgeTimeout"/>).</summary>
     private readonly TimeSpan _applyWedgeTimeout;
@@ -595,6 +627,9 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS
     /// <param name="commsRecoveryAuditDelay">Archreview 03/S15: how long after a driver reconnects the
     /// recovery audit waits before reporting refs still latched <c>BadNoCommunication</c>. Defaults to
     /// <see cref="DefaultCommsRecoveryAuditDelay"/>; tests shorten it.</param>
+    /// <param name="applyOutcomeTimeout">Gitea #601: how long an Applied ack waits for the address-space
+    /// apply outcome before it is sent anyway, marked degraded ("outcome unknown"). Defaults to
+    /// <see cref="DefaultApplyOutcomeTimeout"/>; tests shorten it.</param>
     /// <returns>The Akka.NET <see cref="Akka.Actor.Props"/> used to spawn this actor.</returns>
     public static Props Props(
         IDbContextFactory<OtOpcUaConfigDbContext>? dbFactory,
@@ -626,7 +661,8 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS
         TimeSpan? applyWedgeTimeout = null,
         TimeSpan? reprovisionBackoff = null,
         TimeSpan? commsRecoveryAuditDelay = null,
-        Func<CommonsNodeId, bool>? sharesAlarmQueueWithLocal = null) =>
+        Func<CommonsNodeId, bool>? sharesAlarmQueueWithLocal = null,
+        TimeSpan? applyOutcomeTimeout = null) =>
         // WARNING: this forwarding list is POSITIONAL, and Props.Create compiles it into an
         // expression tree. Six IActorRef? parameters and several interface-typed ones mean a
         // mis-ordered argument is usually type-compatible and therefore compiles clean, then binds
@@ -639,7 +675,7 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS
             deploymentArtifactCache, redundancyRoleView, replicationPeerHost, fetchAndCacheMode, artifactFetcher,
             alarmStateStore, telemetryHub, primaryRoleSource, historizationRoleView,
             sharesHistorizationScopeWithLocal, applyWedgeTimeout, reprovisionBackoff, commsRecoveryAuditDelay,
-            sharesAlarmQueueWithLocal));
+            sharesAlarmQueueWithLocal, applyOutcomeTimeout));
 
     /// <summary>Initializes a new DriverHostActor with the specified dependencies.</summary>
     /// <param name="dbFactory">Database context factory for configuration database access.</param>
@@ -737,8 +773,10 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS
         TimeSpan? applyWedgeTimeout = null,
         TimeSpan? reprovisionBackoff = null,
         TimeSpan? commsRecoveryAuditDelay = null,
-        Func<CommonsNodeId, bool>? sharesAlarmQueueWithLocal = null)
+        Func<CommonsNodeId, bool>? sharesAlarmQueueWithLocal = null,
+        TimeSpan? applyOutcomeTimeout = null)
     {
+        _applyOutcomeTimeout = applyOutcomeTimeout ?? DefaultApplyOutcomeTimeout;
         _sharesAlarmQueueWithLocal = sharesAlarmQueueWithLocal;
         _applyWedgeTimeout = applyWedgeTimeout ?? DefaultApplyWedgeTimeout;
         _reprovisionBackoff = reprovisionBackoff ?? DefaultReprovisionBackoff;
@@ -1288,6 +1326,10 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS
     private void Steady()
     {
         Receive<DispatchDeployment>(HandleDispatchFromSteady);
+        // #601 — the held Applied ack is released by the address-space outcome (or its timeout); both can
+        // land in any behaviour, so all three register them.
+        Receive<OpcUa.OpcUaPublishActor.RebuildCompleted>(HandleRebuildCompleted);
+        Receive<ApplyOutcomeTimedOut>(HandleApplyOutcomeTimedOut);
         Receive<ArtifactCached>(HandleArtifactCached);
         Receive<ArtifactCacheFailed>(HandleArtifactCacheFailed);
         // A wedge tick already in the mailbox when the apply finished — dropped at Debug, see the handler.
@@ -1348,6 +1390,8 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS
         // Direct ConfigDb read (BeginFetchAndCacheApply / BeginDirectApply ran in Steady, then
         // Become(Applying)); completing the apply transitions back to Steady.
         Receive<ArtifactForApply>(HandleArtifactForApply);
+        Receive<OpcUa.OpcUaPublishActor.RebuildCompleted>(HandleRebuildCompleted);
+        Receive<ApplyOutcomeTimedOut>(HandleApplyOutcomeTimedOut);
         // The cache write is piped back too, and it outlives the apply that started it by design — so
         // its completion can land in EITHER state and both must handle it or it dead-letters.
         Receive<ArtifactCached>(HandleArtifactCached);
@@ -2173,6 +2217,10 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS
         // The redundancy-state topic also carries OpcUaProbeResult (OpcUaPublishActor peer-probes).
         // We don't consume it here — drop it so it doesn't dead-letter (matches PeerProbeSupervisor).
         Receive<ZB.MOM.WW.OtOpcUa.Runtime.Health.PeerOpcUaProbeActor.OpcUaProbeResult>(_ => { });
+        // #601 — an apply that completed just before the host went Stale still owes its ack; the
+        // coordinator is not behind the ConfigDb, so release it exactly as in Steady.
+        Receive<OpcUa.OpcUaPublishActor.RebuildCompleted>(HandleRebuildCompleted);
+        Receive<ApplyOutcomeTimedOut>(HandleApplyOutcomeTimedOut);
         // An inbound operator write can't be serviced while the config DB is unreachable — fast-fail so the
         // node-manager's bounded Ask gets an immediate clear status instead of dead-lettering into a timeout.
         Receive<RouteNodeWrite>(_ =>
@@ -2219,7 +2267,12 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS
 
     private void HandleDispatchFromSteady(DispatchDeployment msg)
     {
-        if (_currentRevision is { } cur && cur == msg.RevisionHash)
+        // #601: the short-circuit claims a CLEAN Applied, so it is only honest while the current revision's
+        // address space is known-good — not after a degraded apply, and not while its outcome is still
+        // pending. Either way the dispatch re-applies instead, which is also what repairs a degraded apply.
+        var currentRevisionUnconfirmed = _currentRevisionDegraded
+            || _pendingAcks.Values.Any(p => p.Revision == msg.RevisionHash);
+        if (_currentRevision is { } cur && cur == msg.RevisionHash && !currentRevisionUnconfirmed)
         {
             // Idempotent — already at this rev. Ack and stay Steady.
             _log.Debug("DriverHost {Node}: dispatch {Id} matches current rev {Rev}; immediate ACK",
@@ -2228,6 +2281,11 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS
             return;
         }
 
+        if (_currentRevision is { } same && same == msg.RevisionHash)
+            _log.Info(
+                "DriverHost {Node}: dispatch {Id} matches current rev {Rev}, but its address-space apply was degraded or is unconfirmed — re-applying rather than acking clean",
+                _localNode, msg.DeploymentId, msg.RevisionHash);
+
         if (_fetchAndCacheMode)
         {
             BeginFetchAndCacheApply(msg.DeploymentId, msg.RevisionHash, msg.CorrelationId);
@@ -2387,7 +2445,13 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS
 
             _currentRevision = msg.Revision;
             UpsertNodeDeploymentState(msg.DeploymentId, NodeDeploymentStatus.Applied, failureReason: null);
-            SendAck(msg.DeploymentId, ApplyAckOutcome.Applied, failureReason: null, msg.Correlation);
+            // Gitea #601: the Applied ack is NOT sent here any more. It used to go out right now, before the
+            // address-space apply below had even been dequeued, so a degraded apply (swallowed sink faults,
+            // _lastApplied not advanced) still sealed the deployment green fleet-wide. The ack is now held
+            // until OpcUaPublishActor replies RebuildCompleted (HandleRebuildCompleted) or the outcome
+            // timeout fires (sent degraded, "outcome unknown"). With no publish actor there is no address
+            // space to wait for, so the ack goes out clean immediately — exactly the old behaviour.
+            AwaitAddressSpaceOutcomeThenAck(msg.DeploymentId, msg.Revision, msg.Correlation);
             // Rebuild the address space + push subscriptions FROM the bytes this apply already has, so
             // nothing downstream re-reads the artifact: under FetchAndCache OpcUaPublishActor has no
             // central SQL to read at all, and under Direct a bytes-less RebuildAddressSpace would make
@@ -2399,8 +2463,11 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS
             // different actor with its own mailbox, and ParsedDeploymentArtifact is a disposable,
             // non-thread-safe object owned by this handler (disposed in the finally below, likely before
             // the other actor even dequeues). It opens its own single parse on the far side.
+            // ReportOutcome: the reply (to Self — this Tell runs inside a receive, so the implicit sender is
+            // this actor) releases the held ack. Plain Tell + reply message, no Ask/await, so nothing here
+            // resumes off the actor thread.
             _opcUaPublishActor?.Tell(new ZB.MOM.WW.OtOpcUa.Runtime.OpcUa.OpcUaPublishActor.RebuildAddressSpace(
-                msg.Correlation, msg.DeploymentId, appliedBlob));
+                msg.Correlation, msg.DeploymentId, appliedBlob, ReportOutcome: true));
             // SubscribeBulk pass: hand each driver its desired tag references so live values flow into
             // the just-rebuilt address space instead of staying BadWaitingForInitialData.
             PushDesiredSubscriptionsFromParsed(msg.DeploymentId, parsed);
@@ -2420,6 +2487,9 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS
         }
         catch (Exception ex)
         {
+            // #601: a throw after the Applied ack was put on hold — the Failed ack below is this apply's one
+            // answer, so the held Applied must never follow it when the address-space outcome lands.
+            DropPendingAck(msg.Correlation);
             UpsertNodeDeploymentState(msg.DeploymentId, NodeDeploymentStatus.Failed, ex.Message);
             SendAck(msg.DeploymentId, ApplyAckOutcome.Failed, ex.Message, msg.Correlation);
             OtOpcUaTelemetry.DeploymentApplied.Add(1, new KeyValuePair<string, object?>("outcome", "reject"));
@@ -2440,6 +2510,94 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS
         }
     }
 
+    /// <summary>Gitea #601 — an Applied ack held until the address-space apply reports its outcome.</summary>
+    private sealed record PendingAck(DeploymentId DeploymentId, RevisionHash Revision, CorrelationId Correlation);
+
+    /// <summary>Gitea #601 — self-timer: the address-space outcome for this correlation did not arrive in time.</summary>
+    private sealed record ApplyOutcomeTimedOut(CorrelationId Correlation);
+
+    private static string ApplyOutcomeTimerKey(CorrelationId correlation) => $"apply-outcome-{correlation}";
+
+    /// <summary>
+    ///     Gitea #601 — holds the Applied ack for an apply whose revision has just advanced until the
+    ///     address-space apply outcome is known, arming <see cref="_applyOutcomeTimeout"/> as the backstop.
+    ///     With no publish actor wired (tests, a node with no OPC UA data plane) there is no outcome to wait
+    ///     for, so the ack is sent clean at once — byte-for-byte the pre-#601 behaviour.
+    /// </summary>
+    private void AwaitAddressSpaceOutcomeThenAck(DeploymentId deploymentId, RevisionHash revision, CorrelationId correlation)
+    {
+        if (_opcUaPublishActor is null)
+        {
+            _currentRevisionDegraded = false;
+            SendAck(deploymentId, ApplyAckOutcome.Applied, failureReason: null, correlation);
+            return;
+        }
+
+        if (_pendingAcks.Remove(correlation, out var superseded))
+            _log.Warning(
+                "DriverHost {Node}: a second apply reused correlation {Correlation} while deployment {Old}'s ack was still waiting for its address-space outcome — that ack is superseded",
+                _localNode, correlation, superseded.DeploymentId);
+        _pendingAcks[correlation] = new PendingAck(deploymentId, revision, correlation);
+        Timers.StartSingleTimer(ApplyOutcomeTimerKey(correlation), new ApplyOutcomeTimedOut(correlation), _applyOutcomeTimeout);
+    }
+
+    /// <summary>Gitea #601 — forgets a held ack (its apply failed after all, and acks Failed instead).</summary>
+    private void DropPendingAck(CorrelationId correlation)
+    {
+        if (_pendingAcks.Remove(correlation))
+            Timers.Cancel(ApplyOutcomeTimerKey(correlation));
+    }
+
+    /// <summary>
+    ///     Gitea #601 — the address-space apply reported its outcome: release the held ack, clean or
+    ///     degraded. Registered in every behaviour (Steady, Applying, Stale) because the reply can land in
+    ///     any of them — the host returns to Steady before the publish actor has even dequeued the rebuild.
+    /// </summary>
+    private void HandleRebuildCompleted(OpcUa.OpcUaPublishActor.RebuildCompleted msg)
+    {
+        if (!_pendingAcks.Remove(msg.Correlation, out var pending))
+        {
+            _log.Debug("DriverHost {Node}: address-space outcome for {Correlation} has no held ack (already timed out or superseded); dropping",
+                _localNode, msg.Correlation);
+            return;
+        }
+
+        Timers.Cancel(ApplyOutcomeTimerKey(msg.Correlation));
+        ReleasePendingAck(pending, msg.Degraded, msg.DegradedReason);
+    }
+
+    /// <summary>
+    ///     Gitea #601 — the address-space outcome did not arrive within <see cref="_applyOutcomeTimeout"/>.
+    ///     The ack is sent anyway so the coordinator does not time the deployment out on a node that DID
+    ///     advance its revision — but degraded, never clean: nobody has confirmed the address space.
+    /// </summary>
+    private void HandleApplyOutcomeTimedOut(ApplyOutcomeTimedOut msg)
+    {
+        if (!_pendingAcks.Remove(msg.Correlation, out var pending)) return;
+
+        ReleasePendingAck(pending, degraded: true,
+            $"the address-space apply did not report an outcome within {_applyOutcomeTimeout.TotalSeconds:0}s; it may still be running or may have been lost");
+    }
+
+    private void ReleasePendingAck(PendingAck pending, bool degraded, string? reason)
+    {
+        // Only the CURRENT revision's flag is ours to set — an older apply's late outcome must not mark a
+        // newer revision degraded (or clean).
+        if (_currentRevision is { } cur && cur == pending.Revision)
+            _currentRevisionDegraded = degraded;
+
+        if (degraded)
+        {
+            OtOpcUaTelemetry.DeploymentApplied.Add(1, new KeyValuePair<string, object?>("outcome", "degraded"));
+            _log.Warning(
+                "DriverHost {Node}: deployment {Id} (rev {Rev}) applied DEGRADED — {Reason}. Acking Applied+degraded; re-dispatching the same revision re-applies it",
+                _localNode, pending.DeploymentId, pending.Revision, reason);
+        }
+
+        SendAck(pending.DeploymentId, ApplyAckOutcome.Applied, failureReason: null, pending.Correlation,
+            degraded, degraded ? reason ?? "address-space apply degraded" : null);
+    }
+
     /// <summary>
     ///     Self-timer telling this host it has been in <see cref="Applying"/> for
     ///     <see cref="_applyWedgeTimeout"/> without finishing.
@@ -3662,9 +3820,11 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS
         }
     }
 
-    private void SendAck(DeploymentId deploymentId, ApplyAckOutcome outcome, string? failureReason, CorrelationId correlation)
+    private void SendAck(
+        DeploymentId deploymentId, ApplyAckOutcome outcome, string? failureReason, CorrelationId correlation,
+        bool degraded = false, string? degradedReason = null)
     {
-        var ack = new ApplyAck(deploymentId, _localNode, outcome, failureReason, correlation);
+        var ack = new ApplyAck(deploymentId, _localNode, outcome, failureReason, correlation, degraded, degradedReason);
         if (_ackRouter is not null)
         {
             // Either a direct coordinator handle (tests) or, under MeshTransport:Mode = Grpc, the

tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverHostActorBootFromCacheTests.cs

@@ -67,7 +67,7 @@ public sealed class DriverHostActorBootFromCacheTests : RuntimeActorTestBase
         var blob = EmptyArtifact();
         var cached = new CachedDeploymentArtifact(
             DeploymentId.NewId().ToString(), CachedRev.Value, blob, DateTimeOffset.UtcNow.AddHours(-1));
-        var publishProbe = CreateTestProbe();
+        var publishProbe = this.CreateCleanOutcomePublishProbe();
 
         Sys.ActorOf(DriverHostActor.Props(
             new ThrowingDbFactory(), TestNode, ackRouter: null,
@@ -92,7 +92,7 @@ public sealed class DriverHostActorBootFromCacheTests : RuntimeActorTestBase
         var cached = new CachedDeploymentArtifact(
             DeploymentId.NewId().ToString(), CachedRev.Value,
             System.Text.Encoding.Latin1.GetBytes(text), DateTimeOffset.UtcNow.AddHours(-1));
-        var publishProbe = CreateTestProbe();
+        var publishProbe = this.CreateCleanOutcomePublishProbe();
 
         var actor = Sys.ActorOf(DriverHostActor.Props(
             new ThrowingDbFactory(), TestNode, ackRouter: null,

tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverHostActorDbLessTests.cs

@@ -112,7 +112,7 @@ public sealed class DriverHostActorDbLessTests : RuntimeActorTestBase
         var cache = new RecordingArtifactCache();
         var fetcher = new RecordingFetcher(ArtifactBytes());
         var deploymentId = DeploymentId.NewId();
-        var publish = CreateTestProbe();
+        var publish = this.CreateCleanOutcomePublishProbe();
         var coordinator = CreateTestProbe();
 
         var actor = Sys.ActorOf(DriverHostActor.Props(
@@ -141,7 +141,7 @@ public sealed class DriverHostActorDbLessTests : RuntimeActorTestBase
     [Fact]
     public void DbLessNode_WithLocalDbStore_SpawnsScriptedAlarmHost()
     {
-        var publish = CreateTestProbe();
+        var publish = this.CreateCleanOutcomePublishProbe();
         var actor = Sys.ActorOf(DriverHostActor.Props(
             dbFactory: null,
             TestNode,
@@ -162,7 +162,7 @@ public sealed class DriverHostActorDbLessTests : RuntimeActorTestBase
     [Fact]
     public void DbLessNode_WithNoStoreAndNoDbFactory_SkipsScriptedAlarmHost()
     {
-        var publish = CreateTestProbe();
+        var publish = this.CreateCleanOutcomePublishProbe();
         var actor = Sys.ActorOf(DriverHostActor.Props(
             dbFactory: null,
             TestNode,
@@ -185,7 +185,7 @@ public sealed class DriverHostActorDbLessTests : RuntimeActorTestBase
     [Fact]
     public void DbBackedNode_WithNoStore_SkipsScriptedAlarmHost()
     {
-        var publish = CreateTestProbe();
+        var publish = this.CreateCleanOutcomePublishProbe();
         var actor = Sys.ActorOf(DriverHostActor.Props(
             dbFactory: NewInMemoryDbFactory(),
             TestNode,

tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverHostActorDegradedAckTests.cs

@@ -0,0 +1,203 @@
+using Akka.Actor;
+using Microsoft.EntityFrameworkCore;
+using Microsoft.Extensions.Logging.Abstractions;
+using Shouldly;
+using Xunit;
+using ZB.MOM.WW.OtOpcUa.Commons.Messages.Deploy;
+using ZB.MOM.WW.OtOpcUa.Commons.OpcUa;
+using ZB.MOM.WW.OtOpcUa.Commons.Types;
+using ZB.MOM.WW.OtOpcUa.Configuration;
+using ZB.MOM.WW.OtOpcUa.OpcUaServer;
+using ZB.MOM.WW.OtOpcUa.Runtime.Drivers;
+using ZB.MOM.WW.OtOpcUa.Runtime.OpcUa;
+using ZB.MOM.WW.OtOpcUa.Runtime.Tests.Harness;
+
+namespace ZB.MOM.WW.OtOpcUa.Runtime.Tests.Drivers;
+
+/// <summary>
+///     Gitea #601 (archreview 01/S-9) — the deploy ack must be honest about the address-space apply. It used
+///     to be sent the moment the revision advanced, before <c>OpcUaPublishActor</c> had even dequeued the
+///     rebuild, so a degraded apply (swallowed sink faults; <c>_lastApplied</c> deliberately not advanced)
+///     still acked a clean <see cref="ApplyAckOutcome.Applied"/> and the deployment sealed green fleet-wide.
+///     Now the Applied ack waits for the publish actor's <see cref="OpcUaPublishActor.RebuildCompleted"/> and
+///     carries its <see cref="ApplyAck.Degraded"/> verdict; a missing verdict times out to degraded, never clean.
+/// </summary>
+[Collection("opcua-publish-meters")]
+public sealed class DriverHostActorDegradedAckTests : RuntimeActorTestBase
+{
+    private static readonly NodeId TestNode = NodeId.Parse("driver-test");
+    private static readonly RevisionHash RevA = RevisionHash.Parse(new string('a', 64));
+
+    [Fact]
+    public void The_Applied_ack_waits_for_the_address_space_outcome_and_a_clean_one_acks_clean()
+    {
+        var (db, deploymentId) = SeedOneTagDeployment();
+        var coordinator = CreateTestProbe();
+        var publish = CreateTestProbe();
+        var actor = Sys.ActorOf(DriverHostActor.Props(
+            db, TestNode, coordinator.Ref, localRoles: new HashSet<string> { "driver" },
+            opcUaPublishActor: publish.Ref));
+
+        actor.Tell(new DispatchDeployment(deploymentId, RevA, CorrelationId.NewId()));
+
+        var rebuild = publish.FishForMessage<OpcUaPublishActor.RebuildAddressSpace>(_ => true, PresenceBudget);
+        rebuild.ReportOutcome.ShouldBeTrue();
+        // The ack is HELD until the outcome lands — the pre-#601 code had already sent it by now.
+        coordinator.ExpectNoMsg(TimeSpan.FromMilliseconds(300));
+
+        publish.Reply(new OpcUaPublishActor.RebuildCompleted(rebuild.Correlation, rebuild.DeploymentId, false, null));
+
+        var ack = coordinator.ExpectMsg<ApplyAck>(PresenceBudget);
+        ack.Outcome.ShouldBe(ApplyAckOutcome.Applied);
+        ack.Degraded.ShouldBeFalse();
+        ack.DegradedReason.ShouldBeNull();
+    }
+
+    [Fact]
+    public void A_degraded_address_space_outcome_acks_Applied_but_degraded_with_the_reason()
+    {
+        var (db, deploymentId) = SeedOneTagDeployment();
+        var coordinator = CreateTestProbe();
+        var publish = CreateTestProbe();
+        var actor = Sys.ActorOf(DriverHostActor.Props(
+            db, TestNode, coordinator.Ref, localRoles: new HashSet<string> { "driver" },
+            opcUaPublishActor: publish.Ref));
+
+        actor.Tell(new DispatchDeployment(deploymentId, RevA, CorrelationId.NewId()));
+        var rebuild = publish.FishForMessage<OpcUaPublishActor.RebuildAddressSpace>(_ => true, PresenceBudget);
+        publish.Reply(new OpcUaPublishActor.RebuildCompleted(
+            rebuild.Correlation, rebuild.DeploymentId, Degraded: true, DegradedReason: "failedNodes=2"));
+
+        var ack = coordinator.ExpectMsg<ApplyAck>(PresenceBudget);
+        ack.Outcome.ShouldBe(ApplyAckOutcome.Applied);
+        ack.Degraded.ShouldBeTrue();
+        ack.DegradedReason.ShouldBe("failedNodes=2");
+        ack.DeploymentId.ShouldBe(deploymentId);
+    }
+
+    [Fact]
+    public void A_real_publish_actor_whose_sink_faults_produces_a_degraded_ack_end_to_end()
+    {
+        // No hand-written verdict: the real OpcUaPublishActor + AddressSpaceApplier over a sink that faults on
+        // every node. This is the production shape of the defect — a degraded apply that sealed green.
+        var (db, deploymentId) = SeedOneTagDeployment();
+        var coordinator = CreateTestProbe();
+        var sink = new FaultingSink();
+        var publish = Sys.ActorOf(OpcUaPublishActor.PropsForTests(
+            sink: sink, applier: new AddressSpaceApplier(sink, NullLogger<AddressSpaceApplier>.Instance)));
+        var actor = Sys.ActorOf(DriverHostActor.Props(
+            db, TestNode, coordinator.Ref, localRoles: new HashSet<string> { "driver" },
+            opcUaPublishActor: publish));
+
+        actor.Tell(new DispatchDeployment(deploymentId, RevA, CorrelationId.NewId()));
+
+        var ack = coordinator.ExpectMsg<ApplyAck>(PresenceBudget);
+        ack.Outcome.ShouldBe(ApplyAckOutcome.Applied);
+        ack.Degraded.ShouldBeTrue();
+        ack.DegradedReason.ShouldNotBeNull();
+        ack.DegradedReason.ShouldContain("failedNodes=");
+    }
+
+    [Fact]
+    public void A_real_publish_actor_with_a_healthy_sink_produces_a_clean_ack_end_to_end()
+    {
+        // The control for the test above: same wiring, a sink that does not fault ⇒ clean.
+        var (db, deploymentId) = SeedOneTagDeployment();
+        var coordinator = CreateTestProbe();
+        var sink = new FaultingSink(fault: false);
+        var publish = Sys.ActorOf(OpcUaPublishActor.PropsForTests(
+            sink: sink, applier: new AddressSpaceApplier(sink, NullLogger<AddressSpaceApplier>.Instance)));
+        var actor = Sys.ActorOf(DriverHostActor.Props(
+            db, TestNode, coordinator.Ref, localRoles: new HashSet<string> { "driver" },
+            opcUaPublishActor: publish));
+
+        actor.Tell(new DispatchDeployment(deploymentId, RevA, CorrelationId.NewId()));
+
+        var ack = coordinator.ExpectMsg<ApplyAck>(PresenceBudget);
+        ack.Outcome.ShouldBe(ApplyAckOutcome.Applied);
+        ack.Degraded.ShouldBeFalse();
+    }
+
+    [Fact]
+    public void No_outcome_within_the_timeout_acks_degraded_never_clean()
+    {
+        var (db, deploymentId) = SeedOneTagDeployment();
+        var coordinator = CreateTestProbe();
+        var publish = CreateTestProbe(); // never replies — a hung / lost apply
+        var actor = Sys.ActorOf(DriverHostActor.Props(
+            db, TestNode, coordinator.Ref, localRoles: new HashSet<string> { "driver" },
+            opcUaPublishActor: publish.Ref, applyOutcomeTimeout: TimeSpan.FromMilliseconds(300)));
+
+        actor.Tell(new DispatchDeployment(deploymentId, RevA, CorrelationId.NewId()));
+
+        var ack = coordinator.ExpectMsg<ApplyAck>(PresenceBudget);
+        ack.Outcome.ShouldBe(ApplyAckOutcome.Applied);
+        ack.Degraded.ShouldBeTrue();
+        ack.DegradedReason.ShouldNotBeNull();
+        ack.DegradedReason.ShouldContain("did not report an outcome");
+
+        // A late outcome after the timeout must not produce a second ack.
+        var rebuild = publish.FishForMessage<OpcUaPublishActor.RebuildAddressSpace>(_ => true, PresenceBudget);
+        publish.Reply(new OpcUaPublishActor.RebuildCompleted(rebuild.Correlation, rebuild.DeploymentId, false, null));
+        coordinator.ExpectNoMsg(TimeSpan.FromMilliseconds(300));
+    }
+
+    [Fact]
+    public void Redispatching_a_degraded_revision_reapplies_instead_of_acking_clean()
+    {
+        var (db, deploymentId) = SeedOneTagDeployment();
+        var coordinator = CreateTestProbe();
+        var publish = CreateTestProbe();
+        var actor = Sys.ActorOf(DriverHostActor.Props(
+            db, TestNode, coordinator.Ref, localRoles: new HashSet<string> { "driver" },
+            opcUaPublishActor: publish.Ref));
+
+        actor.Tell(new DispatchDeployment(deploymentId, RevA, CorrelationId.NewId()));
+        var first = publish.FishForMessage<OpcUaPublishActor.RebuildAddressSpace>(_ => true, PresenceBudget);
+        publish.Reply(new OpcUaPublishActor.RebuildCompleted(first.Correlation, first.DeploymentId, true, "failedNodes=1"));
+        coordinator.ExpectMsg<ApplyAck>(PresenceBudget).Degraded.ShouldBeTrue();
+
+        // Same revision again: the old idempotent short-circuit would ack a CLEAN Applied right here. It must
+        // re-apply instead (a second rebuild request), and ack whatever THAT apply reports.
+        actor.Tell(new DispatchDeployment(deploymentId, RevA, CorrelationId.NewId()));
+        var second = publish.FishForMessage<OpcUaPublishActor.RebuildAddressSpace>(_ => true, PresenceBudget);
+        coordinator.ExpectNoMsg(TimeSpan.FromMilliseconds(200));
+        publish.Reply(new OpcUaPublishActor.RebuildCompleted(second.Correlation, second.DeploymentId, false, null));
+        coordinator.ExpectMsg<ApplyAck>(PresenceBudget).Degraded.ShouldBeFalse();
+
+        // Now clean: a third dispatch of the same revision takes the short-circuit again (no rebuild).
+        actor.Tell(new DispatchDeployment(deploymentId, RevA, CorrelationId.NewId()));
+        var third = coordinator.ExpectMsg<ApplyAck>(PresenceBudget);
+        third.Degraded.ShouldBeFalse();
+        publish.ExpectNoMsg(TimeSpan.FromMilliseconds(200));
+    }
+
+    private (IDbContextFactory<OtOpcUaConfigDbContext> Db, DeploymentId Id) SeedOneTagDeployment()
+    {
+        var db = NewInMemoryDbFactory();
+        var id = V3ArtifactSeeder.SeedV3Deployment(db, RevA,
+            rawTags: new[] { (Driver: "drv-1", DriverName: "Modbus", Device: "dev1", Tag: "speed", DataType: "Double", Writable: false) },
+            unsRefs: new[] { (Area: "filling", Line: "line1", Equip: "station1", BackingTag: "speed", Effective: (string?)null) });
+        return (db, id);
+    }
+
+    /// <summary>A sink that throws from every node-creating call when <c>fault</c> is set.</summary>
+    private sealed class FaultingSink(bool fault = true) : IOpcUaAddressSpaceSink
+    {
+        private void MaybeFault()
+        {
+            if (fault) throw new InvalidOperationException("simulated sink fault");
+        }
+
+        public void WriteValue(string nodeId, object? value, uint statusCode, DateTime sourceTimestampUtc, AddressSpaceRealm realm = AddressSpaceRealm.Uns) { }
+        public void WriteAlarmQuality(string alarmNodeId, uint statusCode, DateTime sourceTimestampUtc, AddressSpaceRealm realm) { }
+        public void WriteAlarmCondition(string alarmNodeId, AlarmConditionSnapshot state, DateTime occurredUtc, AddressSpaceRealm realm = AddressSpaceRealm.Uns) { }
+        public void MaterialiseAlarmCondition(string alarmNodeId, string equipmentNodeId, string displayName, string alarmType, int severity, AddressSpaceRealm realm, bool isNative = false) => MaybeFault();
+        public void EnsureFolder(string folderNodeId, string? parentNodeId, string displayName, AddressSpaceRealm realm = AddressSpaceRealm.Uns) => MaybeFault();
+        public void EnsureVariable(string variableNodeId, string? parentFolderNodeId, string displayName, string dataType, bool writable, AddressSpaceRealm realm, string? historianTagname = null, bool isArray = false, uint? arrayLength = null, bool refuseArrayOnScalar = false) => MaybeFault();
+        public void RebuildAddressSpace() { }
+        public void RaiseNodesAddedModelChange(string affectedNodeId, AddressSpaceRealm realm = AddressSpaceRealm.Uns) { }
+        public void WireAlarmNotifiers(string alarmNodeId, AddressSpaceRealm alarmRealm, IReadOnlyList<string> notifierFolderNodeIds, AddressSpaceRealm notifierFolderRealm) { }
+        public void AddReference(string sourceNodeId, AddressSpaceRealm sourceRealm, string targetNodeId, AddressSpaceRealm targetRealm, string referenceType = "Organizes") { }
+    }
+}

tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverHostActorFetchAndCacheBootstrapTests.cs

@@ -42,7 +42,7 @@ public sealed class DriverHostActorFetchAndCacheBootstrapTests : RuntimeActorTes
         var cached = new CachedDeploymentArtifact(
             DeploymentId.NewId().ToString(), CachedRev.Value, blob, DateTimeOffset.UtcNow.AddHours(-1));
         var cache = new StubArtifactCache(cached);
-        var publishProbe = CreateTestProbe();
+        var publishProbe = this.CreateCleanOutcomePublishProbe();
 
         var actor = Sys.ActorOf(DriverHostActor.Props(
             new ThrowingDbFactory(), TestNode, ackRouter: null,

tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverHostActorFetchAndCacheUnreadableTests.cs

@@ -42,7 +42,7 @@ public sealed class DriverHostActorFetchAndCacheUnreadableTests : RuntimeActorTe
             [RevA.Value] = Artifact(),
             [RevB.Value] = null,
         });
-        var publishProbe = CreateTestProbe();
+        var publishProbe = this.CreateCleanOutcomePublishProbe();
         var coordinator = CreateTestProbe();
 
         var actor = Sys.ActorOf(DriverHostActor.Props(
@@ -72,7 +72,7 @@ public sealed class DriverHostActorFetchAndCacheUnreadableTests : RuntimeActorTe
         // empty blob handed to it directly: ReconcileDriversFromBlob's #485 guard fails the apply.
         var db = NewInMemoryDbFactory();
         var fetcher = new MapFetcher(new Dictionary<string, byte[]?> { [RevA.Value] = Array.Empty<byte>() });
-        var publishProbe = CreateTestProbe();
+        var publishProbe = this.CreateCleanOutcomePublishProbe();
         var coordinator = CreateTestProbe();
 
         var actor = Sys.ActorOf(DriverHostActor.Props(
@@ -95,7 +95,7 @@ public sealed class DriverHostActorFetchAndCacheUnreadableTests : RuntimeActorTe
         // A truncated/corrupt cached artifact surfaces as GetCurrentUnkeyedAsync == null (the cache's
         // integrity check is deliberately indistinguishable from a miss). Boot must land
         // Steady-no-revision and apply NOTHING — never a partial address space.
-        var publishProbe = CreateTestProbe();
+        var publishProbe = this.CreateCleanOutcomePublishProbe();
 
         var actor = Sys.ActorOf(DriverHostActor.Props(
             new ThrowingDbFactory(), TestNode, ackRouter: null,

tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverHostActorHistoryWriterTests.cs

@@ -53,7 +53,7 @@ public sealed class DriverHostActorHistoryWriterTests : RuntimeActorTestBase
         // A publish actor MUST be wired for SpawnVirtualTagHost to spawn the real host (its ctor
         // requires a non-null sink). No virtualTagHostOverride ⇒ the real VirtualTagHostActor is
         // spawned with _historyWriter threaded in.
-        var publish = CreateTestProbe();
+        var publish = this.CreateCleanOutcomePublishProbe();
         var writer = new CapturingHistoryWriter();
 
         var actor = Sys.ActorOf(DriverHostActor.Props(

tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverHostActorNativeAlarmTests.cs

@@ -441,7 +441,7 @@ public sealed class DriverHostActorNativeAlarmTests : RuntimeActorTestBase
         IDbContextFactory<OtOpcUaConfigDbContext> db, DeploymentId deploymentId)
     {
         var coordinator = CreateTestProbe();
-        var publish = CreateTestProbe();
+        var publish = this.CreateCleanOutcomePublishProbe();
         var mux = CreateTestProbe();
         var vtHost = CreateTestProbe();
 

tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverHostActorPrimaryGateTests.cs

@@ -251,7 +251,7 @@ public sealed class DriverHostActorPrimaryGateTests : RuntimeActorTestBase
         IDbContextFactory<OtOpcUaConfigDbContext> db, DeploymentId dep, int driverMemberCount)
     {
         var coordinator = CreateTestProbe();
-        var publish = CreateTestProbe();
+        var publish = this.CreateCleanOutcomePublishProbe();
         var mux = CreateTestProbe();
         var vtHost = CreateTestProbe();
 

tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverHostActorTests.cs

@@ -226,7 +226,7 @@ public sealed class DriverHostActorTests : RuntimeActorTestBase
         seeded.Length.ShouldBeGreaterThan(0);
 
         var coordinator = CreateTestProbe();
-        var publish = CreateTestProbe();
+        var publish = this.CreateCleanOutcomePublishProbe();
         var actor = Sys.ActorOf(DriverHostActor.Props(
             db, TestNode, coordinator.Ref,
             localRoles: new HashSet<string> { "driver" },

tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Harness/PublishProbes.cs

@@ -0,0 +1,34 @@
+using Akka.Actor;
+using Akka.TestKit;
+using ZB.MOM.WW.OtOpcUa.Runtime.OpcUa;
+
+namespace ZB.MOM.WW.OtOpcUa.Runtime.Tests.Harness;
+
+/// <summary>
+///     Gitea #601 — since the deploy ack waits for <see cref="OpcUaPublishActor"/>'s
+///     <see cref="OpcUaPublishActor.RebuildCompleted"/> reply, a bare <see cref="TestProbe"/> standing in for the
+///     publish actor would hold every Applied ack until the outcome timeout (and then ack it degraded). Tests
+///     that are not ABOUT the ack's honesty use this probe instead: it still records every message exactly
+///     like a plain probe (the auto-pilot runs in addition to queueing), and answers each
+///     <see cref="OpcUaPublishActor.RebuildAddressSpace"/> that asks for an outcome with a CLEAN one — the
+///     reply the real publish actor gives a clean apply.
+/// </summary>
+public static class PublishProbes
+{
+    /// <summary>Creates a publish-actor probe that reports a clean address-space outcome for every rebuild
+    /// carrying <see cref="OpcUaPublishActor.RebuildAddressSpace.ReportOutcome"/>.</summary>
+    /// <param name="kit">The calling TestKit.</param>
+    /// <returns>The probe.</returns>
+    public static TestProbe CreateCleanOutcomePublishProbe(this TestKitBase kit)
+    {
+        var probe = kit.CreateTestProbe();
+        probe.SetAutoPilot(new DelegateAutoPilot((sender, message) =>
+        {
+            if (message is OpcUaPublishActor.RebuildAddressSpace { ReportOutcome: true } rebuild)
+                sender.Tell(new OpcUaPublishActor.RebuildCompleted(
+                    rebuild.Correlation, rebuild.DeploymentId, Degraded: false, DegradedReason: null));
+            return AutoPilot.KeepRunning;
+        }));
+        return probe;
+    }
+}

tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Harness/V3ArtifactSeeder.cs

@@ -151,7 +151,7 @@ public static class V3ArtifactSeeder
         TimeSpan? reprovisionBackoff = null,
         TimeSpan? commsRecoveryAuditDelay = null)
     {
-        var publish = kit.CreateTestProbe();
+        var publish = kit.CreateCleanOutcomePublishProbe();
         var (actor, mux, coordinator) =
             Spawn(kit, db, localNode, publish.Ref, driverFactory, reprovisionBackoff, commsRecoveryAuditDelay);
 

tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Telemetry/PublishSeamEmitsToHubTests.cs

@@ -232,7 +232,7 @@ public sealed class PublishSeamEmitsToHubTests : RuntimeActorTestBase
         IDbContextFactory<OtOpcUaConfigDbContext> db, DeploymentId deploymentId, ITelemetryLocalHub hub)
     {
         var coordinator = CreateTestProbe();
-        var publish = CreateTestProbe();
+        var publish = this.CreateCleanOutcomePublishProbe();
         var mux = CreateTestProbe();
         var vtHost = CreateTestProbe();