using Akka.Actor;
using Akka.TestKit;
using Akka.TestKit.Xunit2;
using Microsoft.Extensions.Logging.Abstractions;
using ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Protocol;
using ZB.MOM.WW.ScadaBridge.Commons.Messages.DataConnection;
using ZB.MOM.WW.ScadaBridge.Commons.Messages.Instance;
using ZB.MOM.WW.ScadaBridge.Commons.Messages.Streaming;
using ZB.MOM.WW.ScadaBridge.Commons.Types.Flattening;
using ZB.MOM.WW.ScadaBridge.SiteRuntime.Actors;
using ZB.MOM.WW.ScadaBridge.SiteRuntime.Persistence;
using ZB.MOM.WW.ScadaBridge.SiteRuntime.Scripts;
using System.Text.Json;

namespace ZB.MOM.WW.ScadaBridge.SiteRuntime.Tests.Actors;

/// <summary>
/// Tests for the event-driven <c>WaitForAttribute</c> one-shot waiter registry in
/// <see cref="InstanceActor"/> (Attributes.WaitAsync spec §3-§5). Covers the
/// fast-path, change-match, timeout, no-leak (timeout-canceled-on-match), and
/// predicate-overload acceptance criteria.
/// </summary>
public class InstanceActorWaitForAttributeTests : TestKit, IDisposable
{
    private readonly SiteStorageService _storage;
    private readonly ScriptCompilationService _compilationService;
    private readonly SharedScriptLibrary _sharedScriptLibrary;
    private readonly SiteRuntimeOptions _options;
    private readonly string _dbFile;

    public InstanceActorWaitForAttributeTests()
    {
        _dbFile = Path.Combine(Path.GetTempPath(), $"instance-waitfor-test-{Guid.NewGuid():N}.db");
        _storage = new SiteStorageService(
            $"Data Source={_dbFile}",
            NullLogger<SiteStorageService>.Instance);
        _storage.InitializeAsync().GetAwaiter().GetResult();
        _compilationService = new ScriptCompilationService(
            NullLogger<ScriptCompilationService>.Instance);
        _sharedScriptLibrary = new SharedScriptLibrary(
            _compilationService, NullLogger<SharedScriptLibrary>.Instance);
        _options = new SiteRuntimeOptions();
    }

    private IActorRef CreateInstanceActor(string instanceName, FlattenedConfiguration config)
    {
        return ActorOf(Props.Create(() => new InstanceActor(
            instanceName,
            JsonSerializer.Serialize(config),
            _storage,
            _compilationService,
            _sharedScriptLibrary,
            null, // no stream manager in tests
            _options,
            NullLogger<InstanceActor>.Instance)));
    }

    void IDisposable.Dispose()
    {
        Shutdown();
        try { File.Delete(_dbFile); } catch { /* cleanup */ }
    }

    // ── 1. Fast-path: attribute already at target ────────────────────────────

    /// <summary>
    /// Acceptance §7.1: when the attribute already equals the target at the time
    /// the waiter registers, the actor must reply immediately with Matched=true
    /// (carrying the current value), without scheduling a timeout.
    /// </summary>
    [Fact]
    public void WaitForAttribute_FastPath_AlreadyAtTarget_RepliesMatchedImmediately()
    {
        var config = new FlattenedConfiguration
        {
            InstanceUniqueName = "Pump1",
            Attributes =
            [
                new ResolvedAttribute { CanonicalName = "Flag", Value = "true", DataType = "Boolean" }
            ]
        };

        var actor = CreateInstanceActor("Pump1", config);

        actor.Tell(new WaitForAttributeRequest(
            "wfa-fast", "Pump1", "Flag",
            "true", null, TimeSpan.FromSeconds(30), DateTimeOffset.UtcNow));

        var response = ExpectMsg<WaitForAttributeResponse>(TimeSpan.FromSeconds(5));
        Assert.True(response.Matched);
        Assert.False(response.TimedOut);
        Assert.Equal("wfa-fast", response.CorrelationId);
        Assert.Equal("true", response.Value?.ToString());
    }

    // ── 2. Change-match: register first, then drive a value change ───────────

