feat(auditlog): SiteAuditTelemetryActor + ISiteStreamAuditClient seam (#23)

src/ScadaLink.AuditLog/ScadaLink.AuditLog.csproj

@@ -8,6 +8,9 @@
   </PropertyGroup>
 
   <ItemGroup>
+    <!-- Bundle D D1: SiteAuditTelemetryActor + (D2) AuditLogIngestActor live
+         in this project, so Akka is an explicit dependency. -->
+    <PackageReference Include="Akka" />
     <PackageReference Include="Microsoft.Data.Sqlite" />
     <PackageReference Include="Microsoft.Extensions.DependencyInjection.Abstractions" />
     <PackageReference Include="Microsoft.Extensions.Logging.Abstractions" />

src/ScadaLink.AuditLog/Site/SqliteAuditWriter.cs

@@ -2,6 +2,7 @@ using System.Threading.Channels;
 using Microsoft.Data.Sqlite;
 using Microsoft.Extensions.Logging;
 using Microsoft.Extensions.Options;
+using ScadaLink.AuditLog.Site.Telemetry;
 using ScadaLink.Commons.Entities.Audit;
 using ScadaLink.Commons.Interfaces.Services;
 using ScadaLink.Commons.Types.Enums;
@@ -28,7 +29,7 @@ namespace ScadaLink.AuditLog.Site;
 /// the site SQLite schema — central stamps it on ingest.
 /// </para>
 /// </remarks>
-public class SqliteAuditWriter : IAuditWriter, IAsyncDisposable, IDisposable
+public class SqliteAuditWriter : IAuditWriter, ISiteAuditQueue, IAsyncDisposable, IDisposable
 {
     // Microsoft.Data.Sqlite reports a generic SQLITE_CONSTRAINT (error code 19)
     // on a PRIMARY KEY violation; the extended subcode 1555 (SQLITE_CONSTRAINT_PRIMARYKEY)

src/ScadaLink.AuditLog/Site/Telemetry/ISiteAuditQueue.cs

@@ -0,0 +1,34 @@
+using ScadaLink.Commons.Entities.Audit;
+
+namespace ScadaLink.AuditLog.Site.Telemetry;
+
+/// <summary>
+/// Site-local audit-log queue surface consumed by <see cref="SiteAuditTelemetryActor"/>.
+/// Extracted from <see cref="SqliteAuditWriter"/> so the telemetry actor can be
+/// unit-tested against a stub without touching SQLite. <see cref="SqliteAuditWriter"/>
+/// implements this interface; production wiring injects the same instance.
+/// </summary>
+/// <remarks>
+/// Only the two methods the drain loop needs are exposed — the hot-path
+/// <c>WriteAsync</c> stays on <see cref="Commons.Interfaces.Services.IAuditWriter"/>
+/// (script-thread surface), separated by concern from the
+/// telemetry-actor surface so each side can be mocked independently.
+/// </remarks>
+public interface ISiteAuditQueue
+{
+    /// <summary>
+    /// Returns up to <paramref name="limit"/> rows currently in
+    /// <see cref="ScadaLink.Commons.Types.Enums.AuditForwardState.Pending"/>,
+    /// oldest first. Idempotent — repeated calls before
+    /// <see cref="MarkForwardedAsync"/> will yield the same rows again.
+    /// </summary>
+    Task<IReadOnlyList<AuditEvent>> ReadPendingAsync(int limit, CancellationToken ct = default);
+
+    /// <summary>
+    /// Flips the supplied EventIds from
+    /// <see cref="ScadaLink.Commons.Types.Enums.AuditForwardState.Pending"/> to
+    /// <see cref="ScadaLink.Commons.Types.Enums.AuditForwardState.Forwarded"/>.
+    /// Non-existent or already-forwarded ids are silent no-ops.
+    /// </summary>
+    Task MarkForwardedAsync(IReadOnlyList<Guid> eventIds, CancellationToken ct = default);
+}

src/ScadaLink.AuditLog/Site/Telemetry/ISiteStreamAuditClient.cs

@@ -0,0 +1,23 @@
+using ScadaLink.Communication.Grpc;
+
+namespace ScadaLink.AuditLog.Site.Telemetry;
+
+/// <summary>
+/// Mockable abstraction over the central site-stream gRPC client surface that
+/// <see cref="SiteAuditTelemetryActor"/> uses to push <see cref="AuditEventBatch"/>
+/// payloads. The production implementation (added in Bundle E host wiring)
+/// wraps the auto-generated <c>SiteStreamService.SiteStreamServiceClient</c>;
+/// unit tests substitute via NSubstitute against this interface so the actor
+/// never needs a live gRPC channel.
+/// </summary>
+public interface ISiteStreamAuditClient
+{
+    /// <summary>
+    /// Pushes <paramref name="batch"/> to the central <c>IngestAuditEvents</c>
+    /// RPC. The returned <see cref="IngestAck"/> carries the
+    /// <c>accepted_event_ids</c> the actor will flip to
+    /// <see cref="ScadaLink.Commons.Types.Enums.AuditForwardState.Forwarded"/>
+    /// in the site SQLite queue.
+    /// </summary>
+    Task<IngestAck> IngestAuditEventsAsync(AuditEventBatch batch, CancellationToken ct);
+}

src/ScadaLink.AuditLog/Site/Telemetry/SiteAuditTelemetryActor.cs

@@ -0,0 +1,179 @@
+using Akka.Actor;
+using Microsoft.Extensions.Logging;
+using Microsoft.Extensions.Options;
+using ScadaLink.AuditLog.Telemetry;
+using ScadaLink.Commons.Entities.Audit;
+using ScadaLink.Communication.Grpc;
+
+namespace ScadaLink.AuditLog.Site.Telemetry;
+
+/// <summary>
+/// Site-side actor that drains the local SQLite audit queue and pushes Pending
+/// rows to central via the <c>IngestAuditEvents</c> gRPC RPC. On a successful
+/// ack the matching EventIds flip to
+/// <see cref="ScadaLink.Commons.Types.Enums.AuditForwardState.Forwarded"/>; on
+/// a gRPC failure the rows stay Pending and the next drain retries.
+/// </summary>
+/// <remarks>
+/// <para>
+/// The drain self-tick is a private <c>Drain</c> message scheduled via the
+/// actor system scheduler. The cadence is options-driven: <c>BusyIntervalSeconds</c>
+/// when the previous drain found rows (or faulted — we want quick recovery),
+/// <c>IdleIntervalSeconds</c> when the queue was empty.
+/// </para>
+/// <para>
+/// Both collaborators are injected as interfaces (<see cref="ISiteAuditQueue"/>
+/// and <see cref="ISiteStreamAuditClient"/>) so unit tests substitute with
+/// NSubstitute and never touch real SQLite or gRPC.
+/// </para>
+/// <para>
+/// Per Bundle D's brief, audit-write paths must be fail-safe — a thrown
+/// exception inside the actor MUST NOT crash it. The Drain handler wraps the
+/// pipeline in a top-level try/catch that logs and re-schedules, and the
+/// actor's <see cref="SupervisorStrategy"/> defaults to
+/// <see cref="Akka.Actor.SupervisorStrategy.DefaultStrategy"/>'s Restart for
+/// child actors — but this actor has no children, so the catch is what matters.
+/// </para>
+/// </remarks>
+public class SiteAuditTelemetryActor : ReceiveActor
+{
+    private readonly ISiteAuditQueue _queue;
+    private readonly ISiteStreamAuditClient _client;
+    private readonly SiteAuditTelemetryOptions _options;
+    private readonly ILogger<SiteAuditTelemetryActor> _logger;
+    private ICancelable? _pendingTick;
+
+    public SiteAuditTelemetryActor(
+        ISiteAuditQueue queue,
+        ISiteStreamAuditClient client,
+        IOptions<SiteAuditTelemetryOptions> options,
+        ILogger<SiteAuditTelemetryActor> logger)
+    {
+        ArgumentNullException.ThrowIfNull(queue);
+        ArgumentNullException.ThrowIfNull(client);
+        ArgumentNullException.ThrowIfNull(options);
+        ArgumentNullException.ThrowIfNull(logger);
+
+        _queue = queue;
+        _client = client;
+        _options = options.Value;
+        _logger = logger;
+
+        ReceiveAsync<Drain>(_ => OnDrainAsync());
+    }
+
+    protected override void PreStart()
+    {
+        base.PreStart();
+        // Initial tick fires on the busy interval so the actor starts polling
+        // soon after host startup. A subsequent empty drain will move to the
+        // idle interval naturally.
+        ScheduleNext(TimeSpan.FromSeconds(_options.BusyIntervalSeconds));
+    }
+
+    protected override void PostStop()
+    {
+        _pendingTick?.Cancel();
+        base.PostStop();
+    }
+
+    private async Task OnDrainAsync()
+    {
+        var nextDelay = TimeSpan.FromSeconds(_options.BusyIntervalSeconds);
+        try
+        {
+            var pending = await _queue.ReadPendingAsync(_options.BatchSize, CancellationToken.None)
+                .ConfigureAwait(false);
+            if (pending.Count == 0)
+            {
+                // No rows — settle into the idle cadence until the next write
+                // bumps us back into the busy cadence.
+                nextDelay = TimeSpan.FromSeconds(_options.IdleIntervalSeconds);
+                return;
+            }
+
+            var batch = BuildBatch(pending);
+
+            IngestAck ack;
+            try
+            {
+                ack = await _client.IngestAuditEventsAsync(batch, CancellationToken.None)
+                    .ConfigureAwait(false);
+            }
+            catch (Exception ex)
+            {
+                // gRPC fault — leave the rows in Pending so the next drain
+                // retries. Bundle D's brief: "On gRPC exception (any), log
+                // Warning, schedule next Drain in BusyIntervalSeconds."
+                _logger.LogWarning(ex,
+                    "IngestAuditEvents push failed for {Count} pending events; will retry next drain.",
+                    pending.Count);
+                return;
+            }
+
+            var acceptedIds = ParseAcceptedIds(ack);
+            if (acceptedIds.Count > 0)
+            {
+                await _queue.MarkForwardedAsync(acceptedIds, CancellationToken.None)
+                    .ConfigureAwait(false);
+            }
+        }
+        catch (Exception ex)
+        {
+            // Catch-all so a SQLite hiccup or mapper bug never crashes the
+            // actor. The next tick is still scheduled in the finally block.
+            _logger.LogError(ex, "Unexpected error during audit-log telemetry drain.");
+        }
+        finally
+        {
+            ScheduleNext(nextDelay);
+        }
+    }
+
+    private static AuditEventBatch BuildBatch(IReadOnlyList<AuditEvent> events)
+    {
+        var batch = new AuditEventBatch();
+        foreach (var e in events)
+        {
+            batch.Events.Add(AuditEventMapper.ToDto(e));
+        }
+        return batch;
+    }
+
+    private static IReadOnlyList<Guid> ParseAcceptedIds(IngestAck ack)
+    {
+        if (ack.AcceptedEventIds.Count == 0)
+        {
+            return Array.Empty<Guid>();
+        }
+
+        var list = new List<Guid>(ack.AcceptedEventIds.Count);
+        foreach (var raw in ack.AcceptedEventIds)
+        {
+            if (Guid.TryParse(raw, out var id))
+            {
+                list.Add(id);
+            }
+            // Malformed ids are ignored — central should never emit them, but
+            // we refuse to crash the actor over a bad string.
+        }
+        return list;
+    }
+
+    private void ScheduleNext(TimeSpan delay)
+    {
+        _pendingTick?.Cancel();
+        _pendingTick = Context.System.Scheduler.ScheduleTellOnceCancelable(
+            delay,
+            Self,
+            Drain.Instance,
+            Self);
+    }
+
+    /// <summary>Self-tick message that triggers a drain cycle.</summary>
+    private sealed class Drain
+    {
+        public static readonly Drain Instance = new();
+        private Drain() { }
+    }
+}

src/ScadaLink.AuditLog/Site/Telemetry/SiteAuditTelemetryOptions.cs

@@ -0,0 +1,28 @@
+namespace ScadaLink.AuditLog.Site.Telemetry;
+
+/// <summary>
+/// Tuning knobs for the site-side <see cref="SiteAuditTelemetryActor"/> drain
+/// loop. Defaults mirror Bundle D's plan: drain every 5 s while rows are
+/// flowing (busy), every 30 s when the queue is empty (idle).
+/// </summary>
+public sealed class SiteAuditTelemetryOptions
+{
+    /// <summary>
+    /// Maximum number of <see cref="ScadaLink.Commons.Entities.Audit.AuditEvent"/>
+    /// rows read from the site SQLite queue and pushed in a single gRPC batch.
+    /// </summary>
+    public int BatchSize { get; set; } = 256;
+
+    /// <summary>
+    /// Delay between drains when the previous drain found at least one Pending
+    /// row OR the previous push faulted. Re-drain quickly to keep telemetry
+    /// flowing and to retry transient gRPC errors.
+    /// </summary>
+    public int BusyIntervalSeconds { get; set; } = 5;
+
+    /// <summary>
+    /// Delay between drains when the previous drain found no Pending rows.
+    /// Longer interval avoids hammering an idle SQLite + gRPC channel.
+    /// </summary>
+    public int IdleIntervalSeconds { get; set; } = 30;
+}

tests/ScadaLink.AuditLog.Tests/ScadaLink.AuditLog.Tests.csproj

@@ -9,14 +9,30 @@
   </PropertyGroup>
 
   <ItemGroup>
+    <PackageReference Include="Akka.TestKit.Xunit2" />
     <PackageReference Include="coverlet.collector" />
+    <!--
+      Bundle D D2 needs Microsoft.Data.SqlClient for the MsSqlMigrationFixture
+      (mirroring ScadaLink.ConfigurationDatabase.Tests). Pinning 6.1.1 here for
+      the same reason: EF SqlServer 10.0.7 needs >= 6.1.1 but the central pin
+      is 6.0.2 (production ExternalSystemGateway). Override is test-only.
+    -->
+    <PackageReference Include="Microsoft.Data.SqlClient" VersionOverride="6.1.1" />
     <PackageReference Include="Microsoft.Data.Sqlite" />
+    <PackageReference Include="Microsoft.EntityFrameworkCore.SqlServer" />
     <PackageReference Include="Microsoft.Extensions.Configuration.Json" />
+    <PackageReference Include="Microsoft.Extensions.DependencyInjection" />
     <PackageReference Include="Microsoft.Extensions.Logging.Abstractions" />
     <PackageReference Include="Microsoft.NET.Test.Sdk" />
     <PackageReference Include="NSubstitute" />
     <PackageReference Include="xunit" />
     <PackageReference Include="xunit.runner.visualstudio" />
+    <!--
+      SkippableFact pattern (xunit 2.9.x has no native Assert.Skip) — used by
+      Bundle D D2 MSSQL-backed AuditLogIngestActor tests to report Skipped when
+      the dev MSSQL container is not reachable.
+    -->
+    <PackageReference Include="Xunit.SkippableFact" />
   </ItemGroup>
 
   <ItemGroup>
@@ -25,6 +41,13 @@
 
   <ItemGroup>
     <ProjectReference Include="../../src/ScadaLink.AuditLog/ScadaLink.AuditLog.csproj" />
+    <!--
+      D2: the AuditLogIngestActor tests use the real AuditLogRepository against
+      a per-test MSSQL database via MsSqlMigrationFixture. The fixture lives in
+      ScadaLink.ConfigurationDatabase.Tests; we reference that test project so
+      the fixture + EF migrations come along without duplicating them.
+    -->
+    <ProjectReference Include="../ScadaLink.ConfigurationDatabase.Tests/ScadaLink.ConfigurationDatabase.Tests.csproj" />
   </ItemGroup>
 
 </Project>

tests/ScadaLink.AuditLog.Tests/Site/Telemetry/SiteAuditTelemetryActorTests.cs

@@ -0,0 +1,235 @@
+using Akka.Actor;
+using Akka.TestKit.Xunit2;
+using Google.Protobuf.WellKnownTypes;
+using Microsoft.Extensions.Logging.Abstractions;
+using Microsoft.Extensions.Options;
+using NSubstitute;
+using NSubstitute.ExceptionExtensions;
+using ScadaLink.AuditLog.Site.Telemetry;
+using ScadaLink.Commons.Entities.Audit;
+using ScadaLink.Commons.Types.Enums;
+using ScadaLink.Communication.Grpc;
+
+namespace ScadaLink.AuditLog.Tests.Site.Telemetry;
+
+/// <summary>
+/// Bundle D D1 tests for <see cref="SiteAuditTelemetryActor"/>. The actor drains
+/// the site SQLite queue via <see cref="ISiteAuditQueue"/>, pushes batches via
+/// <see cref="ISiteStreamAuditClient"/>, and flips ack'd rows to Forwarded.
+/// Both collaborators are NSubstitute mocks so the tests never touch real
+/// SQLite or gRPC.
+/// </summary>
+public class SiteAuditTelemetryActorTests : TestKit
+{
+    private readonly ISiteAuditQueue _queue = Substitute.For<ISiteAuditQueue>();
+    private readonly ISiteStreamAuditClient _client = Substitute.For<ISiteStreamAuditClient>();
+
+    /// <summary>
+    /// Fast options so tests don't stall waiting for the scheduler. 1s busy /
+    /// 2s idle still exercises the busy-vs-idle branching, but each test
+    /// completes in &lt; 5 s wall-clock.
+    /// </summary>
+    private static IOptions<SiteAuditTelemetryOptions> Opts(
+        int batchSize = 256,
+        int busySeconds = 1,
+        int idleSeconds = 2) =>
+        Options.Create(new SiteAuditTelemetryOptions
+        {
+            BatchSize = batchSize,
+            BusyIntervalSeconds = busySeconds,
+            IdleIntervalSeconds = idleSeconds,
+        });
+
+    private IActorRef CreateActor(IOptions<SiteAuditTelemetryOptions>? options = null) =>
+        Sys.ActorOf(Props.Create(() => new SiteAuditTelemetryActor(
+            _queue,
+            _client,
+            options ?? Opts(),
+            NullLogger<SiteAuditTelemetryActor>.Instance)));
+
+    private static AuditEvent NewEvent(Guid? id = null) => new()
+    {
+        EventId = id ?? Guid.NewGuid(),
+        OccurredAtUtc = new DateTime(2026, 5, 20, 10, 0, 0, DateTimeKind.Utc),
+        Channel = AuditChannel.ApiOutbound,
+        Kind = AuditKind.ApiCall,
+        Status = AuditStatus.Delivered,
+        SourceSiteId = "site-1",
+        ForwardState = AuditForwardState.Pending,
+    };
+
+    private static IngestAck AckAll(IReadOnlyList<AuditEvent> events)
+    {
+        var ack = new IngestAck();
+        foreach (var e in events)
+        {
+            ack.AcceptedEventIds.Add(e.EventId.ToString());
+        }
+        return ack;
+    }
+
+    [Fact]
+    public async Task Drain_With_50PendingRows_Sends_OneBatch_Of_50_Then_FlipsToForwarded()
+    {
+        // Arrange — 50 pending rows on the first read, then empty on subsequent
+        // reads so the actor settles after one productive drain.
+        var pending = Enumerable.Range(0, 50).Select(_ => NewEvent()).ToList();
+        _queue.ReadPendingAsync(Arg.Any<int>(), Arg.Any<CancellationToken>())
+            .Returns(
+                Task.FromResult<IReadOnlyList<AuditEvent>>(pending),
+                Task.FromResult<IReadOnlyList<AuditEvent>>(Array.Empty<AuditEvent>()));
+
+        AuditEventBatch? capturedBatch = null;
+        _client.IngestAuditEventsAsync(Arg.Any<AuditEventBatch>(), Arg.Any<CancellationToken>())
+            .Returns(call =>
+            {
+                capturedBatch = call.Arg<AuditEventBatch>();
+                return Task.FromResult(AckAll(pending));
+            });
+
+        // Act
+        CreateActor();
+
+        // Assert — give the scheduler time to fire the initial Drain tick.
+        await AwaitAssertAsync(async () =>
+        {
+            await _client.Received(1).IngestAuditEventsAsync(
+                Arg.Any<AuditEventBatch>(), Arg.Any<CancellationToken>());
+            await _queue.Received(1).MarkForwardedAsync(
+                Arg.Is<IReadOnlyList<Guid>>(g => g.Count == 50), Arg.Any<CancellationToken>());
+        }, TimeSpan.FromSeconds(5));
+
+        Assert.NotNull(capturedBatch);
+        Assert.Equal(50, capturedBatch!.Events.Count);
+
+        var expected = pending.Select(e => e.EventId).ToHashSet();
+        await _queue.Received(1).MarkForwardedAsync(
+            Arg.Is<IReadOnlyList<Guid>>(g => g.ToHashSet().SetEquals(expected)),
+            Arg.Any<CancellationToken>());
+    }
+
+    [Fact]
+    public async Task Drain_GrpcThrows_RowsStayPending_NextDrainRetries()
+    {
+        // Arrange — first read returns 3 rows; the gRPC client throws on the
+        // first push, then succeeds on the second. After the second push the
+        // queue returns empty so the actor settles.
+        var batch = Enumerable.Range(0, 3).Select(_ => NewEvent()).ToList();
+        _queue.ReadPendingAsync(Arg.Any<int>(), Arg.Any<CancellationToken>())
+            .Returns(
+                Task.FromResult<IReadOnlyList<AuditEvent>>(batch),
+                Task.FromResult<IReadOnlyList<AuditEvent>>(batch),
+                Task.FromResult<IReadOnlyList<AuditEvent>>(Array.Empty<AuditEvent>()));
+
+        var calls = 0;
+        _client.IngestAuditEventsAsync(Arg.Any<AuditEventBatch>(), Arg.Any<CancellationToken>())
+            .Returns(_ =>
+            {
+                calls++;
+                if (calls == 1)
+                {
+                    throw new InvalidOperationException("simulated gRPC failure");
+                }
+                return Task.FromResult(AckAll(batch));
+            });
+
+        // Act
+        CreateActor();
+
+        // Assert — eventually MarkForwardedAsync is called exactly once (after
+        // the retry succeeded). The first failure must NOT have called
+        // MarkForwardedAsync because the rows stay Pending.
+        await AwaitAssertAsync(async () =>
+        {
+            await _queue.Received(1).MarkForwardedAsync(
+                Arg.Any<IReadOnlyList<Guid>>(), Arg.Any<CancellationToken>());
+        }, TimeSpan.FromSeconds(10));
+
+        Assert.True(calls >= 2, $"Expected at least 2 client calls (1 failure + 1 retry); saw {calls}");
+    }
+
+    [Fact]
+    public async Task Drain_ZeroPending_SchedulesAtIdleInterval_NoClientCall()
+    {
+        // Arrange — queue always empty.
+        _queue.ReadPendingAsync(Arg.Any<int>(), Arg.Any<CancellationToken>())
+            .Returns(Task.FromResult<IReadOnlyList<AuditEvent>>(Array.Empty<AuditEvent>()));
+
+        // Idle interval = 2 s. Pause 3 s after the first tick (1 s busy on
+        // PreStart) and assert the empty-queue branch did NOT push to the
+        // client.
+        CreateActor(Opts(busySeconds: 1, idleSeconds: 2));
+
+        // Allow the initial tick (~1 s) + a generous window for the idle re-tick.
+        await Task.Delay(TimeSpan.FromSeconds(3));
+
+        await _client.DidNotReceiveWithAnyArgs().IngestAuditEventsAsync(default!, default);
+
+        // ReadPendingAsync was called at least once (initial tick), and at
+        // most twice within the 3 s window (initial + one idle re-tick).
+        var readCalls = _queue.ReceivedCalls()
+            .Count(c => c.GetMethodInfo().Name == nameof(ISiteAuditQueue.ReadPendingAsync));
+        Assert.InRange(readCalls, 1, 2);
+    }
+
+    [Fact]
+    public async Task Drain_NonZeroPending_SchedulesAtBusyInterval()
+    {
+        // Arrange — every read returns 1 row. With busy=1s the actor should
+        // re-drain quickly, producing multiple client calls inside a short
+        // window.
+        var single = new List<AuditEvent> { NewEvent() };
+        _queue.ReadPendingAsync(Arg.Any<int>(), Arg.Any<CancellationToken>())
+            .Returns(Task.FromResult<IReadOnlyList<AuditEvent>>(single));
+
+        _client.IngestAuditEventsAsync(Arg.Any<AuditEventBatch>(), Arg.Any<CancellationToken>())
+            .Returns(call => Task.FromResult(AckAll(single)));
+
+        CreateActor(Opts(busySeconds: 1, idleSeconds: 10));
+
+        // 3-second window with busy=1s should fit at least 2 drains.
+        await Task.Delay(TimeSpan.FromSeconds(3));
+
+        var pushCalls = _client.ReceivedCalls()
+            .Count(c => c.GetMethodInfo().Name == nameof(ISiteStreamAuditClient.IngestAuditEventsAsync));
+        Assert.True(pushCalls >= 2,
+            $"Expected ≥2 pushes within 3s when busy=1s; saw {pushCalls}");
+    }
+
+    [Fact]
+    public async Task Drain_AcceptedEventIdsSubset_OnlyMarksAccepted()
+    {
+        // Arrange — 5 rows pushed, but the central ack only lists 3.
+        var rows = Enumerable.Range(0, 5).Select(_ => NewEvent()).ToList();
+        var ackedIds = rows.Take(3).Select(r => r.EventId).ToList();
+
+        _queue.ReadPendingAsync(Arg.Any<int>(), Arg.Any<CancellationToken>())
+            .Returns(
+                Task.FromResult<IReadOnlyList<AuditEvent>>(rows),
+                Task.FromResult<IReadOnlyList<AuditEvent>>(Array.Empty<AuditEvent>()));
+
+        var partialAck = new IngestAck();
+        foreach (var id in ackedIds)
+        {
+            partialAck.AcceptedEventIds.Add(id.ToString());
+        }
+        _client.IngestAuditEventsAsync(Arg.Any<AuditEventBatch>(), Arg.Any<CancellationToken>())
+            .Returns(Task.FromResult(partialAck));
+
+        // Act
+        CreateActor();
+
+        await AwaitAssertAsync(async () =>
+        {
+            await _queue.Received(1).MarkForwardedAsync(
+                Arg.Any<IReadOnlyList<Guid>>(), Arg.Any<CancellationToken>());
+        }, TimeSpan.FromSeconds(5));
+
+        // Assert — exactly the 3 ack'd ids made it to MarkForwardedAsync, not
+        // the other 2.
+        var ackedSet = ackedIds.ToHashSet();
+        await _queue.Received(1).MarkForwardedAsync(
+            Arg.Is<IReadOnlyList<Guid>>(g => g.Count == 3 && g.ToHashSet().SetEquals(ackedSet)),
+            Arg.Any<CancellationToken>());
+    }
+}