Merge b640-601: honest deploy acks — Applied only after the address-space outcome, degraded flag + reason (#601)

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

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

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

@@ -84,6 +84,25 @@ 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>
+    ///     Derived from central's apply deadline (<see cref="DeployTimings.ApplyDeadline"/>, 2 min — the same
+    ///     constant <c>ConfigPublishCoordinator.DefaultApplyDeadline</c> reads) minus
+    ///     <see cref="DeployTimings.ApplyOutcomeSafetyMargin"/> (15 s) = 105 s: long enough that a
+    ///     slow-but-clean rebuild of a large address space is not labelled degraded, short enough that the ack
+    ///     still lands before central times the deployment out. An ack sent on this timeout is marked
+    ///     <b>degraded</b> with a reason starting <see cref="OutcomeNotReportedReasonPrefix"/>, never clean —
+    ///     the revision did advance, but nobody has confirmed the address space matches it.
+    /// </remarks>
+    public static readonly TimeSpan DefaultApplyOutcomeTimeout = DeployTimings.ApplyOutcomeTimeout;
+
+    /// <summary>Gitea #601 — the reason prefix of a degraded ack sent because the outcome TIMED OUT, so an
+    /// operator can tell "not reported in time" from a real sink failure.</summary>
+    public const string OutcomeNotReportedReasonPrefix = "address-space apply outcome not reported in time";
+
     /// <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 +263,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 +465,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 +633,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 +667,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 +681,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 +779,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 +1332,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 +1396,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);
@@ -2180,6 +2230,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>(_ =>
@@ -2226,7 +2280,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",
@@ -2235,6 +2294,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);
@@ -2394,7 +2458,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
@@ -2406,8 +2476,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);
@@ -2427,6 +2500,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"));
@@ -2447,6 +2523,99 @@ 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,
+            $"{OutcomeNotReportedReasonPrefix} ({_applyOutcomeTimeout.TotalSeconds:0}s) — no sink failure was observed; the apply may still be running (slow) or its outcome was 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)
+        {
+            // #601 review — the apply wrote this node's own row Applied/null before the outcome was known. Central
+            // adds the reason from the ack only when it ACCEPTS the ack (a timed-out or superseded deployment's ack
+            // is discarded), so a node with a ConfigDb records it itself; a driver-only node no-ops here.
+            UpsertNodeDeploymentState(pending.DeploymentId, NodeDeploymentStatus.Applied,
+                ApplyAck.DegradedReasonPrefix + (reason ?? "address-space apply 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.
@@ -3677,9 +3846,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