    /// <summary>
    /// Acceptance §7.1/§7.4: registering when the value does NOT match, then
    /// driving the attribute to the target value (via a DCL TagValueUpdate) must
    /// produce a single Matched=true reply carrying the new value.
    /// </summary>
    [Fact]
    public void WaitForAttribute_ChangeMatch_RepliesMatchedWithNewValue()
    {
        const string tag = "ns=3;s=Recipe.Processed";
        var config = new FlattenedConfiguration
        {
            InstanceUniqueName = "Pump1",
            Attributes =
            [
                new ResolvedAttribute
                {
                    CanonicalName = "Processed", Value = "false", DataType = "Boolean",
                    DataSourceReference = tag, BoundDataConnectionName = "PLC"
                }
            ]
        };

        var dcl = CreateTestProbe();
        var actor = ActorOf(Props.Create(() => new InstanceActor(
            "Pump1",
            JsonSerializer.Serialize(config),
            _storage,
            _compilationService,
            _sharedScriptLibrary,
            null,
            _options,
            NullLogger<InstanceActor>.Instance,
            dcl.Ref)));

        dcl.ExpectMsg<SubscribeTagsRequest>(TimeSpan.FromSeconds(5));

        // Register: current value "false" does not match the target. The value
        // arrives from the DCL as a boolean true, whose codec-encoded form is
        // "True" — so the target must be encoded the same way the accessor would
        // (AttributeValueCodec.Encode(true)), NOT the literal string "true".
        var target = ZB.MOM.WW.ScadaBridge.Commons.Types.AttributeValueCodec.Encode(true);
        actor.Tell(new WaitForAttributeRequest(
            "wfa-change", "Pump1", "Processed",
            target, null, TimeSpan.FromSeconds(30), DateTimeOffset.UtcNow));

        // No reply yet — the value has not changed to the target.
        ExpectNoMsg(TimeSpan.FromMilliseconds(300));

        // Drive the value to the target through the DCL ingest path.
        actor.Tell(new TagValueUpdate("PLC", tag, true, QualityCode.Good, DateTimeOffset.UtcNow));

        var response = ExpectMsg<WaitForAttributeResponse>(TimeSpan.FromSeconds(5));
        Assert.True(response.Matched);
        Assert.False(response.TimedOut);
        Assert.Equal("wfa-change", response.CorrelationId);
        Assert.Equal(true, response.Value);
        Assert.Equal("Good", response.Quality);
    }

    // ── 3. Timeout: value never matches ──────────────────────────────────────

    /// <summary>
    /// Acceptance §7.2: when the attribute never reaches the target within the
    /// timeout, the actor replies Matched=false, TimedOut=true (no throw).
    /// </summary>
    [Fact]
    public void WaitForAttribute_Timeout_RepliesNotMatchedTimedOut()
    {
        var config = new FlattenedConfiguration
        {
            InstanceUniqueName = "Pump1",
            Attributes =
            [
                new ResolvedAttribute { CanonicalName = "Flag", Value = "false", DataType = "Boolean" }
            ]
        };

        var actor = CreateInstanceActor("Pump1", config);

        actor.Tell(new WaitForAttributeRequest(
            "wfa-timeout", "Pump1", "Flag",
            "true", null, TimeSpan.FromMilliseconds(300), DateTimeOffset.UtcNow));

        // The scheduled timeout fires; allow a tolerant deadline.
        var response = ExpectMsg<WaitForAttributeResponse>(TimeSpan.FromSeconds(3));
        Assert.False(response.Matched);
        Assert.True(response.TimedOut);
        Assert.Equal("wfa-timeout", response.CorrelationId);
    }

    // ── 4. No-leak: timeout canceled on match (no second reply) ──────────────

    /// <summary>
    /// Acceptance §7.5: after a successful change-match, the scheduled timeout
    /// must have been canceled and the waiter removed — so NO second (timeout)
    /// response arrives after the match.
    /// </summary>
    [Fact]
    public void WaitForAttribute_Match_CancelsTimeout_NoSecondReply()
    {
        var config = new FlattenedConfiguration
        {
            InstanceUniqueName = "Pump1",
            Attributes =
            [
                new ResolvedAttribute { CanonicalName = "Flag", Value = "false", DataType = "Boolean" }
            ]
        };

        var actor = CreateInstanceActor("Pump1", config);

        // Register with a short timeout, then match BEFORE it would fire.
        actor.Tell(new WaitForAttributeRequest(
            "wfa-noleak", "Pump1", "Flag",
            "true", null, TimeSpan.FromMilliseconds(500), DateTimeOffset.UtcNow));

        // Drive the static value to the target; the actor publishes via
        // HandleAttributeValueChanged, satisfying the waiter.
        actor.Tell(new SetStaticAttributeCommand(
            "set-flag", "Pump1", "Flag", "true", DateTimeOffset.UtcNow));

        // First reply: the match. (A SetStaticAttributeResponse also arrives for
        // the set command — filter for the WaitForAttributeResponse.)
        var matched = ExpectMsg<WaitForAttributeResponse>(TimeSpan.FromSeconds(5));
        Assert.True(matched.Matched);
        Assert.False(matched.TimedOut);

        // The set command's own ack — drain it so the no-msg assert below is clean.
        ExpectMsg<SetStaticAttributeResponse>(TimeSpan.FromSeconds(5));

        // No second WaitForAttributeResponse (the timeout was canceled) for longer
        // than the original 500ms timeout window.
        ExpectNoMsg(TimeSpan.FromSeconds(1));
    }

    // ── 5. Predicate overload ────────────────────────────────────────────────

    /// <summary>
    /// Acceptance §7 (predicate form): registering with a site-local predicate and
    /// then flipping the value so the predicate passes must produce Matched=true.
    /// </summary>
    [Fact]
    public void WaitForAttribute_PredicateOverload_MatchesOnPredicatePass()
    {
        const string tag = "ns=3;s=Level";
        var config = new FlattenedConfiguration
        {
            InstanceUniqueName = "Pump1",
            Attributes =
            [
                new ResolvedAttribute
                {
                    CanonicalName = "Level", Value = "0", DataType = "Int32",
                    DataSourceReference = tag, BoundDataConnectionName = "PLC"
                }
            ]
        };

        var dcl = CreateTestProbe();
        var actor = ActorOf(Props.Create(() => new InstanceActor(
            "Pump1",
            JsonSerializer.Serialize(config),
            _storage,
            _compilationService,
            _sharedScriptLibrary,
            null,
            _options,
            NullLogger<InstanceActor>.Instance,
            dcl.Ref)));

        dcl.ExpectMsg<SubscribeTagsRequest>(TimeSpan.FromSeconds(5));

        // Predicate: value > 50 (current is 0, so no immediate match).
        Func<object?, bool> predicate = v =>
            v is not null && int.TryParse(v.ToString(), out var n) && n > 50;

        actor.Tell(new WaitForAttributeRequest(
            "wfa-pred", "Pump1", "Level",
            null, predicate, TimeSpan.FromSeconds(30), DateTimeOffset.UtcNow));

        ExpectNoMsg(TimeSpan.FromMilliseconds(300));

        // A value below the threshold must NOT satisfy the predicate.
        actor.Tell(new TagValueUpdate("PLC", tag, 25, QualityCode.Good, DateTimeOffset.UtcNow));
        ExpectNoMsg(TimeSpan.FromMilliseconds(300));

        // A value above the threshold satisfies it.
        actor.Tell(new TagValueUpdate("PLC", tag, 75, QualityCode.Good, DateTimeOffset.UtcNow));

        var response = ExpectMsg<WaitForAttributeResponse>(TimeSpan.FromSeconds(5));
        Assert.True(response.Matched);
        Assert.False(response.TimedOut);
        Assert.Equal(75, response.Value);
    }

    // ── 6. "any change" (null target + null predicate) ───────────────────────

    /// <summary>
    /// Spec §4.1: a null TargetValueEncoded + null Predicate means "wait for any
    /// change" (test <c>_ => true</c>). When the attribute ALREADY holds a value at
    /// registration, the fast-path matches IMMEDIATELY — there is no need to wait for
    /// a subsequent update. (A separate test covers the absent-at-registration case.)
    /// </summary>
    [Fact]
    public void WaitForAttribute_AnyChange_MatchesImmediatelyWhenAttributePresent()
    {
        const string tag = "ns=3;s=Speed";
        var config = new FlattenedConfiguration
        {
            InstanceUniqueName = "Pump1",
            Attributes =
            [
                new ResolvedAttribute
                {
                    CanonicalName = "Speed", Value = "0", DataType = "Int32",
                    DataSourceReference = tag, BoundDataConnectionName = "PLC"
                }
            ]
        };

        var dcl = CreateTestProbe();
        var actor = ActorOf(Props.Create(() => new InstanceActor(
            "Pump1",
            JsonSerializer.Serialize(config),
            _storage,
            _compilationService,
            _sharedScriptLibrary,
            null,
            _options,
            NullLogger<InstanceActor>.Instance,
            dcl.Ref)));

        dcl.ExpectMsg<SubscribeTagsRequest>(TimeSpan.FromSeconds(5));

        // "any change" registers with a non-trivial timeout. The fast-path uses
        // `_ => true`, so a currently-present attribute matches immediately.
        actor.Tell(new WaitForAttributeRequest(
            "wfa-any", "Pump1", "Speed",
            null, null, TimeSpan.FromSeconds(30), DateTimeOffset.UtcNow));

        // Speed=0 is already present, so the "any change" test (_ => true) matches
        // immediately on the fast path.
        var response = ExpectMsg<WaitForAttributeResponse>(TimeSpan.FromSeconds(5));
        Assert.True(response.Matched);
        Assert.False(response.TimedOut);
    }

    /// <summary>
    /// Spec §4.1 (companion to the immediate-match case): when the attribute is
    /// ABSENT at registration (no entry in <c>_attributes</c>), the "any change"
    /// waiter does NOT fast-path — it registers, and a later value update on that
    /// attribute is the first thing that satisfies it.
    /// </summary>
    [Fact]
    public void WaitForAttribute_AnyChange_AttributeAbsent_MatchesOnLaterSet()
    {
        var config = new FlattenedConfiguration
        {
            InstanceUniqueName = "Pump1",
            Attributes =
            [
                new ResolvedAttribute { CanonicalName = "Known", Value = "x", DataType = "String" }
            ]
        };

        var actor = CreateInstanceActor("Pump1", config);

        // "Ghost" is not a configured attribute, so _attributes has no entry — the
        // fast-path TryGetValue misses and the waiter registers rather than matching.
        actor.Tell(new WaitForAttributeRequest(
            "wfa-absent", "Pump1", "Ghost",
            null, null, TimeSpan.FromSeconds(30), DateTimeOffset.UtcNow));

        ExpectNoMsg(TimeSpan.FromMilliseconds(300));

        // A direct AttributeValueChanged for "Ghost" populates _attributes and
        // re-evaluates the waiter; the any-change test now matches the new value.
        actor.Tell(new AttributeValueChanged(
            "Pump1", "Ghost", "Ghost", "appeared", "Good", DateTimeOffset.UtcNow));

        var response = ExpectMsg<WaitForAttributeResponse>(TimeSpan.FromSeconds(5));
        Assert.True(response.Matched);
        Assert.False(response.TimedOut);
        Assert.Equal("wfa-absent", response.CorrelationId);
        Assert.Equal("appeared", response.Value);
    }

    // ── 7. CRITICAL 1: no spurious match on a quality-only republish ─────────

    /// <summary>
    /// CRITICAL 1 regression: the List-coerce-failure Bad-quality path republishes
    /// the OLD value (quality flipped to Bad) WITHOUT changing <c>_attributes</c>, so
    /// it passes <c>evaluateWaiters:false</c> — registered waiters are NOT re-evaluated
    /// on this non-change republish, must NOT spuriously fire, and must STILL resolve
    /// on the next genuine value change.
    ///
    /// <para>
    /// We register an "any-change" waiter (which correctly fast-path matches the
    /// present value and is drained) plus a pending predicate waiter that does not yet
    /// match, then drive the Bad-quality republish and assert NO match is delivered for
    /// the pending waiter, and that a subsequent REAL change resolves it. (Note: the
    /// purest "any-change fires on a non-change republish" symptom is not directly
    /// reproducible — an any-change waiter against a present attribute always fast-path
    /// matches and so never stays pending across a republish; this test guards the
    /// republish path against double-firing / stranding waiters and against the
    /// predicate being re-evaluated on the non-change republish.)
    /// </para>
    /// </summary>
    [Fact]
    public void WaitForAttribute_BadQualityRepublish_NoValueChange_DoesNotMatch()
    {
        const string tag = "ns=3;s=Items";
        var config = new FlattenedConfiguration
        {
            InstanceUniqueName = "Pump1",
            Attributes =
            [
                new ResolvedAttribute
                {
                    // Static default {1,2}: a real list value is present from
                    // construction so the Bad-quality republish has an OLD value to
                    // republish. The waiter below targets a DIFFERENT value so it is
                    // genuinely pending (no fast-path match) when the republish fires.
                    CanonicalName = "Items", Value = "[1,2]", DataType = "List",
                    ElementDataType = "Int32",
                    DataSourceReference = tag, BoundDataConnectionName = "PLC"
                }
            ]
        };

        var dcl = CreateTestProbe();
        var actor = ActorOf(Props.Create(() => new InstanceActor(
            "Pump1",
            JsonSerializer.Serialize(config),
            _storage,
            _compilationService,
            _sharedScriptLibrary,
            null,
            _options,
            NullLogger<InstanceActor>.Instance,
            dcl.Ref)));

        dcl.ExpectMsg<SubscribeTagsRequest>(TimeSpan.FromSeconds(5));

        // A predicate waiter that matches a list of length >= 3. Current value is
        // {1,2} (length 2) so it does NOT fast-path match — it registers and stays
        // pending. Crucially, the Bad-quality republish below carries the SAME OLD
        // value {1,2} (length 2); with the bug (evaluateWaiters always true) the
        // predicate would be re-evaluated against {1,2} → still false, so this probe
        // also guards the predicate-isolation contract on the republish path.
        Func<object?, bool> lenAtLeast3 = v =>
            v is System.Collections.IList list && list.Count >= 3;
        actor.Tell(new WaitForAttributeRequest(
            "wfa-len3", "Pump1", "Items",
            null, lenAtLeast3, TimeSpan.FromSeconds(30), DateTimeOffset.UtcNow));

        // Also register an "any-change" waiter while the attribute is present — it
        // fast-path matches the current {1,2} immediately. Drain that correct match;
        // it is the documented immediate-match behaviour, not the bug under test.
        actor.Tell(new WaitForAttributeRequest(
            "wfa-any", "Pump1", "Items",
            null, null, TimeSpan.FromSeconds(30), DateTimeOffset.UtcNow));
        var immediate = ExpectMsg<WaitForAttributeResponse>(TimeSpan.FromSeconds(5));
        Assert.Equal("wfa-any", immediate.CorrelationId);
        Assert.True(immediate.Matched);

        // Drive the List-coerce-FAILURE Bad-quality republish: a scalar int cannot
        // coerce to List<Int32>, so the actor sets quality Bad and republishes the
        // OLD value {1,2} WITHOUT changing _attributes (evaluateWaiters:false).
        actor.Tell(new TagValueUpdate("PLC", tag, 999, QualityCode.Good, DateTimeOffset.UtcNow));

        // The pending length>=3 waiter must NOT fire on this non-change republish.
        ExpectNoMsg(TimeSpan.FromMilliseconds(500));

        // A REAL change to a length-3 list resolves the still-pending waiter.
        actor.Tell(new TagValueUpdate("PLC", tag, new[] { 7, 8, 9 }, QualityCode.Good, DateTimeOffset.UtcNow));
        var realChange = ExpectMsg<WaitForAttributeResponse>(TimeSpan.FromSeconds(5));
        Assert.Equal("wfa-len3", realChange.CorrelationId);
        Assert.True(realChange.Matched);
        Assert.False(realChange.TimedOut);
    }

    // ── 8. CRITICAL 2: throwing predicate is isolated ────────────────────────

    /// <summary>
    /// CRITICAL 2 regression: two waiters on the SAME attribute — one with a
    /// predicate that throws, one a normal value-equality. A single value change
    /// must (a) NOT crash the actor, (b) evict the throwing waiter with a
    /// non-matched error reply, and (c) STILL resolve the normal sibling. Finally
    /// the actor must remain responsive to a subsequent request.
    /// </summary>
    [Fact]
    public void WaitForAttribute_ThrowingPredicate_IsIsolated_SiblingStillMatches()
    {
        const string tag = "ns=3;s=State";
        var config = new FlattenedConfiguration
        {
            InstanceUniqueName = "Pump1",
            Attributes =
            [
                new ResolvedAttribute
                {
                    CanonicalName = "State", Value = "init", DataType = "String",
                    DataSourceReference = tag, BoundDataConnectionName = "PLC"
                }
            ]
        };

        var dcl = CreateTestProbe();
        var actor = ActorOf(Props.Create(() => new InstanceActor(
            "Pump1",
            JsonSerializer.Serialize(config),
            _storage,
            _compilationService,
            _sharedScriptLibrary,
            null,
            _options,
            NullLogger<InstanceActor>.Instance,
            dcl.Ref)));

        dcl.ExpectMsg<SubscribeTagsRequest>(TimeSpan.FromSeconds(5));

        // Waiter A: predicate that returns false for the CURRENT value ("init") so
        // it clears the fast-path and registers, but THROWS once the value becomes
        // "ready" — exercising the resolve-loop guard (not the fast-path guard).
        Func<object?, bool> boom = v =>
            v?.ToString() == "ready" ? throw new InvalidOperationException("kaboom") : false;
        actor.Tell(new WaitForAttributeRequest(
            "wfa-throw", "Pump1", "State",
            null, boom, TimeSpan.FromSeconds(30), DateTimeOffset.UtcNow));

        // Waiter B: normal value-equality waiting for "ready".
        var target = ZB.MOM.WW.ScadaBridge.Commons.Types.AttributeValueCodec.Encode("ready");
        actor.Tell(new WaitForAttributeRequest(
            "wfa-normal", "Pump1", "State",
            target, null, TimeSpan.FromSeconds(30), DateTimeOffset.UtcNow));

        ExpectNoMsg(TimeSpan.FromMilliseconds(200));

        // One change to "ready": evaluates BOTH waiters on this attribute. The
        // throwing one must be evicted (error reply); the normal one must match.
        actor.Tell(new TagValueUpdate("PLC", tag, "ready", QualityCode.Good, DateTimeOffset.UtcNow));

        // Collect the two replies (order is registry-iteration dependent).
        var r1 = ExpectMsg<WaitForAttributeResponse>(TimeSpan.FromSeconds(5));
        var r2 = ExpectMsg<WaitForAttributeResponse>(TimeSpan.FromSeconds(5));
        var byId = new[] { r1, r2 }.ToDictionary(r => r.CorrelationId);

        var thrown = byId["wfa-throw"];
        Assert.False(thrown.Matched);
        Assert.False(thrown.TimedOut);
        Assert.NotNull(thrown.ErrorMessage);
        Assert.Contains("Wait predicate threw", thrown.ErrorMessage);

        var normal = byId["wfa-normal"];
        Assert.True(normal.Matched);
        Assert.False(normal.TimedOut);
        Assert.Equal("ready", normal.Value);

        // The actor stayed alive and responsive: a follow-up request resolves.
        actor.Tell(new GetAttributeRequest("get-after", "Pump1", "State", DateTimeOffset.UtcNow));
        var get = ExpectMsg<GetAttributeResponse>(TimeSpan.FromSeconds(5));
        Assert.Equal("ready", get.Value);

        // And the throwing waiter was REMOVED (no longer in the registry): driving
        // another change produces NO further reply for it.
        actor.Tell(new TagValueUpdate("PLC", tag, "again", QualityCode.Good, DateTimeOffset.UtcNow));
        ExpectNoMsg(TimeSpan.FromMilliseconds(500));
    }

    // ── 8b. CRITICAL 2 (fast-path): throwing predicate on already-held value ──

    /// <summary>
    /// CRITICAL 2 regression (fast-path analogue of
    /// <see cref="WaitForAttribute_ThrowingPredicate_IsIsolated_SiblingStillMatches"/>):
    /// a predicate that THROWS is registered against an attribute that ALREADY holds a
    /// value, so the fast-path <c>test(current)</c> runs and throws. The actor must
    /// (a) reply a non-matched <c>WaitForAttributeResponse</c> with a non-null
    /// <c>ErrorMessage</c> (predicate-threw), (b) stay alive/responsive (it answers a
    /// subsequent <c>GetAttributeRequest</c>), and (c) NOT register the waiter — there
    /// is no later/second reply even after a value change on that attribute (the
    /// fast-path guard returns WITHOUT scheduling a timeout or storing the waiter).
    /// </summary>
    [Fact]
    public void WaitForAttribute_ThrowingPredicate_FastPath_RepliesError_NoRegistration_ActorStaysAlive()
    {
        const string tag = "ns=3;s=State";
        var config = new FlattenedConfiguration
        {
            InstanceUniqueName = "Pump1",
            Attributes =
            [
                new ResolvedAttribute
                {
                    // Present from construction so the fast-path TryGetValue HITS and
                    // the predicate runs on the current value (and throws).
                    CanonicalName = "State", Value = "init", DataType = "String",
                    DataSourceReference = tag, BoundDataConnectionName = "PLC"
                }
            ]
        };

        var dcl = CreateTestProbe();
        var actor = ActorOf(Props.Create(() => new InstanceActor(
            "Pump1",
            JsonSerializer.Serialize(config),
            _storage,
            _compilationService,
            _sharedScriptLibrary,
            null,
            _options,
            NullLogger<InstanceActor>.Instance,
            dcl.Ref)));

        dcl.ExpectMsg<SubscribeTagsRequest>(TimeSpan.FromSeconds(5));

        // Predicate THROWS unconditionally — the current value "init" is already
        // present, so the fast-path test(current) executes it and throws.
        Func<object?, bool> boom = _ => throw new InvalidOperationException("kaboom");
        actor.Tell(new WaitForAttributeRequest(
            "wfa-fp-throw", "Pump1", "State",
            null, boom, TimeSpan.FromSeconds(30), DateTimeOffset.UtcNow));

        // (a) Non-matched error reply (predicate-threw), guarded on the fast-path.
        var response = ExpectMsg<WaitForAttributeResponse>(TimeSpan.FromSeconds(5));
        Assert.Equal("wfa-fp-throw", response.CorrelationId);
        Assert.False(response.Matched);
        Assert.False(response.TimedOut);
        Assert.NotNull(response.ErrorMessage);
        Assert.Contains("Wait predicate threw", response.ErrorMessage);

        // (b) The actor stayed alive and responsive: a follow-up request resolves.
        actor.Tell(new GetAttributeRequest("get-after-fp", "Pump1", "State", DateTimeOffset.UtcNow));
        var get = ExpectMsg<GetAttributeResponse>(TimeSpan.FromSeconds(5));
        Assert.Equal("init", get.Value);

        // (c) The waiter was NOT registered (no timeout scheduled): driving a value
        // change on "State" produces NO further WaitForAttributeResponse.
        actor.Tell(new TagValueUpdate("PLC", tag, "ready", QualityCode.Good, DateTimeOffset.UtcNow));
        ExpectNoMsg(TimeSpan.FromMilliseconds(500));
    }

    // ── 9. Quality-gated ("Good"-only) matching (spec §4.2) ──────────────────

    /// <summary>
    /// Builds a data-connected instance actor with a single attribute backed by a
    /// DCL probe, draining the initial <c>SubscribeTagsRequest</c>. Used by the
    /// quality-gate tests, which drive value+quality through the DCL ingest path.
    /// </summary>
    private IActorRef CreateDataConnectedActor(
        string instanceName, string attribute, string tag, string dataType, TestProbe dcl)
    {
        var config = new FlattenedConfiguration
        {
            InstanceUniqueName = instanceName,
            Attributes =
            [
                new ResolvedAttribute
                {
                    CanonicalName = attribute, Value = "init", DataType = dataType,
                    DataSourceReference = tag, BoundDataConnectionName = "PLC"
                }
            ]
        };

        var actor = ActorOf(Props.Create(() => new InstanceActor(
            instanceName,
            JsonSerializer.Serialize(config),
            _storage,
            _compilationService,
            _sharedScriptLibrary,
            null,
            _options,
            NullLogger<InstanceActor>.Instance,
            dcl.Ref)));

        dcl.ExpectMsg<SubscribeTagsRequest>(TimeSpan.FromSeconds(5));
        return actor;
    }

    /// <summary>
    /// Spec §4.2 (change-match): with <c>RequireGoodQuality:true</c>, a value that
    /// reaches the target but arrives at <b>Bad</b> quality is NOT a match — the
    /// waiter stays pending and times out.
    /// </summary>
    [Fact]
    public void WaitForAttribute_QualityGated_ChangeMatch_BadQuality_DoesNotMatch_TimesOut()
    {
        const string tag = "ns=3;s=State";
        var dcl = CreateTestProbe();
        var actor = CreateDataConnectedActor("Pump1", "State", tag, "String", dcl);

        var target = ZB.MOM.WW.ScadaBridge.Commons.Types.AttributeValueCodec.Encode("ready");
        actor.Tell(new WaitForAttributeRequest(
            "wfa-qg-bad", "Pump1", "State",
            target, null, TimeSpan.FromMilliseconds(500), DateTimeOffset.UtcNow,
            RequireGoodQuality: true));

        // Value reaches the target but at Bad quality → must NOT match.
        actor.Tell(new TagValueUpdate("PLC", tag, "ready", QualityCode.Bad, DateTimeOffset.UtcNow));

        // The only reply must be the timeout (no spurious Bad-quality match).
        var response = ExpectMsg<WaitForAttributeResponse>(TimeSpan.FromSeconds(3));
        Assert.False(response.Matched);
        Assert.True(response.TimedOut);
        Assert.Equal("wfa-qg-bad", response.CorrelationId);
    }

    /// <summary>
    /// Spec §4.2 (change-match, quality-agnostic baseline): the SAME Bad-quality
    /// value-reaches-target scenario DOES match when <c>RequireGoodQuality:false</c>.
    /// </summary>
    [Fact]
    public void WaitForAttribute_QualityAgnostic_ChangeMatch_BadQuality_Matches()
    {
        const string tag = "ns=3;s=State";
        var dcl = CreateTestProbe();
        var actor = CreateDataConnectedActor("Pump1", "State", tag, "String", dcl);

        var target = ZB.MOM.WW.ScadaBridge.Commons.Types.AttributeValueCodec.Encode("ready");
        actor.Tell(new WaitForAttributeRequest(
            "wfa-qa-bad", "Pump1", "State",
            target, null, TimeSpan.FromSeconds(30), DateTimeOffset.UtcNow,
            RequireGoodQuality: false));

        actor.Tell(new TagValueUpdate("PLC", tag, "ready", QualityCode.Bad, DateTimeOffset.UtcNow));

        var response = ExpectMsg<WaitForAttributeResponse>(TimeSpan.FromSeconds(5));
        Assert.True(response.Matched);
        Assert.False(response.TimedOut);
        Assert.Equal("wfa-qa-bad", response.CorrelationId);
        Assert.Equal("ready", response.Value);
        Assert.Equal("Bad", response.Quality);
    }

    /// <summary>
    /// Spec §4.2 (change-match): with <c>RequireGoodQuality:true</c>, a value that
    /// reaches the target at <b>Good</b> quality matches normally. Also proves the
    /// gate is per-quality not per-value: a Bad-quality arrival at the target is
    /// skipped, then a Good-quality arrival at the target resolves the waiter.
    /// </summary>
    [Fact]
    public void WaitForAttribute_QualityGated_ChangeMatch_GoodQuality_Matches()
    {
        const string tag = "ns=3;s=State";
        var dcl = CreateTestProbe();
        var actor = CreateDataConnectedActor("Pump1", "State", tag, "String", dcl);

        var target = ZB.MOM.WW.ScadaBridge.Commons.Types.AttributeValueCodec.Encode("ready");
        actor.Tell(new WaitForAttributeRequest(
            "wfa-qg-good", "Pump1", "State",
            target, null, TimeSpan.FromSeconds(30), DateTimeOffset.UtcNow,
            RequireGoodQuality: true));

        // First arrival at target but Bad quality is skipped (gate holds it pending).
        actor.Tell(new TagValueUpdate("PLC", tag, "ready", QualityCode.Bad, DateTimeOffset.UtcNow));
        ExpectNoMsg(TimeSpan.FromMilliseconds(400));

        // Then a Good-quality arrival at the target resolves it.
        actor.Tell(new TagValueUpdate("PLC", tag, "ready", QualityCode.Good, DateTimeOffset.UtcNow));

        var response = ExpectMsg<WaitForAttributeResponse>(TimeSpan.FromSeconds(5));
        Assert.True(response.Matched);
        Assert.False(response.TimedOut);
        Assert.Equal("wfa-qg-good", response.CorrelationId);
        Assert.Equal("ready", response.Value);
        Assert.Equal("Good", response.Quality);
    }

    /// <summary>
    /// Spec §4.2 (fast-path): the attribute ALREADY holds the target value at
    /// <b>Bad</b> quality when the quality-gated waiter registers. The fast-path must
    /// NOT reply matched — it registers + schedules the timeout like any pending
    /// waiter, and (here) times out because the value never reaches target at Good.
    /// </summary>
    [Fact]
    public void WaitForAttribute_QualityGated_FastPath_AlreadyAtTargetButBad_DoesNotMatch_TimesOut()
    {
        const string tag = "ns=3;s=State";
        var dcl = CreateTestProbe();
        var actor = CreateDataConnectedActor("Pump1", "State", tag, "String", dcl);

        // Seed the attribute to the target value at Bad quality BEFORE registering.
        actor.Tell(new TagValueUpdate("PLC", tag, "ready", QualityCode.Bad, DateTimeOffset.UtcNow));
        ExpectNoMsg(TimeSpan.FromMilliseconds(200)); // no waiter yet → no reply

        var target = ZB.MOM.WW.ScadaBridge.Commons.Types.AttributeValueCodec.Encode("ready");
        actor.Tell(new WaitForAttributeRequest(
            "wfa-qg-fp-bad", "Pump1", "State",
            target, null, TimeSpan.FromMilliseconds(500), DateTimeOffset.UtcNow,
            RequireGoodQuality: true));

        // Fast-path quality-fail → registers, then times out (no fast matched reply).
        var response = ExpectMsg<WaitForAttributeResponse>(TimeSpan.FromSeconds(3));
        Assert.False(response.Matched);
        Assert.True(response.TimedOut);
        Assert.Equal("wfa-qg-fp-bad", response.CorrelationId);
    }

    /// <summary>
    /// Spec §4.2 (fast-path, quality-agnostic baseline): the SAME already-at-target-
    /// but-Bad attribute fast-path MATCHES when <c>RequireGoodQuality:false</c>.
    /// </summary>
    [Fact]
    public void WaitForAttribute_QualityAgnostic_FastPath_AlreadyAtTargetButBad_Matches()
    {
        const string tag = "ns=3;s=State";
        var dcl = CreateTestProbe();
        var actor = CreateDataConnectedActor("Pump1", "State", tag, "String", dcl);

        actor.Tell(new TagValueUpdate("PLC", tag, "ready", QualityCode.Bad, DateTimeOffset.UtcNow));
        ExpectNoMsg(TimeSpan.FromMilliseconds(200));

        var target = ZB.MOM.WW.ScadaBridge.Commons.Types.AttributeValueCodec.Encode("ready");
        actor.Tell(new WaitForAttributeRequest(
            "wfa-qa-fp-bad", "Pump1", "State",
            target, null, TimeSpan.FromSeconds(30), DateTimeOffset.UtcNow,
            RequireGoodQuality: false));

        var response = ExpectMsg<WaitForAttributeResponse>(TimeSpan.FromSeconds(5));
        Assert.True(response.Matched);
        Assert.False(response.TimedOut);
        Assert.Equal("wfa-qa-fp-bad", response.CorrelationId);
        Assert.Equal("ready", response.Value);
        Assert.Equal("Bad", response.Quality);
    }

    /// <summary>
    /// Spec §4.2 (fast-path): the attribute ALREADY holds the target value at
    /// <b>Good</b> quality when the quality-gated waiter registers → the fast-path
    /// matches immediately.
    /// </summary>
    [Fact]
    public void WaitForAttribute_QualityGated_FastPath_AlreadyAtTargetGood_MatchesImmediately()
    {
        const string tag = "ns=3;s=State";
        var dcl = CreateTestProbe();
        var actor = CreateDataConnectedActor("Pump1", "State", tag, "String", dcl);

        actor.Tell(new TagValueUpdate("PLC", tag, "ready", QualityCode.Good, DateTimeOffset.UtcNow));
        ExpectNoMsg(TimeSpan.FromMilliseconds(200));

        var target = ZB.MOM.WW.ScadaBridge.Commons.Types.AttributeValueCodec.Encode("ready");
        actor.Tell(new WaitForAttributeRequest(
            "wfa-qg-fp-good", "Pump1", "State",
            target, null, TimeSpan.FromSeconds(30), DateTimeOffset.UtcNow,
            RequireGoodQuality: true));

        var response = ExpectMsg<WaitForAttributeResponse>(TimeSpan.FromSeconds(5));
        Assert.True(response.Matched);
        Assert.False(response.TimedOut);
        Assert.Equal("wfa-qg-fp-good", response.CorrelationId);
        Assert.Equal("ready", response.Value);
        Assert.Equal("Good", response.Quality);
    }
}