backend: stateless queue consumers become [QueueTrigger] functions; webhooks go through a queue; poll loop receives one message at a time

Audit 2026-09-05 F01/F02. Analyze, VulnerabilityCorrelate, TenantAutoApprove,
IndexReconcile and ImeMsiArchive are now host-managed queue functions under
Functions/Queue bound to the DataStorage connection (identity-based, resolved
by the host): the Flex scale controller wakes an instance for a message, the
host renews the lease and owns the retry/poison ladder. The four cascade
workers with lease/CAS state stay self-polling BackgroundServices; their base
now receives one message per poll so waiting messages can no longer expire
under another worker. Terminal enrollment alerts and hardware-rejection alerts
are enqueued on notification-dispatch (channel ids only, resolved at send
time) instead of fire-and-forget sends the host could drop on scale-in.
Requires the app setting DataStorage__queueServiceUri before deploy.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>

src/Backend/AutopilotMonitor.Functions.Tests/AnalyzeOnEnrollmentEndEnvelopeSerializationTests.cs

@@ -6,7 +6,7 @@ namespace AutopilotMonitor.Functions.Tests;
 /// <summary>
 /// JSON round-trip tests for <see cref="AnalyzeOnEnrollmentEndEnvelope"/>. The producer
 /// (<c>AzureQueueAnalyzeOnEnrollmentEndProducer</c>) serializes with Newtonsoft and the
-/// queue worker (<c>AnalyzeOnEnrollmentEndQueueWorker</c>) deserializes the same shape —
+/// queue worker (<c>AnalyzeOnEnrollmentEndQueueFunction</c>) deserializes the same shape —
 /// pinning the contract here keeps the queue from silently breaking when fields are added.
 /// </summary>
 public class AnalyzeOnEnrollmentEndEnvelopeSerializationTests

src/Backend/AutopilotMonitor.Functions.Tests/AzureQueueImeMsiArchiveProducerTests.cs

@@ -61,7 +61,8 @@ public class AzureQueueImeMsiArchiveProducerTests
     public async Task EnqueueAsync_send_failure_is_swallowed()
     {
         var (producer, queue, _) = BuildSut();
-        queue.Setup(q => q.SendMessageAsync(It.IsAny<string>(), It.IsAny<CancellationToken>()))
+        queue.Setup(q => q.SendMessageAsync(
+                It.IsAny<string>(), It.IsAny<TimeSpan?>(), It.IsAny<TimeSpan?>(), It.IsAny<CancellationToken>()))
             .ThrowsAsync(new RequestFailedException(503, "queue busy"));
 
         // Fire-and-forget continuation on the ingest path — must never throw.
@@ -107,8 +108,9 @@ public class AzureQueueImeMsiArchiveProducerTests
         queue.Setup(q => q.CreateIfNotExistsAsync(
                 It.IsAny<IDictionary<string, string>>(), It.IsAny<CancellationToken>()))
             .ReturnsAsync((Response?)null);
-        queue.Setup(q => q.SendMessageAsync(It.IsAny<string>(), It.IsAny<CancellationToken>()))
-            .Returns<string, CancellationToken>((body, _) =>
+        queue.Setup(q => q.SendMessageAsync(
+                It.IsAny<string>(), It.IsAny<TimeSpan?>(), It.IsAny<TimeSpan?>(), It.IsAny<CancellationToken>()))
+            .Returns<string, TimeSpan?, TimeSpan?, CancellationToken>((body, _, _, _) =>
             {
                 sentBodies.Add(body);
                 var receipt = QueuesModelFactory.SendReceipt(

src/Backend/AutopilotMonitor.Functions.Tests/ImeMsiArchiveQueueFunctionTests.cs

@@ -0,0 +1,180 @@
+using System;
+using System.Threading;
+using System.Threading.Tasks;
+using AutopilotMonitor.Functions.Functions.Queue;
+using AutopilotMonitor.Functions.Services;
+using AutopilotMonitor.Functions.Services.Ime;
+using AutopilotMonitor.Shared.DataAccess;
+using AutopilotMonitor.Shared.Models;
+using Azure.Storage.Queues.Models;
+using Microsoft.Extensions.Caching.Memory;
+using Microsoft.Extensions.Logging.Abstractions;
+using Moq;
+using Newtonsoft.Json;
+using Xunit;
+
+namespace AutopilotMonitor.Functions.Tests;
+
+/// <summary>
+/// Contract of the ime-msi-archive queue function (successor of the self-polling worker):
+/// merge the archiver's outcome onto the ImeVersionHistory row on EVERY attempt (also
+/// failures, so operators see a stuck download), return for terminal outcomes so the host
+/// deletes the message, throw for retryable ones so the host's retry → poison ladder applies,
+/// re-enqueue with a delay — not drop — while ImeMsiArchivingEnabled is off, and drop
+/// version-less envelopes before the archiver runs.
+/// </summary>
+public class ImeMsiArchiveQueueFunctionTests
+{
+    private const string Version = "1.104.102.0";
+
+    [Fact]
+    public async Task Success_merges_row_and_completes()
+    {
+        var h = new Harness();
+        h.Archiver.Result = new ImeMsiArchiveResult(
+            true, ImeMsiArchiver.Statuses.Archived, Retryable: false,
+            $"{Version}/IntuneWindowsAgent.msi", "abc123", 42L, ImeMsiArchiver.FallbackMsiUrls[1]);
+
+        await h.Sut.ProcessAsync(Message(Envelope()), CancellationToken.None);
+
+        h.Repo.Verify(r => r.UpdateImeVersionArchiveInfoAsync(
+            Version, ImeMsiArchiver.Statuses.Archived,
+            $"{Version}/IntuneWindowsAgent.msi", "abc123", 42L, ImeMsiArchiver.FallbackMsiUrls[1]),
+            Times.Once);
+        h.Producer.Verify(p => p.EnqueueAsync(
+            It.IsAny<ImeMsiArchiveEnvelope>(), It.IsAny<TimeSpan?>(), It.IsAny<CancellationToken>()),
+            Times.Never);
+    }
+
+    [Fact]
+    public async Task Retryable_failure_merges_row_and_throws_for_host_retry()
+    {
+        var h = new Harness();
+        h.Archiver.Result = new ImeMsiArchiveResult(
+            false, ImeMsiArchiver.Statuses.FailedDownload, Retryable: true,
+            null, null, null, ImeMsiArchiver.FallbackMsiUrls[1]);
+
+        await Assert.ThrowsAsync<InvalidOperationException>(
+            () => h.Sut.ProcessAsync(Message(Envelope()), CancellationToken.None));
+
+        // Row always tells the truth — the failure status IS merged before the throw.
+        h.Repo.Verify(r => r.UpdateImeVersionArchiveInfoAsync(
+            Version, ImeMsiArchiver.Statuses.FailedDownload,
+            null, null, null, ImeMsiArchiver.FallbackMsiUrls[1]),
+            Times.Once);
+    }
+
+    [Fact]
+    public async Task Permanent_failure_merges_row_and_completes()
+    {
+        var h = new Harness();
+        h.Archiver.Result = new ImeMsiArchiveResult(
+            false, ImeMsiArchiver.Statuses.FailedTooLarge, Retryable: false,
+            null, null, 999L, ImeMsiArchiver.FallbackMsiUrls[1]);
+
+        await h.Sut.ProcessAsync(Message(Envelope()), CancellationToken.None);
+
+        h.Repo.Verify(r => r.UpdateImeVersionArchiveInfoAsync(
+            Version, ImeMsiArchiver.Statuses.FailedTooLarge,
+            null, null, 999L, ImeMsiArchiver.FallbackMsiUrls[1]),
+            Times.Once);
+    }
+
+    [Fact]
+    public async Task Paused_requeues_with_delay_and_skips_archiver()
+    {
+        var h = new Harness(archivingEnabled: false);
+
+        await h.Sut.ProcessAsync(Message(Envelope()), CancellationToken.None);
+
+        Assert.Equal(0, h.Archiver.Calls);
+        h.Producer.Verify(p => p.EnqueueAsync(
+            It.Is<ImeMsiArchiveEnvelope>(e => e.Version == Version),
+            ImeMsiArchiveQueueFunction.PauseRequeueDelay,
+            It.IsAny<CancellationToken>()),
+            Times.Once);
+        h.Repo.Verify(r => r.UpdateImeVersionArchiveInfoAsync(
+            It.IsAny<string>(), It.IsAny<string>(), It.IsAny<string?>(),
+            It.IsAny<string?>(), It.IsAny<long?>(), It.IsAny<string?>()),
+            Times.Never);
+    }
+
+    [Fact]
+    public async Task Envelope_without_version_is_dropped_before_archiver()
+    {
+        var h = new Harness();
+
+        await h.Sut.ProcessAsync(Message(new ImeMsiArchiveEnvelope
+        {
+            Version = string.Empty,
+            EnqueuedAt = DateTime.UtcNow,
+        }), CancellationToken.None);
+
+        Assert.Equal(0, h.Archiver.Calls);
+        h.Repo.Verify(r => r.UpdateImeVersionArchiveInfoAsync(
+            It.IsAny<string>(), It.IsAny<string>(), It.IsAny<string?>(),
+            It.IsAny<string?>(), It.IsAny<long?>(), It.IsAny<string?>()),
+            Times.Never);
+        h.Producer.Verify(p => p.EnqueueAsync(
+            It.IsAny<ImeMsiArchiveEnvelope>(), It.IsAny<TimeSpan?>(), It.IsAny<CancellationToken>()),
+            Times.Never);
+    }
+
+    // ============================================================ Helpers ====
+
+    private static ImeMsiArchiveEnvelope Envelope() => new()
+    {
+        Version = Version,
+        MsiDownloadUrl = ImeMsiArchiver.FallbackMsiUrls[1],
+        MsiMatchedBy = "productVersion",
+        TenantId = "11111111-1111-1111-1111-111111111111",
+        SessionId = "aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa",
+        EnqueuedAt = DateTime.UtcNow,
+    };
+
+    private static QueueMessage Message(ImeMsiArchiveEnvelope envelope) => QueuesModelFactory.QueueMessage(
+        messageId: "msg-" + Guid.NewGuid().ToString("N"),
+        popReceipt: "pop-" + Guid.NewGuid().ToString("N"),
+        body: new BinaryData(JsonConvert.SerializeObject(envelope)),
+        dequeueCount: 1);
+
+    /// <summary>Scripted archiver — the function only consumes the classified result.</summary>
+    private sealed class StubArchiver : ImeMsiArchiver
+    {
+        private int _calls;
+
+        public ImeMsiArchiveResult Result { get; set; } = new(
+            true, Statuses.Archived, Retryable: false, null, null, null, null);
+
+        public int Calls => Volatile.Read(ref _calls);
+
+        public override Task<ImeMsiArchiveResult> ArchiveAsync(
+            ImeMsiArchiveEnvelope envelope, CancellationToken cancellationToken = default)
+        {
+            Interlocked.Increment(ref _calls);
+            return Task.FromResult(Result);
+        }
+    }
+
+    private sealed class Harness
+    {
+        public StubArchiver Archiver { get; } = new();
+        public Mock<ISessionRepository> Repo { get; } = new();
+        public Mock<IImeMsiArchiveProducer> Producer { get; } = new();
+        public ImeMsiArchiveQueueFunction Sut { get; }
+
+        public Harness(bool archivingEnabled = true)
+        {
+            var adminConfig = new Mock<AdminConfigurationService>(
+                Mock.Of<IConfigRepository>(),
+                NullLogger<AdminConfigurationService>.Instance,
+                new MemoryCache(new MemoryCacheOptions()));
+            adminConfig.Setup(a => a.GetConfigurationAsync())
+                .ReturnsAsync(new AdminConfiguration { ImeMsiArchivingEnabled = archivingEnabled });
+
+            Sut = new ImeMsiArchiveQueueFunction(
+                Archiver, Repo.Object, adminConfig.Object, Producer.Object,
+                NullLogger<ImeMsiArchiveQueueFunction>.Instance);
+        }
+    }
+}

src/Backend/AutopilotMonitor.Functions.Tests/ImeMsiArchiveQueueWorkerTests.cs

@@ -1,261 +0,0 @@
-using System;
-using System.Collections.Generic;
-using System.Linq;
-using System.Threading;
-using System.Threading.Tasks;
-using Azure;
-using Azure.Storage.Queues;
-using Azure.Storage.Queues.Models;
-using AutopilotMonitor.Functions.Services;
-using AutopilotMonitor.Functions.Services.Ime;
-using AutopilotMonitor.Shared.DataAccess;
-using AutopilotMonitor.Shared.Models;
-using Microsoft.Extensions.Caching.Memory;
-using Microsoft.Extensions.Logging.Abstractions;
-using Moq;
-using Newtonsoft.Json;
-using Xunit;
-
-namespace AutopilotMonitor.Functions.Tests;
-
-/// <summary>
-/// Worker-level tests for the ime-msi-archive queue consumer (daily review 2026-08-18:
-/// previously untested). The worker's contract: merge the archiver's outcome onto the
-/// ImeVersionHistory row on EVERY attempt (also failures, so operators see a stuck
-/// download), delete the message for terminal outcomes, rethrow for retryable ones
-/// (visibility-timeout ladder), pause — not drop — while ImeMsiArchivingEnabled is off,
-/// and drop version-less envelopes without invoking the archiver.
-/// </summary>
-public class ImeMsiArchiveQueueWorkerTests
-{
-    private const string Version = "1.104.102.0";
-
-    [Fact]
-    public async Task Worker_success_merges_row_and_deletes_message()
-    {
-        var harness = new Harness();
-        harness.Archiver.Result = new ImeMsiArchiveResult(
-            true, ImeMsiArchiver.Statuses.Archived, Retryable: false,
-            $"{Version}/IntuneWindowsAgent.msi", "abc123", 42L, ImeMsiArchiver.FallbackMsiUrls[1]);
-        harness.EnqueueMessage(JsonConvert.SerializeObject(Envelope()));
-
-        await harness.RunUntilAsync(() => harness.MainQueueDeleted());
-
-        harness.Repo.Verify(r => r.UpdateImeVersionArchiveInfoAsync(
-            Version, ImeMsiArchiver.Statuses.Archived,
-            $"{Version}/IntuneWindowsAgent.msi", "abc123", 42L, ImeMsiArchiver.FallbackMsiUrls[1]),
-            Times.Once);
-        harness.MainQueue.Verify(q => q.DeleteMessageAsync(
-            It.IsAny<string>(), It.IsAny<string>(), It.IsAny<CancellationToken>()),
-            Times.Once);
-    }
-
-    [Fact]
-    public async Task Worker_retryable_failure_merges_row_but_leaves_message()
-    {
-        var harness = new Harness();
-        harness.Archiver.Result = new ImeMsiArchiveResult(
-            false, ImeMsiArchiver.Statuses.FailedDownload, Retryable: true,
-            null, null, null, ImeMsiArchiver.FallbackMsiUrls[1]);
-        harness.EnqueueMessage(JsonConvert.SerializeObject(Envelope()));
-
-        await harness.RunUntilAsync(() => harness.Archiver.Calls > 0);
-
-        // Row always tells the truth — the failure status IS merged...
-        harness.Repo.Verify(r => r.UpdateImeVersionArchiveInfoAsync(
-            Version, ImeMsiArchiver.Statuses.FailedDownload,
-            null, null, null, ImeMsiArchiver.FallbackMsiUrls[1]),
-            Times.AtLeastOnce);
-        // ...but the message stays for the visibility-timeout retry → poison ladder.
-        harness.MainQueue.Verify(q => q.DeleteMessageAsync(
-            It.IsAny<string>(), It.IsAny<string>(), It.IsAny<CancellationToken>()),
-            Times.Never);
-    }
-
-    [Fact]
-    public async Task Worker_permanent_failure_merges_row_and_completes_message()
-    {
-        var harness = new Harness();
-        harness.Archiver.Result = new ImeMsiArchiveResult(
-            false, ImeMsiArchiver.Statuses.FailedTooLarge, Retryable: false,
-            null, null, 999L, ImeMsiArchiver.FallbackMsiUrls[1]);
-        harness.EnqueueMessage(JsonConvert.SerializeObject(Envelope()));
-
-        await harness.RunUntilAsync(() => harness.MainQueueDeleted());
-
-        harness.Repo.Verify(r => r.UpdateImeVersionArchiveInfoAsync(
-            Version, ImeMsiArchiver.Statuses.FailedTooLarge,
-            null, null, 999L, ImeMsiArchiver.FallbackMsiUrls[1]),
-            Times.Once);
-        harness.MainQueue.Verify(q => q.DeleteMessageAsync(
-            It.IsAny<string>(), It.IsAny<string>(), It.IsAny<CancellationToken>()),
-            Times.Once);
-    }
-
-    [Fact]
-    public async Task Worker_pauses_while_archiving_disabled()
-    {
-        var harness = new Harness(archivingEnabled: false);
-        harness.EnqueueMessage(JsonConvert.SerializeObject(Envelope()));
-
-        await harness.RunForAsync(TimeSpan.FromMilliseconds(500));
-
-        // Pause, not drop: no receive, no archive attempt — the message parks in the queue.
-        harness.MainQueue.Verify(q => q.ReceiveMessagesAsync(
-            It.IsAny<int>(), It.IsAny<TimeSpan>(), It.IsAny<CancellationToken>()),
-            Times.Never);
-        Assert.Equal(0, harness.Archiver.Calls);
-    }
-
-    [Fact]
-    public async Task Worker_drops_envelope_without_version_before_archiver()
-    {
-        var harness = new Harness();
-        harness.EnqueueMessage(JsonConvert.SerializeObject(new ImeMsiArchiveEnvelope
-        {
-            Version = string.Empty,
-            EnqueuedAt = DateTime.UtcNow,
-        }));
-
-        await harness.RunUntilAsync(() => harness.MainQueueDeleted());
-
-        Assert.Equal(0, harness.Archiver.Calls);
-        harness.Repo.Verify(r => r.UpdateImeVersionArchiveInfoAsync(
-            It.IsAny<string>(), It.IsAny<string>(), It.IsAny<string?>(),
-            It.IsAny<string?>(), It.IsAny<long?>(), It.IsAny<string?>()),
-            Times.Never);
-    }
-
-    // ============================================================ Helpers ====
-
-    private static ImeMsiArchiveEnvelope Envelope() => new()
-    {
-        Version = Version,
-        MsiDownloadUrl = ImeMsiArchiver.FallbackMsiUrls[1],
-        MsiMatchedBy = "productVersion",
-        TenantId = "11111111-1111-1111-1111-111111111111",
-        SessionId = "aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa",
-        EnqueuedAt = DateTime.UtcNow,
-    };
-
-    /// <summary>Scripted archiver — the worker only consumes the classified result.</summary>
-    private sealed class StubArchiver : ImeMsiArchiver
-    {
-        private int _calls;
-
-        public ImeMsiArchiveResult Result { get; set; } = new(
-            true, Statuses.Archived, Retryable: false, null, null, null, null);
-
-        public int Calls => Volatile.Read(ref _calls);
-
-        public override Task<ImeMsiArchiveResult> ArchiveAsync(
-            ImeMsiArchiveEnvelope envelope, CancellationToken cancellationToken = default)
-        {
-            Interlocked.Increment(ref _calls);
-            return Task.FromResult(Result);
-        }
-    }
-
-    /// <summary>Mock-queue poll-loop harness, same shape as SessionDeletionWorkerTests.</summary>
-    private sealed class Harness
-    {
-        public Mock<QueueClient> MainQueue { get; }
-        public Mock<QueueClient> PoisonQueue { get; }
-        public StubArchiver Archiver { get; } = new();
-        public Mock<ISessionRepository> Repo { get; } = new();
-        public ImeMsiArchiveQueueWorker Sut { get; }
-
-        private readonly Queue<QueueMessage> _pendingMessages = new();
-
-        public Harness(bool archivingEnabled = true)
-        {
-            MainQueue = new Mock<QueueClient>();
-            PoisonQueue = new Mock<QueueClient>();
-
-            MainQueue.Setup(q => q.CreateIfNotExistsAsync(
-                    It.IsAny<IDictionary<string, string>>(), It.IsAny<CancellationToken>()))
-                .ReturnsAsync((Response?)null);
-            PoisonQueue.Setup(q => q.CreateIfNotExistsAsync(
-                    It.IsAny<IDictionary<string, string>>(), It.IsAny<CancellationToken>()))
-                .ReturnsAsync((Response?)null);
-
-            MainQueue.Setup(q => q.ReceiveMessagesAsync(
-                    It.IsAny<int>(), It.IsAny<TimeSpan>(), It.IsAny<CancellationToken>()))
-                .Returns<int, TimeSpan?, CancellationToken>((maxMessages, _, _) =>
-                {
-                    var batch = new List<QueueMessage>();
-                    lock (_pendingMessages)
-                    {
-                        while (batch.Count < maxMessages && _pendingMessages.Count > 0)
-                            batch.Add(_pendingMessages.Dequeue());
-                    }
-                    return Task.FromResult(Response.FromValue(batch.ToArray(), new Mock<Response>().Object));
-                });
-
-            MainQueue.Setup(q => q.DeleteMessageAsync(
-                    It.IsAny<string>(), It.IsAny<string>(), It.IsAny<CancellationToken>()))
-                .ReturnsAsync(new Mock<Response>().Object);
-
-            MainQueue.Setup(q => q.UpdateMessageAsync(
-                    It.IsAny<string>(), It.IsAny<string>(),
-                    It.IsAny<string>(), It.IsAny<TimeSpan>(), It.IsAny<CancellationToken>()))
-                .Returns<string, string, string, TimeSpan, CancellationToken>((_, _, _, vis, _) =>
-                {
-                    var updated = QueuesModelFactory.UpdateReceipt(
-                        "pop-extended-" + Guid.NewGuid().ToString("N"),
-                        DateTimeOffset.UtcNow.Add(vis));
-                    return Task.FromResult(Response.FromValue(updated, new Mock<Response>().Object));
-                });
-
-            var adminConfig = new Mock<AdminConfigurationService>(
-                Mock.Of<IConfigRepository>(),
-                NullLogger<AdminConfigurationService>.Instance,
-                new MemoryCache(new MemoryCacheOptions()));
-            adminConfig.Setup(a => a.GetConfigurationAsync())
-                .ReturnsAsync(new AdminConfiguration { ImeMsiArchivingEnabled = archivingEnabled });
-
-            Sut = new ImeMsiArchiveQueueWorker(
-                MainQueue.Object, PoisonQueue.Object,
-                Archiver, Repo.Object, adminConfig.Object,
-                NullLogger<ImeMsiArchiveQueueWorker>.Instance,
-                pollInterval: TimeSpan.FromMilliseconds(50));
-        }
-
-        public void EnqueueMessage(string body)
-        {
-            var msg = QueuesModelFactory.QueueMessage(
-                messageId: "msg-" + Guid.NewGuid().ToString("N"),
-                popReceipt: "pop-" + Guid.NewGuid().ToString("N"),
-                body: new BinaryData(body),
-                dequeueCount: 1);
-            lock (_pendingMessages) _pendingMessages.Enqueue(msg);
-        }
-
-        public async Task RunForAsync(TimeSpan duration)
-        {
-            using var cts = new CancellationTokenSource(duration);
-            try { await Sut.StartAsync(cts.Token); }
-            catch (OperationCanceledException) { }
-            try { await Task.Delay(duration, cts.Token); }
-            catch (OperationCanceledException) { }
-            try { await Sut.StopAsync(CancellationToken.None); }
-            catch (OperationCanceledException) { }
-        }
-
-        public async Task RunUntilAsync(Func<bool> condition, TimeSpan? timeout = null)
-        {
-            var deadline = DateTime.UtcNow + (timeout ?? TimeSpan.FromSeconds(10));
-            using var cts = new CancellationTokenSource();
-            try { await Sut.StartAsync(cts.Token); }
-            catch (OperationCanceledException) { }
-            while (!condition() && DateTime.UtcNow < deadline)
-                await Task.Delay(25);
-            cts.Cancel();
-            try { await Sut.StopAsync(CancellationToken.None); }
-            catch (OperationCanceledException) { }
-        }
-
-        public bool MainQueueDeleted() =>
-            MainQueue.Invocations.Any(i => i.Method.Name == nameof(QueueClient.DeleteMessageAsync));
-    }
-}

src/Backend/AutopilotMonitor.Functions.Tests/NotificationDispatchHandlerTests.cs

@@ -0,0 +1,150 @@
+using System;
+using System.Collections.Generic;
+using System.Linq;
+using System.Threading;
+using System.Threading.Tasks;
+using AutopilotMonitor.Functions.Services;
+using AutopilotMonitor.Functions.Services.Notifications;
+using AutopilotMonitor.Shared.DataAccess;
+using AutopilotMonitor.Shared.Models;
+using AutopilotMonitor.Shared.Models.Notifications;
+using Microsoft.Extensions.Caching.Memory;
+using Microsoft.Extensions.Logging.Abstractions;
+using Moq;
+using Newtonsoft.Json;
+using Xunit;
+
+namespace AutopilotMonitor.Functions.Tests;
+
+/// <summary>
+/// The notification-dispatch consumer resolves channel ids against the tenant's CURRENT
+/// configuration — no URL or secret ever rides in the queue — and sends only to channels that
+/// are still enabled. Plus the wire contract: the envelope round-trips through Newtonsoft with
+/// the alert's facts, sections and severity intact.
+/// </summary>
+public class NotificationDispatchHandlerTests
+{
+    private const string Tenant = "11111111-1111-1111-1111-111111111111";
+
+    [Fact]
+    public async Task Sends_to_enabled_channels_matching_the_envelope_ids_only()
+    {
+        var h = new Harness(new[]
+        {
+            Channel("teams", enabled: true),
+            Channel("slack", enabled: true),
+            Channel("old", enabled: false),
+        });
+
+        await h.Sut.HandleAsync(Envelope("teams", "old"), CancellationToken.None);
+
+        var sent = Assert.Single(h.Sent);
+        Assert.Equal(new[] { "teams" }, sent.Channels.Select(c => c.Id));
+        Assert.Equal("enrollment_succeeded", sent.Alert.EventType);
+    }
+
+    [Fact]
+    public async Task Unknown_channel_ids_are_skipped_without_sending()
+    {
+        var h = new Harness(new[] { Channel("teams", enabled: true) });
+
+        await h.Sut.HandleAsync(Envelope("deleted-meanwhile"), CancellationToken.None);
+
+        Assert.Empty(h.Sent);
+    }
+
+    [Fact]
+    public async Task Missing_tenant_configuration_drops_the_alert()
+    {
+        var h = new Harness(channels: null);
+
+        await h.Sut.HandleAsync(Envelope("teams"), CancellationToken.None);
+
+        Assert.Empty(h.Sent);
+    }
+
+    [Fact]
+    public void Envelope_round_trips_through_newtonsoft()
+    {
+        var envelope = Envelope("teams");
+        envelope.Alert.Facts.Add(new NotificationFact { Name = "Device", Value = "PC-0001" });
+        envelope.Alert.Severity = NotificationSeverity.Error;
+
+        var restored = JsonConvert.DeserializeObject<NotificationDispatchEnvelope>(
+            JsonConvert.SerializeObject(envelope))!;
+
+        Assert.Equal(Tenant, restored.TenantId);
+        Assert.Equal(new[] { "teams" }, restored.ChannelIds);
+        Assert.Equal("enrollment_succeeded", restored.Alert.EventType);
+        Assert.Equal(NotificationSeverity.Error, restored.Alert.Severity);
+        var fact = Assert.Single(restored.Alert.Facts);
+        Assert.Equal("PC-0001", fact.Value);
+    }
+
+    // ============================================================ Helpers ====
+
+    private static NotificationChannel Channel(string id, bool enabled) => new()
+    {
+        Id = id,
+        Name = id,
+        ProviderType = (int)WebhookProviderType.Slack,
+        Url = $"https://example.invalid/{id}",
+        Enabled = enabled,
+        NotifyOnSuccess = true,
+    };
+
+    private static NotificationDispatchEnvelope Envelope(params string[] channelIds) => new()
+    {
+        TenantId = Tenant,
+        SessionId = "aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa",
+        ChannelIds = channelIds.ToList(),
+        Alert = new NotificationAlert
+        {
+            EventType = "enrollment_succeeded",
+            Title = "Enrollment Succeeded",
+            Summary = "PC-0001 enrolled",
+            Severity = NotificationSeverity.Info,
+            ThemeColor = "00AA00",
+        },
+        EnqueuedAt = DateTime.UtcNow,
+    };
+
+    private sealed class Harness
+    {
+        public List<(IReadOnlyList<NotificationChannel> Channels, NotificationAlert Alert)> Sent { get; } = new();
+        public NotificationDispatchHandler Sut { get; }
+
+        public Harness(IEnumerable<NotificationChannel>? channels)
+        {
+            var configService = new Mock<TenantConfigurationService>(
+                Mock.Of<IConfigRepository>(),
+                NullLogger<TenantConfigurationService>.Instance,
+                new MemoryCache(new MemoryCacheOptions()))
+            { CallBase = false };
+
+            if (channels is null)
+            {
+                configService.Setup(c => c.TryGetConfigurationAsync(Tenant))
+                    .ReturnsAsync((new TenantConfiguration { TenantId = Tenant }, false));
+            }
+            else
+            {
+                var config = new TenantConfiguration
+                {
+                    TenantId = Tenant,
+                    NotificationChannelsJson = NotificationChannel.SerializeList(channels),
+                };
+                configService.Setup(c => c.TryGetConfigurationAsync(Tenant))
+                    .ReturnsAsync((config, true));
+            }
+
+            var dispatcher = new Mock<NotificationChannelDispatcher>(null!, null!);
+            dispatcher.Setup(d => d.SendToChannelsAsync(It.IsAny<IEnumerable<NotificationChannel>>(), It.IsAny<NotificationAlert>()))
+                .Callback<IEnumerable<NotificationChannel>, NotificationAlert>((c, a) => Sent.Add((c.ToList(), a)))
+                .Returns(Task.CompletedTask);
+
+            Sut = new NotificationDispatchHandler(
+                configService.Object, dispatcher.Object, NullLogger<NotificationDispatchHandler>.Instance);
+        }
+    }
+}

src/Backend/AutopilotMonitor.Functions.Tests/QueueEnvelopeReaderTests.cs

@@ -0,0 +1,84 @@
+using System;
+using AutopilotMonitor.Functions.Services.Queueing;
+using Azure.Storage.Queues.Models;
+using Microsoft.Extensions.Logging.Abstractions;
+using Newtonsoft.Json;
+using Xunit;
+
+namespace AutopilotMonitor.Functions.Tests;
+
+/// <summary>
+/// The shared decode step of every [QueueTrigger] function: a <c>false</c> means the message is
+/// permanently unusable and the function returns so the host deletes it — never a throw, which
+/// would burn dequeue attempts on a message that can never succeed.
+/// </summary>
+public class QueueEnvelopeReaderTests
+{
+    private sealed class Envelope
+    {
+        public string TenantId { get; set; } = string.Empty;
+        public int Count { get; set; }
+    }
+
+    [Fact]
+    public void Valid_json_yields_envelope()
+    {
+        var msg = Message(JsonConvert.SerializeObject(new Envelope { TenantId = "t1", Count = 7 }));
+
+        var ok = QueueEnvelopeReader.TryRead<Envelope>(msg, NullLogger.Instance, out var envelope);
+
+        Assert.True(ok);
+        Assert.Equal("t1", envelope.TenantId);
+        Assert.Equal(7, envelope.Count);
+    }
+
+    [Fact]
+    public void Malformed_json_is_rejected_without_throwing()
+    {
+        var msg = Message("{ not json");
+
+        var ok = QueueEnvelopeReader.TryRead<Envelope>(msg, NullLogger.Instance, out var envelope);
+
+        Assert.False(ok);
+        Assert.Null(envelope);
+    }
+
+    [Fact]
+    public void Null_literal_is_rejected()
+    {
+        var msg = Message("null");
+
+        var ok = QueueEnvelopeReader.TryRead<Envelope>(msg, NullLogger.Instance, out _);
+
+        Assert.False(ok);
+    }
+
+    [Fact]
+    public void Failed_validation_is_rejected()
+    {
+        var msg = Message(JsonConvert.SerializeObject(new Envelope { TenantId = string.Empty }));
+
+        var ok = QueueEnvelopeReader.TryRead<Envelope>(
+            msg, NullLogger.Instance, out _, validate: e => !string.IsNullOrEmpty(e.TenantId));
+
+        Assert.False(ok);
+    }
+
+    [Fact]
+    public void Validation_passes_when_predicate_holds()
+    {
+        var msg = Message(JsonConvert.SerializeObject(new Envelope { TenantId = "t1" }));
+
+        var ok = QueueEnvelopeReader.TryRead<Envelope>(
+            msg, NullLogger.Instance, out var envelope, validate: e => !string.IsNullOrEmpty(e.TenantId));
+
+        Assert.True(ok);
+        Assert.Equal("t1", envelope.TenantId);
+    }
+
+    private static QueueMessage Message(string body) => QueuesModelFactory.QueueMessage(
+        messageId: "msg-" + Guid.NewGuid().ToString("N"),
+        popReceipt: "pop",
+        body: new BinaryData(body),
+        dequeueCount: 1);
+}

src/Backend/AutopilotMonitor.Functions.Tests/QueuePollingWorkerBatchSizeTests.cs

@@ -0,0 +1,107 @@
+using System;
+using System.Collections.Generic;
+using System.Linq;
+using System.Threading;
+using System.Threading.Tasks;
+using AutopilotMonitor.Functions.Services.Queueing;
+using Azure;
+using Azure.Storage.Queues;
+using Azure.Storage.Queues.Models;
+using Microsoft.Extensions.Logging.Abstractions;
+using Moq;
+using Newtonsoft.Json;
+using Xunit;
+
+namespace AutopilotMonitor.Functions.Tests;
+
+/// <summary>
+/// Audit 2026-09-05 F02: the self-managed poll loop processes a batch sequentially and the
+/// heartbeat only covers the in-flight message, so every message received beyond the first sat
+/// invisible with its clock running and could reappear for another worker before this loop
+/// reached it. The base therefore receives ONE message per poll and re-receives immediately
+/// after a non-empty result — no worker may inherit a larger batch by accident.
+/// </summary>
+public class QueuePollingWorkerBatchSizeTests
+{
+    private sealed class Envelope
+    {
+        public int Seq { get; set; }
+    }
+
+    private sealed class ProbeWorker : QueuePollingWorker<Envelope>
+    {
+        public List<int> Handled { get; } = new();
+
+        public ProbeWorker(QueueClient main, QueueClient poison)
+            : base(main, poison, NullLogger.Instance, pollIntervalOverride: TimeSpan.FromMilliseconds(20))
+        {
+        }
+
+        public int ExposedBatchSize => BatchSize;
+
+        protected override Task HandleAsync(Envelope envelope, CancellationToken ct)
+        {
+            lock (Handled) Handled.Add(envelope.Seq);
+            return Task.CompletedTask;
+        }
+    }
+
+    [Fact]
+    public void Default_batch_size_is_one()
+    {
+        var sut = new ProbeWorker(new Mock<QueueClient>().Object, new Mock<QueueClient>().Object);
+        Assert.Equal(1, sut.ExposedBatchSize);
+    }
+
+    [Fact]
+    public async Task Loop_receives_one_message_at_a_time_and_drains_all_pending()
+    {
+        var pending = new Queue<QueueMessage>();
+        for (var i = 1; i <= 3; i++)
+        {
+            pending.Enqueue(QueuesModelFactory.QueueMessage(
+                messageId: "msg-" + i,
+                popReceipt: "pop-" + i,
+                body: new BinaryData(JsonConvert.SerializeObject(new Envelope { Seq = i })),
+                dequeueCount: 1));
+        }
+
+        var requestedBatchSizes = new List<int>();
+        var main = new Mock<QueueClient>();
+        var poison = new Mock<QueueClient>();
+        main.Setup(q => q.CreateIfNotExistsAsync(It.IsAny<IDictionary<string, string>>(), It.IsAny<CancellationToken>()))
+            .ReturnsAsync((Response?)null);
+        poison.Setup(q => q.CreateIfNotExistsAsync(It.IsAny<IDictionary<string, string>>(), It.IsAny<CancellationToken>()))
+            .ReturnsAsync((Response?)null);
+        main.Setup(q => q.ReceiveMessagesAsync(It.IsAny<int>(), It.IsAny<TimeSpan>(), It.IsAny<CancellationToken>()))
+            .Returns<int, TimeSpan?, CancellationToken>((max, _, _) =>
+            {
+                var batch = new List<QueueMessage>();
+                lock (pending)
+                {
+                    requestedBatchSizes.Add(max);
+                    while (batch.Count < max && pending.Count > 0)
+                        batch.Add(pending.Dequeue());
+                }
+                return Task.FromResult(Response.FromValue(batch.ToArray(), new Mock<Response>().Object));
+            });
+        main.Setup(q => q.DeleteMessageAsync(It.IsAny<string>(), It.IsAny<string>(), It.IsAny<CancellationToken>()))
+            .ReturnsAsync(new Mock<Response>().Object);
+
+        var sut = new ProbeWorker(main.Object, poison.Object);
+        using var cts = new CancellationTokenSource();
+        await sut.StartAsync(cts.Token);
+        var deadline = DateTime.UtcNow + TimeSpan.FromSeconds(10);
+        while (DateTime.UtcNow < deadline)
+        {
+            lock (sut.Handled) if (sut.Handled.Count == 3) break;
+            await Task.Delay(20);
+        }
+        cts.Cancel();
+        await sut.StopAsync(CancellationToken.None);
+
+        Assert.Equal(new[] { 1, 2, 3 }, sut.Handled);
+        Assert.NotEmpty(requestedBatchSizes);
+        Assert.All(requestedBatchSizes, n => Assert.Equal(1, n));
+    }
+}

src/Backend/AutopilotMonitor.Functions.Tests/RuleEngineFailurePropagationTests.cs

@@ -10,7 +10,7 @@ namespace AutopilotMonitor.Functions.Tests;
 /// Pins the queue-path retry contract: <see cref="RuleEngine.AnalyzeSessionAsync"/> must
 /// propagate storage exceptions to the caller. The previous outer try/catch swallowed all
 /// errors and returned an empty <see cref="AnalysisOutcome"/>, which caused
-/// <c>AnalyzeOnEnrollmentEndQueueWorker</c> to delete the message as success and never retry —
+/// <c>AnalyzeOnEnrollmentEndQueueFunction</c> to delete the message as success and never retry —
 /// rule results were silently lost on transient Table Storage failures.
 /// </summary>
 public class RuleEngineFailurePropagationTests

src/Backend/AutopilotMonitor.Functions.Tests/VulnerabilityCorrelateEnvelopeSerializationTests.cs

@@ -6,7 +6,7 @@ namespace AutopilotMonitor.Functions.Tests;
 /// <summary>
 /// JSON round-trip tests for <see cref="VulnerabilityCorrelateEnvelope"/>. The producer
 /// (<c>AzureQueueVulnerabilityCorrelateProducer</c>) serializes with Newtonsoft and the
-/// queue worker (<c>VulnerabilityCorrelateQueueWorker</c>) deserializes the same shape —
+/// queue worker (<c>VulnerabilityCorrelateQueueFunction</c>) deserializes the same shape —
 /// pinning the contract here keeps the queue from silently breaking when fields are added.
 /// </summary>
 public class VulnerabilityCorrelateEnvelopeSerializationTests

src/Backend/AutopilotMonitor.Functions/Functions/Ingest/ReportDistressFunction.cs

@@ -37,7 +37,7 @@ namespace AutopilotMonitor.Functions.Functions.Ingest
         private readonly IDistressReportRepository _repository;
         private readonly TelemetryClient _telemetryClient;
         private readonly HardwareRejectionThrottleService _hardwareThrottle;
-        private readonly NotificationChannelDispatcher _channelDispatcher;
+        private readonly INotificationDispatchProducer _notificationProducer;
         private readonly IHardwareRejectionNotificationTracker _hardwareBellTracker;
         private readonly TenantNotificationService _tenantNotificationService;
 
@@ -77,7 +77,7 @@ namespace AutopilotMonitor.Functions.Functions.Ingest
             IDistressReportRepository repository,
             TelemetryClient telemetryClient,
             HardwareRejectionThrottleService hardwareThrottle,
-            NotificationChannelDispatcher channelDispatcher,
+            INotificationDispatchProducer notificationProducer,
             IHardwareRejectionNotificationTracker hardwareBellTracker,
             TenantNotificationService tenantNotificationService)
         {
@@ -87,7 +87,7 @@ namespace AutopilotMonitor.Functions.Functions.Ingest
             _repository = repository;
             _telemetryClient = telemetryClient;
             _hardwareThrottle = hardwareThrottle;
-            _channelDispatcher = channelDispatcher;
+            _notificationProducer = notificationProducer;
             _hardwareBellTracker = hardwareBellTracker;
             _tenantNotificationService = tenantNotificationService;
         }
@@ -217,7 +217,9 @@ namespace AutopilotMonitor.Functions.Functions.Ingest
 
                 await _repository.SaveDistressReportAsync(tenantId, entry);
 
-                // Notify tenant webhook for hardware whitelist rejections (fire-and-forget, opt-in)
+                // Notify tenant channels for hardware whitelist rejections (opt-in). Durable via
+                // the notification-dispatch queue — a fire-and-forget send after the response was
+                // not counted as open work by the host and could be dropped on scale-in.
                 if (report.ErrorType == DistressErrorType.HardwareNotAllowed
                     && _hardwareThrottle.ShouldNotify(tenantId, manufacturer, model))
                 {
@@ -232,7 +234,13 @@ namespace AutopilotMonitor.Functions.Functions.Ingest
                             if (hwChannels.Count > 0)
                             {
                                 var alert = NotificationAlertBuilder.BuildHardwareRejectedAlert(manufacturer, model, serialNumber);
-                                _ = _channelDispatcher.SendToChannelsAsync(hwChannels, alert);
+                                await _notificationProducer.EnqueueAsync(new NotificationDispatchEnvelope
+                                {
+                                    TenantId = tenantId,
+                                    ChannelIds = hwChannels.Select(c => c.Id).ToList(),
+                                    Alert = alert,
+                                    EnqueuedAt = DateTime.UtcNow,
+                                });
                             }
                         }
                     }

src/Backend/AutopilotMonitor.Functions/Functions/Queue/AnalyzeOnEnrollmentEndQueueFunction.cs

@@ -0,0 +1,47 @@
+using System.Threading;
+using System.Threading.Tasks;
+using AutopilotMonitor.Functions.Services.Analyze;
+using AutopilotMonitor.Functions.Services.Queueing;
+using AutopilotMonitor.Shared;
+using AutopilotMonitor.Shared.Models;
+using Azure.Storage.Queues.Models;
+using Microsoft.Azure.Functions.Worker;
+using Microsoft.Extensions.Logging;
+
+namespace AutopilotMonitor.Functions.Functions.Queue
+{
+    /// <summary>
+    /// Host-managed consumer of the <c>analyze-on-enrollment-end</c> queue. The Functions host
+    /// owns the lease (renewed while the handler runs), the retry ladder (throw ⇒ redelivery after
+    /// <c>extensions.queues.visibilityTimeout</c>) and the poison move after
+    /// <c>maxDequeueCount</c> attempts to <c>analyze-on-enrollment-end-poison</c>. On Flex
+    /// Consumption the queue length is the scale signal, so a message enqueued while no instance
+    /// is running wakes one — the property the earlier self-polling BackgroundService lacked.
+    /// The rule-engine work lives unchanged in <see cref="AnalyzeOnEnrollmentEndHandler"/>.
+    /// </summary>
+    public sealed class AnalyzeOnEnrollmentEndQueueFunction
+    {
+        private readonly AnalyzeOnEnrollmentEndHandler _handler;
+        private readonly ILogger<AnalyzeOnEnrollmentEndQueueFunction> _logger;
+
+        public AnalyzeOnEnrollmentEndQueueFunction(
+            AnalyzeOnEnrollmentEndHandler handler,
+            ILogger<AnalyzeOnEnrollmentEndQueueFunction> logger)
+        {
+            _handler = handler;
+            _logger = logger;
+        }
+
+        [Function("AnalyzeOnEnrollmentEndQueue")]
+        public Task Run(
+            [QueueTrigger(Constants.QueueNames.AnalyzeOnEnrollmentEnd, Connection = Constants.QueueNames.TriggerConnection)]
+            QueueMessage message,
+            CancellationToken cancellationToken)
+        {
+            if (!QueueEnvelopeReader.TryRead<AnalyzeOnEnrollmentEndEnvelope>(message, _logger, out var envelope))
+                return Task.CompletedTask;
+
+            return _handler.HandleAsync(envelope, cancellationToken);
+        }
+    }
+}

src/Backend/AutopilotMonitor.Functions/Functions/Queue/ImeMsiArchiveQueueFunction.cs

@@ -0,0 +1,115 @@
+using System;
+using System.Threading;
+using System.Threading.Tasks;
+using AutopilotMonitor.Functions.Services;
+using AutopilotMonitor.Functions.Services.Ime;
+using AutopilotMonitor.Functions.Services.Queueing;
+using AutopilotMonitor.Shared;
+using AutopilotMonitor.Shared.DataAccess;
+using AutopilotMonitor.Shared.Models;
+using Azure.Storage.Queues.Models;
+using Microsoft.Azure.Functions.Worker;
+using Microsoft.Extensions.Logging;
+
+namespace AutopilotMonitor.Functions.Functions.Queue
+{
+    /// <summary>
+    /// Host-managed consumer of the <c>ime-msi-archive</c> queue: downloads a newly sighted IME
+    /// installer into the permanent <c>ime-archive</c> container and merges the outcome onto the
+    /// ImeVersionHistory row. Volume is ~one message per Microsoft IME release.
+    /// <para>
+    /// Retry semantics: the archiver never throws — it classifies. Permanent rejections (bad
+    /// version, size cap) complete the message with their <c>Failed:*</c> status persisted;
+    /// transient failures (download/timeout) throw so the host's retry → poison ladder applies,
+    /// with the last status visible on the row. The host renews the message lease for the
+    /// up-to-10-minute download window. The status row-merge happens on EVERY attempt so
+    /// operators see a stuck download as <c>Failed:Download</c> instead of nothing.
+    /// </para>
+    /// <para>
+    /// <b>Pause</b> (<see cref="AdminConfiguration.ImeMsiArchivingEnabled"/> off): a trigger
+    /// cannot leave a message untouched, so the envelope is re-enqueued with
+    /// <see cref="PauseRequeueDelay"/> and the current message completes. The message parks in
+    /// the queue until the flag returns; its dequeue count starts over, which is right for a
+    /// pause — it is not a failure. The re-enqueue is fail-soft like every other archive step;
+    /// operator-side blob backfill remains the fallback.
+    /// </para>
+    /// </summary>
+    public sealed class ImeMsiArchiveQueueFunction
+    {
+        /// <summary>How long a parked message stays invisible while archiving is switched off.</summary>
+        internal static readonly TimeSpan PauseRequeueDelay = TimeSpan.FromMinutes(10);
+
+        private readonly ImeMsiArchiver _archiver;
+        private readonly ISessionRepository _sessionRepo;
+        private readonly AdminConfigurationService _adminConfigService;
+        private readonly IImeMsiArchiveProducer _producer;
+        private readonly ILogger<ImeMsiArchiveQueueFunction> _logger;
+
+        public ImeMsiArchiveQueueFunction(
+            ImeMsiArchiver archiver,
+            ISessionRepository sessionRepo,
+            AdminConfigurationService adminConfigService,
+            IImeMsiArchiveProducer producer,
+            ILogger<ImeMsiArchiveQueueFunction> logger)
+        {
+            _archiver = archiver ?? throw new ArgumentNullException(nameof(archiver));
+            _sessionRepo = sessionRepo ?? throw new ArgumentNullException(nameof(sessionRepo));
+            _adminConfigService = adminConfigService ?? throw new ArgumentNullException(nameof(adminConfigService));
+            _producer = producer ?? throw new ArgumentNullException(nameof(producer));
+            _logger = logger ?? throw new ArgumentNullException(nameof(logger));
+        }
+
+        [Function("ImeMsiArchiveQueue")]
+        public Task Run(
+            [QueueTrigger(Constants.QueueNames.ImeMsiArchive, Connection = Constants.QueueNames.TriggerConnection)]
+            QueueMessage message,
+            CancellationToken cancellationToken)
+            => ProcessAsync(message, cancellationToken);
+
+        /// <summary>Testable core — the trigger entry above only forwards.</summary>
+        internal async Task ProcessAsync(QueueMessage message, CancellationToken cancellationToken)
+        {
+            if (!QueueEnvelopeReader.TryRead<ImeMsiArchiveEnvelope>(
+                    message, _logger, out var envelope,
+                    validate: e => !string.IsNullOrEmpty(e.Version)))
+                return;
+
+            if (!await IsArchivingEnabledAsync().ConfigureAwait(false))
+            {
+                _logger.LogWarning(
+                    "ImeMsiArchive paused (ImeMsiArchivingEnabled=false) — re-enqueuing version {Version} with {Delay} delay",
+                    envelope.Version, PauseRequeueDelay);
+                await _producer.EnqueueAsync(envelope, PauseRequeueDelay, cancellationToken).ConfigureAwait(false);
+                return;
+            }
+
+            var result = await _archiver.ArchiveAsync(envelope, cancellationToken).ConfigureAwait(false);
+
+            // Merge the outcome first — also for failures, so the row always tells the truth.
+            await _sessionRepo.UpdateImeVersionArchiveInfoAsync(
+                envelope.Version, result.Status, result.BlobPath, result.Sha256, result.SizeBytes, result.SourceUrl)
+                .ConfigureAwait(false);
+
+            if (!result.Success && result.Retryable)
+            {
+                // Host retry → poison after maxDequeueCount. The thrown message lands in the
+                // function's failure log line.
+                throw new InvalidOperationException(
+                    $"IME MSI archive attempt failed with {result.Status} for version {envelope.Version} (dequeue {message.DequeueCount})");
+            }
+        }
+
+        private async Task<bool> IsArchivingEnabledAsync()
+        {
+            try
+            {
+                var config = await _adminConfigService.GetConfigurationAsync().ConfigureAwait(false);
+                return config.ImeMsiArchivingEnabled;
+            }
+            catch
+            {
+                return true; // config unreadable → keep working (archive is fail-soft anyway)
+            }
+        }
+    }
+}

src/Backend/AutopilotMonitor.Functions/Functions/Queue/IndexReconcileQueueFunction.cs

@@ -0,0 +1,44 @@
+using System.Threading;
+using System.Threading.Tasks;
+using AutopilotMonitor.Functions.Services.Indexing;
+using AutopilotMonitor.Functions.Services.Queueing;
+using AutopilotMonitor.Shared;
+using AutopilotMonitor.Shared.Models;
+using Azure.Storage.Queues.Models;
+using Microsoft.Azure.Functions.Worker;
+using Microsoft.Extensions.Logging;
+
+namespace AutopilotMonitor.Functions.Functions.Queue
+{
+    /// <summary>
+    /// Host-managed consumer of the <c>telemetry-index-reconcile</c> queue (Plan §M5.d.3; lease,
+    /// retry ladder and poison move owned by the Functions host — see
+    /// <see cref="AnalyzeOnEnrollmentEndQueueFunction"/>). Index writes live unchanged in
+    /// <see cref="IndexReconcileHandler"/>; the 2h <c>IndexReconcileTimer</c> remains the safety net.
+    /// </summary>
+    public sealed class IndexReconcileQueueFunction
+    {
+        private readonly IndexReconcileHandler _handler;
+        private readonly ILogger<IndexReconcileQueueFunction> _logger;
+
+        public IndexReconcileQueueFunction(
+            IndexReconcileHandler handler,
+            ILogger<IndexReconcileQueueFunction> logger)
+        {
+            _handler = handler;
+            _logger = logger;
+        }
+
+        [Function("IndexReconcileQueue")]
+        public Task Run(
+            [QueueTrigger(Constants.QueueNames.TelemetryIndexReconcile, Connection = Constants.QueueNames.TriggerConnection)]
+            QueueMessage message,
+            CancellationToken cancellationToken)
+        {
+            if (!QueueEnvelopeReader.TryRead<IndexReconcileEnvelope>(message, _logger, out var envelope))
+                return Task.CompletedTask;
+
+            return _handler.HandleAsync(envelope, cancellationToken);
+        }
+    }
+}

src/Backend/AutopilotMonitor.Functions/Functions/Queue/NotificationDispatchQueueFunction.cs

@@ -0,0 +1,45 @@
+using System.Threading;
+using System.Threading.Tasks;
+using AutopilotMonitor.Functions.Services.Notifications;
+using AutopilotMonitor.Functions.Services.Queueing;
+using AutopilotMonitor.Shared;
+using AutopilotMonitor.Shared.Models.Notifications;
+using Azure.Storage.Queues.Models;
+using Microsoft.Azure.Functions.Worker;
+using Microsoft.Extensions.Logging;
+
+namespace AutopilotMonitor.Functions.Functions.Queue
+{
+    /// <summary>
+    /// Host-managed consumer of the <c>notification-dispatch</c> queue (lease, retry ladder and
+    /// poison move owned by the Functions host — see <see cref="AnalyzeOnEnrollmentEndQueueFunction"/>).
+    /// Channel resolution and the send live in <see cref="NotificationDispatchHandler"/>.
+    /// </summary>
+    public sealed class NotificationDispatchQueueFunction
+    {
+        private readonly NotificationDispatchHandler _handler;
+        private readonly ILogger<NotificationDispatchQueueFunction> _logger;
+
+        public NotificationDispatchQueueFunction(
+            NotificationDispatchHandler handler,
+            ILogger<NotificationDispatchQueueFunction> logger)
+        {
+            _handler = handler;
+            _logger = logger;
+        }
+
+        [Function("NotificationDispatchQueue")]
+        public Task Run(
+            [QueueTrigger(Constants.QueueNames.NotificationDispatch, Connection = Constants.QueueNames.TriggerConnection)]
+            QueueMessage message,
+            CancellationToken cancellationToken)
+        {
+            if (!QueueEnvelopeReader.TryRead<NotificationDispatchEnvelope>(
+                    message, _logger, out var envelope,
+                    validate: e => !string.IsNullOrEmpty(e.TenantId) && e.Alert is not null))
+                return Task.CompletedTask;
+
+            return _handler.HandleAsync(envelope, cancellationToken);
+        }
+    }
+}

src/Backend/AutopilotMonitor.Functions/Functions/Queue/TenantAutoApproveQueueFunction.cs

@@ -0,0 +1,45 @@
+using System.Threading;
+using System.Threading.Tasks;
+using AutopilotMonitor.Functions.Services.Activation;
+using AutopilotMonitor.Functions.Services.Queueing;
+using AutopilotMonitor.Shared;
+using Azure.Storage.Queues.Models;
+using Microsoft.Azure.Functions.Worker;
+using Microsoft.Extensions.Logging;
+
+namespace AutopilotMonitor.Functions.Functions.Queue
+{
+    /// <summary>
+    /// Host-managed consumer of the <c>tenant-auto-approve</c> queue (lease, retry ladder and
+    /// poison move owned by the Functions host — see <see cref="AnalyzeOnEnrollmentEndQueueFunction"/>).
+    /// The producer's ~1-minute visibility delay still applies; every decision lives in
+    /// <see cref="TenantAutoApproveHandler"/>.
+    /// </summary>
+    public sealed class TenantAutoApproveQueueFunction
+    {
+        private readonly TenantAutoApproveHandler _handler;
+        private readonly ILogger<TenantAutoApproveQueueFunction> _logger;
+
+        public TenantAutoApproveQueueFunction(
+            TenantAutoApproveHandler handler,
+            ILogger<TenantAutoApproveQueueFunction> logger)
+        {
+            _handler = handler;
+            _logger = logger;
+        }
+
+        [Function("TenantAutoApproveQueue")]
+        public Task Run(
+            [QueueTrigger(Constants.QueueNames.TenantAutoApprove, Connection = Constants.QueueNames.TriggerConnection)]
+            QueueMessage message,
+            CancellationToken cancellationToken)
+        {
+            if (!QueueEnvelopeReader.TryRead<TenantAutoApproveEnvelope>(
+                    message, _logger, out var envelope,
+                    validate: e => !string.IsNullOrWhiteSpace(e.TenantId)))
+                return Task.CompletedTask;
+
+            return _handler.HandleAsync(envelope, cancellationToken);
+        }
+    }
+}

src/Backend/AutopilotMonitor.Functions/Functions/Queue/VulnerabilityCorrelateQueueFunction.cs

@@ -0,0 +1,43 @@
+using System.Threading;
+using System.Threading.Tasks;
+using AutopilotMonitor.Functions.Services.Queueing;
+using AutopilotMonitor.Functions.Services.Vulnerability;
+using AutopilotMonitor.Shared;
+using AutopilotMonitor.Shared.Models;
+using Azure.Storage.Queues.Models;
+using Microsoft.Azure.Functions.Worker;
+using Microsoft.Extensions.Logging;
+
+namespace AutopilotMonitor.Functions.Functions.Queue
+{
+    /// <summary>
+    /// Host-managed consumer of the <c>vulnerability-correlate</c> queue (lease, retry ladder and
+    /// poison move owned by the Functions host — see <see cref="AnalyzeOnEnrollmentEndQueueFunction"/>).
+    /// Correlation itself lives unchanged in <see cref="VulnerabilityCorrelateHandler"/>.
+    /// </summary>
+    public sealed class VulnerabilityCorrelateQueueFunction
+    {
+        private readonly VulnerabilityCorrelateHandler _handler;
+        private readonly ILogger<VulnerabilityCorrelateQueueFunction> _logger;
+
+        public VulnerabilityCorrelateQueueFunction(
+            VulnerabilityCorrelateHandler handler,
+            ILogger<VulnerabilityCorrelateQueueFunction> logger)
+        {
+            _handler = handler;
+            _logger = logger;
+        }
+
+        [Function("VulnerabilityCorrelateQueue")]
+        public Task Run(
+            [QueueTrigger(Constants.QueueNames.VulnerabilityCorrelate, Connection = Constants.QueueNames.TriggerConnection)]
+            QueueMessage message,
+            CancellationToken cancellationToken)
+        {
+            if (!QueueEnvelopeReader.TryRead<VulnerabilityCorrelateEnvelope>(message, _logger, out var envelope))
+                return Task.CompletedTask;
+
+            return _handler.HandleAsync(envelope, cancellationToken);
+        }
+    }
+}

src/Backend/AutopilotMonitor.Functions/Program.cs

@@ -322,32 +322,27 @@ builder.Services.AddSingleton<
 builder.Services.AddSingleton<
     AutopilotMonitor.Functions.Services.Indexing.IndexReconcileHandler>();
 
-// Background poll-loop for the telemetry-index-reconcile queue (Plan §M5.d.3). Replaces the
-// earlier QueueTrigger function, which required a Functions-host-specific
-// `<Connection>__queueServiceUri` app-setting that diverged from the rest of the project's
-// AzureStorageAccountName + DefaultAzureCredential pattern. This worker uses the same
-// resolution as AzureQueueIndexReconcileProducer — Managed Identity by account name, with
-// connection-string fallback — and provides full QueueTrigger parity (visibility-timeout
-// retries, poison-queue move after 5 failed attempts).
-builder.Services.AddHostedService<
-    AutopilotMonitor.Functions.Services.Indexing.IndexReconcileQueueWorker>();
+// Consumer: Functions/Queue/IndexReconcileQueueFunction — a [QueueTrigger] on the
+// Constants.QueueNames.TriggerConnection connection (host-resolved app setting
+// `DataStorage__queueServiceUri`; see host.json extensions.queues). Stateless queue consumers
+// are host-managed so the Flex scale controller wakes an instance for a message and renews the
+// lease; only the cascade workers with lease/CAS state machines stay BackgroundServices.
 
 // Auto-analyze fan-out at session end. Replaces the previous in-function fire-and-forget
 // Task.Run that ran the rule engine after enrollment_complete / enrollment_failed / async
 // vulnerability correlation — Functions scale-in could kill the Task.Run mid-flight, leaving
 // rule results un-persisted (manual "Analyze Now" was the only recovery). Same producer +
-// worker pattern as IndexReconcile above.
+// handler pattern as IndexReconcile above; consumer = Functions/Queue/AnalyzeOnEnrollmentEndQueueFunction.
 builder.Services.AddSingleton<
     AutopilotMonitor.Functions.Services.Analyze.IAnalyzeOnEnrollmentEndProducer,
     AutopilotMonitor.Functions.Services.Analyze.AzureQueueAnalyzeOnEnrollmentEndProducer>();
 builder.Services.AddSingleton<
     AutopilotMonitor.Functions.Services.Analyze.AnalyzeOnEnrollmentEndHandler>();
-builder.Services.AddHostedService<
-    AutopilotMonitor.Functions.Services.Analyze.AnalyzeOnEnrollmentEndQueueWorker>();
 
 // Vulnerability-correlate fan-out triggered by the shutdown software_inventory_analysis
 // event. Replaces the in-function fire-and-forget Task.Run inside EventIngestProcessor.
-// Same producer + worker pattern as the analyze queue. Inventory loader is DI-shared with
+// Same producer + handler pattern as the analyze queue; consumer =
+// Functions/Queue/VulnerabilityCorrelateQueueFunction. Inventory loader is DI-shared with
 // the manual rescan endpoint (GetVulnerabilityReportFunction ?rescan=true).
 builder.Services.AddSingleton<
     AutopilotMonitor.Functions.Services.Vulnerability.IVulnerabilityCorrelateProducer,
@@ -357,13 +352,12 @@ builder.Services.AddSingleton<
     AutopilotMonitor.Functions.Services.Vulnerability.VulnerabilityInventoryLoader>();
 builder.Services.AddSingleton<
     AutopilotMonitor.Functions.Services.Vulnerability.VulnerabilityCorrelateHandler>();
-builder.Services.AddHostedService<
-    AutopilotMonitor.Functions.Services.Vulnerability.VulnerabilityCorrelateQueueWorker>();
 
 // IME-installer archiving on first fleet-wide sighting of a new IME version (~monthly).
-// Producer sits in EventIngestProcessor's new-version continuation; the worker downloads
-// the MSI from the CSP-reported (allowlisted) URL into the permanent ime-archive container
-// and merges the outcome onto the ImeVersionHistory row. Typed HttpClient: no redirects
+// Producer sits in EventIngestProcessor's new-version continuation; the consumer
+// (Functions/Queue/ImeMsiArchiveQueueFunction) downloads the MSI from the CSP-reported
+// (allowlisted) URL into the permanent ime-archive container and merges the outcome onto the
+// ImeVersionHistory row. Typed HttpClient: no redirects
 // (SSRF posture — the allowlisted host must serve directly), long timeout owned by the
 // archiver's own CTS, transient retry via the shared external-data policy.
 builder.Services.AddHttpClient<AutopilotMonitor.Functions.Services.Ime.ImeMsiArchiver>()
@@ -377,20 +371,17 @@ builder.Services.AddHttpClient<AutopilotMonitor.Functions.Services.Ime.ImeMsiArc
 builder.Services.AddSingleton<
     AutopilotMonitor.Functions.Services.Ime.IImeMsiArchiveProducer,
     AutopilotMonitor.Functions.Services.Ime.AzureQueueImeMsiArchiveProducer>();
-builder.Services.AddHostedService<
-    AutopilotMonitor.Functions.Services.Ime.ImeMsiArchiveQueueWorker>();
 
 // Delayed tenant auto-approve (public availability). Producer fires at first-login signup
-// with a ~1-minute visibility delay; the worker activates the tenant via the shared
-// TenantApprovalService only if AdminConfiguration.AutoApproveNewTenants is enabled at
-// processing time. Same producer + worker pattern as the analyze queue above.
+// with a ~1-minute visibility delay; the consumer (Functions/Queue/TenantAutoApproveQueueFunction)
+// activates the tenant via the shared TenantApprovalService only if
+// AdminConfiguration.AutoApproveNewTenants is enabled at processing time. Same producer +
+// handler pattern as the analyze queue above.
 builder.Services.AddSingleton<
     AutopilotMonitor.Functions.Services.Activation.ITenantAutoApproveEnqueuer,
     AutopilotMonitor.Functions.Services.Activation.AzureQueueTenantAutoApproveEnqueuer>();
 builder.Services.AddSingleton<
     AutopilotMonitor.Functions.Services.Activation.TenantAutoApproveHandler>();
-builder.Services.AddHostedService<
-    AutopilotMonitor.Functions.Services.Activation.TenantAutoApproveQueueWorker>();
 
 // ── Critical-Table Backup (plan §PR1) ─────────────────────────────────────────
 // Lease-aware service registered separately from the timer/worker so the lease
@@ -406,8 +397,9 @@ builder.Services.AddSingleton<AutopilotMonitor.Functions.Services.Backup.BackupJ
 builder.Services.AddSingleton<
     AutopilotMonitor.Functions.Services.Backup.Queue.ICriticalTableBackupProducer,
     AutopilotMonitor.Functions.Services.Backup.Queue.AzureQueueCriticalTableBackupProducer>();
-// Worker = BackgroundService, not QueueTrigger (matches AnalyzeOnEnrollmentEndQueueWorker
-// pattern per plan §Wave4 #1 — avoids the QueueTrigger-specific connection app-setting).
+// Worker = self-managed BackgroundService, not QueueTrigger: the lease/CAS state machine
+// (Running→Completed under lease, Failed only on poison-move) is a per-message lifecycle the
+// host cannot express. See QueuePollingWorkerBase for the two-class rule.
 builder.Services.AddHostedService<
     AutopilotMonitor.Functions.Services.Backup.Queue.CriticalTableBackupQueueWorker>();
 // PR2: single-row restore. Both validators are pure / I/O-only; the restore
@@ -469,6 +461,14 @@ builder.Services.AddHttpClient<TelegramNotificationService>()
 // Channel-level send API — routes each NotificationChannel to its transport (webhook renderer
 // vs. the platform Telegram bot). Transient: both transports are typed HttpClients.
 builder.Services.AddTransient<AutopilotMonitor.Functions.Services.Notifications.NotificationChannelDispatcher>();
+// Durable channel notifications from the agent hot paths (ingest terminal alerts, hardware
+// rejection): the producer enqueues channel ids + alert, Functions/Queue/NotificationDispatchQueueFunction
+// re-resolves the channels from the tenant config and sends. Handler is transient because the
+// dispatcher it wraps is.
+builder.Services.AddSingleton<
+    AutopilotMonitor.Functions.Services.Notifications.INotificationDispatchProducer,
+    AutopilotMonitor.Functions.Services.Notifications.AzureQueueNotificationDispatchProducer>();
+builder.Services.AddTransient<AutopilotMonitor.Functions.Services.Notifications.NotificationDispatchHandler>();
 // Transactional email (welcome / farewell). Typed client like the other outbound notifiers;
 // the provider is an implementation detail of EmailService + the Email:* settings.
 builder.Services.AddSingleton<EmailTemplateService>();

src/Backend/AutopilotMonitor.Functions/Services/Activation/TenantAutoApproveQueueWorker.cs

@@ -1,51 +0,0 @@
-using System;
-using System.Threading;
-using System.Threading.Tasks;
-using Azure.Storage.Queues;
-using AutopilotMonitor.Functions.Services.Queueing;
-using AutopilotMonitor.Shared;
-using Microsoft.Extensions.Logging;
-
-namespace AutopilotMonitor.Functions.Services.Activation
-{
-    /// <summary>
-    /// Background poll-loop for the <c>tenant-auto-approve</c> queue. Thin queue shell —
-    /// all decisions live in <see cref="TenantAutoApproveHandler"/>. Defaults of
-    /// <see cref="QueuePollingWorkerBase"/> (batch 32, 5-min visibility, 10-s poll,
-    /// poison after 5 attempts) fit this tiny workload.
-    /// </summary>
-    public sealed class TenantAutoApproveQueueWorker : QueuePollingWorker<TenantAutoApproveEnvelope>
-    {
-        private readonly TenantAutoApproveHandler _handler;
-
-        public TenantAutoApproveQueueWorker(
-            QueueClientFactory queueFactory,
-            TenantAutoApproveHandler handler,
-            ILogger<TenantAutoApproveQueueWorker> logger)
-            : base(queueFactory, Constants.QueueNames.TenantAutoApprove, logger)
-        {
-            _handler = handler;
-        }
-
-        /// <summary>Test seam: inject pre-built (possibly in-memory) queues.</summary>
-        internal TenantAutoApproveQueueWorker(
-            QueueClient mainQueue,
-            QueueClient poisonQueue,
-            TenantAutoApproveHandler handler,
-            ILogger logger,
-            TimeSpan? pollIntervalOverride = null)
-            : base(mainQueue, poisonQueue, logger, pollIntervalOverride)
-        {
-            _handler = handler;
-        }
-
-        protected override bool TryValidate(TenantAutoApproveEnvelope envelope)
-            => !string.IsNullOrWhiteSpace(envelope.TenantId);
-
-        protected override string DescribeForLog(TenantAutoApproveEnvelope envelope)
-            => $"tenant={envelope.TenantId}";
-
-        protected override Task HandleAsync(TenantAutoApproveEnvelope envelope, CancellationToken ct)
-            => _handler.HandleAsync(envelope, ct);
-    }
-}

src/Backend/AutopilotMonitor.Functions/Services/Analyze/AnalyzeOnEnrollmentEndQueueWorker.cs

@@ -1,34 +0,0 @@
-using System;
-using System.Threading;
-using System.Threading.Tasks;
-using AutopilotMonitor.Functions.Services.Queueing;
-using AutopilotMonitor.Shared;
-using AutopilotMonitor.Shared.Models;
-using Microsoft.Extensions.Logging;
-
-namespace AutopilotMonitor.Functions.Services.Analyze
-{
-    /// <summary>
-    /// Background worker for the <c>analyze-on-enrollment-end</c> queue. Replaces the previous
-    /// in-function fire-and-forget <c>Task.Run</c> that ran the rule engine after session-terminal
-    /// events — that pattern could be killed mid-flight by Azure Functions scale-in, leaving rule
-    /// results un-persisted. Uses the shared <see cref="QueuePollingWorker{TEnvelope}"/> poll loop
-    /// with vanilla defaults (batch 32, 5-min visibility, 5 attempts → poison).
-    /// </summary>
-    public sealed class AnalyzeOnEnrollmentEndQueueWorker : QueuePollingWorker<AnalyzeOnEnrollmentEndEnvelope>
-    {
-        private readonly AnalyzeOnEnrollmentEndHandler _handler;
-
-        public AnalyzeOnEnrollmentEndQueueWorker(
-            QueueClientFactory queueFactory,
-            AnalyzeOnEnrollmentEndHandler handler,
-            ILogger<AnalyzeOnEnrollmentEndQueueWorker> logger)
-            : base(queueFactory, Constants.QueueNames.AnalyzeOnEnrollmentEnd, logger)
-        {
-            _handler = handler ?? throw new ArgumentNullException(nameof(handler));
-        }
-
-        protected override Task HandleAsync(AnalyzeOnEnrollmentEndEnvelope envelope, CancellationToken ct)
-            => _handler.HandleAsync(envelope, ct);
-    }
-}

src/Backend/AutopilotMonitor.Functions/Services/Backup/Queue/CriticalTableBackupQueueWorker.cs

@@ -31,7 +31,7 @@ namespace AutopilotMonitor.Functions.Services.Backup.Queue
     /// CAS Running→Completed (under lease) → Release Lease → SafeDelete.
     /// </para>
     /// <para>
-    /// <b>Queue-config differs from short-lived workers</b> (plan Wave6): <see cref="BatchSize"/>=1,
+    /// <b>Queue-config differs from short-lived workers</b> (plan Wave6):
     /// <see cref="VisibilityTimeout"/>=60min. The per-run service budget is 50min so the 60min
     /// visibility gives a 10min cushion.
     /// </para>
@@ -61,7 +61,6 @@ namespace AutopilotMonitor.Functions.Services.Backup.Queue
             _clock = clock ?? TimeProvider.System;
         }
 
-        protected override int BatchSize => 1;
         protected override TimeSpan VisibilityTimeout => TimeSpan.FromMinutes(60);
 
         protected override async Task ProcessOneAsync(QueueMessage msg, CancellationToken ct)

src/Backend/AutopilotMonitor.Functions/Services/Deletion/SessionDeletionMaintenanceQueueWorker.cs

@@ -20,7 +20,7 @@ namespace AutopilotMonitor.Functions.Services.Deletion
     /// retried — the operator asked for "a run", and one is already active.
     /// </para>
     /// <para>
-    /// BatchSize=1 + VisibilityTimeout=60min (run budget is 50min plus cushion — same rationale
+    /// VisibilityTimeout=60min (run budget is 50min plus cushion — same rationale
     /// as <c>CriticalTableBackupQueueWorker</c>). Real exceptions rethrow → visibility-timeout
     /// retry → poison after <see cref="QueuePollingWorkerBase.MaxDequeueCount"/>.
     /// </para>
@@ -38,7 +38,6 @@ namespace AutopilotMonitor.Functions.Services.Deletion
             _maintenance = maintenance ?? throw new ArgumentNullException(nameof(maintenance));
         }
 
-        protected override int BatchSize => 1;
         protected override TimeSpan VisibilityTimeout => TimeSpan.FromMinutes(60);
 
         protected override bool TryValidate(SessionDeletionMaintenanceTriggerEnvelope envelope)

src/Backend/AutopilotMonitor.Functions/Services/Deletion/SessionDeletionWorker.cs

@@ -15,10 +15,9 @@ namespace AutopilotMonitor.Functions.Services.Deletion
 {
     /// <summary>
     /// Background poll-loop for the <c>session-deletion</c> queue (plan §5 PR4). Built on the
-    /// shared <see cref="QueuePollingWorker{TEnvelope}"/> with four explicit deviations:
+    /// shared <see cref="QueuePollingWorker{TEnvelope}"/> with three explicit deviations (one
+    /// cascade per receive — tens of MB and minutes of wall-time — is the base default):
     /// <list type="bullet">
-    ///   <item><see cref="BatchSize"/> = <b>1</b> (not 32) — a single cascade can be tens of MB
-    ///       and minutes of wall-time; one-at-a-time bounds memory and poison blast radius.</item>
     ///   <item><b>Kill-switch on entry</b> via <see cref="ShouldPauseAsync"/>: every receive checks
     ///       <c>AdminConfiguration.SessionDeletionKillSwitch</c>; when active the worker idles
     ///       without dequeuing.</item>
@@ -83,13 +82,6 @@ namespace AutopilotMonitor.Functions.Services.Deletion
             _opsEvents = opsEvents ?? throw new ArgumentNullException(nameof(opsEvents));
         }
 
-        /// <summary>
-        /// Explicit deviation from sibling workers — one cascade per receive bounds the worker's
-        /// memory footprint (a 35MB snapshot blob × 32 messages = 1.1GB) and limits poison-queue
-        /// blast radius on a corruption signal. Plan §5 PR4.
-        /// </summary>
-        protected override int BatchSize => 1;
-
         protected override bool UseHeartbeat => true;
 
         protected override async ValueTask<bool> ShouldPauseAsync(CancellationToken ct)

src/Backend/AutopilotMonitor.Functions/Services/EventIngestProcessor.Notifications.cs

@@ -72,9 +72,7 @@ namespace AutopilotMonitor.Functions.Services
                         sessionUrl: sessionUrl);
                     NotificationAlertBuilder.AddRuleResultSections(alert, ruleResults);
 
-                    _ = _channelDispatcher.SendToChannelsAsync(targets, alert)
-                        .ContinueWith(t => _logger.LogWarning(t.Exception?.InnerException,
-                            "Fire-and-forget webhook notification failed"), TaskContinuationOptions.OnlyOnFaulted);
+                    await EnqueueNotificationAsync(request, targets, alert).ConfigureAwait(false);
                 }
             }
 
@@ -94,9 +92,7 @@ namespace AutopilotMonitor.Functions.Services
                     sessionUrl: sessionUrl);
                 NotificationAlertBuilder.AddRuleResultSections(alert, ruleResults);
 
-                _ = _channelDispatcher.SendToChannelsAsync(successChannels, alert)
-                    .ContinueWith(t => _logger.LogWarning(t.Exception?.InnerException,
-                        "Fire-and-forget webhook notification failed"), TaskContinuationOptions.OnlyOnFaulted);
+                await EnqueueNotificationAsync(request, successChannels, alert).ConfigureAwait(false);
             }
 
             if (c.EspFailureEvent != null && updatedSession?.IsPreProvisioned == true && failureChannels.Count > 0)
@@ -115,12 +111,28 @@ namespace AutopilotMonitor.Functions.Services
                     sessionUrl: sessionUrl);
                 NotificationAlertBuilder.AddRuleResultSections(alert, ruleResults);
 
-                _ = _channelDispatcher.SendToChannelsAsync(failureChannels, alert)
-                    .ContinueWith(t => _logger.LogWarning(t.Exception?.InnerException,
-                        "Fire-and-forget webhook notification failed"), TaskContinuationOptions.OnlyOnFaulted);
+                await EnqueueNotificationAsync(request, failureChannels, alert).ConfigureAwait(false);
             }
         }
 
+        /// <summary>
+        /// Hands an alert to the notification-dispatch queue. The send used to be fire-and-forget
+        /// after the HTTP response; the host does not count such work at shutdown, so a Flex
+        /// scale-in could drop a customer's enrollment webhook. The envelope carries channel ids
+        /// only — the consumer re-resolves URLs and secrets from the tenant config at send time.
+        /// The producer is fail-soft, so the agent's response is never blocked.
+        /// </summary>
+        private Task EnqueueNotificationAsync(
+            IngestEventsRequest request, List<NotificationChannel> channels, NotificationAlert alert)
+            => _notificationProducer.EnqueueAsync(new NotificationDispatchEnvelope
+            {
+                TenantId = request.TenantId,
+                SessionId = request.SessionId,
+                ChannelIds = channels.Select(c => c.Id).ToList(),
+                Alert = alert,
+                EnqueuedAt = DateTime.UtcNow,
+            });
+
         private SignalRMessageAction[] BuildSignalRMessages(
             IngestEventsRequest request, SessionSummary? updatedSession, int processedCount,
             List<RuleResult> newRuleResults)

src/Backend/AutopilotMonitor.Functions/Services/EventIngestProcessor.cs

@@ -38,7 +38,7 @@ namespace AutopilotMonitor.Functions.Services
         private readonly TenantConfigurationService _configService;
         private readonly AnalyzeRuleService _analyzeRuleService;
         private readonly GatherRuleService _gatherRuleService;
-        private readonly NotificationChannelDispatcher _channelDispatcher;
+        private readonly INotificationDispatchProducer _notificationProducer;
         private readonly AdminConfigurationService _adminConfigService;
         private readonly OpsEventService _opsEventService;
         private readonly SlaBreachEvaluationService _slaBreachService;
@@ -61,7 +61,7 @@ namespace AutopilotMonitor.Functions.Services
             TenantConfigurationService configService,
             AnalyzeRuleService analyzeRuleService,
             GatherRuleService gatherRuleService,
-            NotificationChannelDispatcher channelDispatcher,
+            INotificationDispatchProducer notificationProducer,
             AdminConfigurationService adminConfigService,
             OpsEventService opsEventService,
             SlaBreachEvaluationService slaBreachService,
@@ -81,7 +81,7 @@ namespace AutopilotMonitor.Functions.Services
             _configService = configService;
             _analyzeRuleService = analyzeRuleService;
             _gatherRuleService = gatherRuleService;
-            _channelDispatcher = channelDispatcher;
+            _notificationProducer = notificationProducer;
             _adminConfigService = adminConfigService;
             _opsEventService = opsEventService;
             _slaBreachService = slaBreachService;

src/Backend/AutopilotMonitor.Functions/Services/Ime/AzureQueueImeMsiArchiveProducer.cs

@@ -33,7 +33,7 @@ namespace AutopilotMonitor.Functions.Services.Ime
             ILogger<AzureQueueImeMsiArchiveProducer> logger)
         {
             _logger = logger;
-            // Base64 encoding matches the consumer (BackgroundService) so messages round-trip.
+            // Base64 encoding matches the host's queue-trigger default so messages round-trip.
             _queueClient = queueFactory.Create(Constants.QueueNames.ImeMsiArchive);
         }
 
@@ -49,7 +49,10 @@ namespace AutopilotMonitor.Functions.Services.Ime
             _logger = logger;
         }
 
-        public async Task EnqueueAsync(ImeMsiArchiveEnvelope envelope, CancellationToken cancellationToken = default)
+        public async Task EnqueueAsync(
+            ImeMsiArchiveEnvelope envelope,
+            TimeSpan? visibilityDelay = null,
+            CancellationToken cancellationToken = default)
         {
             if (envelope is null) return;
             if (string.IsNullOrEmpty(envelope.Version))
@@ -63,10 +66,13 @@ namespace AutopilotMonitor.Functions.Services.Ime
             try
             {
                 var body = JsonConvert.SerializeObject(envelope);
-                await _queueClient.SendMessageAsync(body, cancellationToken: cancellationToken).ConfigureAwait(false);
+                await _queueClient.SendMessageAsync(
+                    body,
+                    visibilityTimeout: visibilityDelay,
+                    cancellationToken: cancellationToken).ConfigureAwait(false);
                 _logger.LogInformation(
-                    "ImeMsiArchive enqueued (version={Version} urlPresent={UrlPresent} tenant={Tenant} session={Session})",
-                    envelope.Version, !string.IsNullOrEmpty(envelope.MsiDownloadUrl), envelope.TenantId, envelope.SessionId);
+                    "ImeMsiArchive enqueued (version={Version} urlPresent={UrlPresent} tenant={Tenant} session={Session} delay={Delay})",
+                    envelope.Version, !string.IsNullOrEmpty(envelope.MsiDownloadUrl), envelope.TenantId, envelope.SessionId, visibilityDelay);
             }
             catch (Exception ex)
             {

src/Backend/AutopilotMonitor.Functions/Services/Ime/IImeMsiArchiveProducer.cs

@@ -1,3 +1,4 @@
+using System;
 using System.Threading;
 using System.Threading.Tasks;
 using AutopilotMonitor.Shared.Models;
@@ -8,9 +9,14 @@ namespace AutopilotMonitor.Functions.Services.Ime
     /// Enqueues an IME-installer archive job when a new IME version is first sighted.
     /// Implementations must be fail-soft: the caller sits on the ingest hot path's
     /// fire-and-forget continuation and must never observe an exception.
+    /// <paramref name="visibilityDelay"/> parks the message: the queue function uses it to
+    /// re-enqueue while archiving is switched off.
     /// </summary>
     public interface IImeMsiArchiveProducer
     {
-        Task EnqueueAsync(ImeMsiArchiveEnvelope envelope, CancellationToken cancellationToken = default);
+        Task EnqueueAsync(
+            ImeMsiArchiveEnvelope envelope,
+            TimeSpan? visibilityDelay = null,
+            CancellationToken cancellationToken = default);
     }
 }

src/Backend/AutopilotMonitor.Functions/Services/Ime/ImeMsiArchiveQueueWorker.cs

@@ -1,107 +0,0 @@
-using AutopilotMonitor.Functions.Services.Queueing;
-using AutopilotMonitor.Shared;
-using AutopilotMonitor.Shared.DataAccess;
-using AutopilotMonitor.Shared.Models;
-using Microsoft.Extensions.Logging;
-
-namespace AutopilotMonitor.Functions.Services.Ime
-{
-    /// <summary>
-    /// Background worker for the <c>ime-msi-archive</c> queue: downloads a newly sighted IME
-    /// installer into the permanent <c>ime-archive</c> container and merges the outcome onto
-    /// the ImeVersionHistory row. Volume is ~one message per Microsoft IME release.
-    /// <para>
-    /// Retry semantics: the archiver never throws — it classifies. Permanent rejections
-    /// (bad version, size cap) complete the message with their <c>Failed:*</c> status
-    /// persisted; transient failures (download/timeout) rethrow so the visibility-timeout
-    /// retry → poison ladder applies, with the last status visible on the row.
-    /// <see cref="UseHeartbeat"/> keeps the message invisible across the up-to-10-minute
-    /// download window. The status row-merge happens on EVERY attempt so operators can see
-    /// a stuck download as <c>Failed:Download</c> instead of nothing.
-    /// </para>
-    /// </summary>
-    public sealed class ImeMsiArchiveQueueWorker : QueuePollingWorker<ImeMsiArchiveEnvelope>
-    {
-        private readonly ImeMsiArchiver _archiver;
-        private readonly ISessionRepository _sessionRepo;
-        private readonly AdminConfigurationService _adminConfigService;
-
-        public ImeMsiArchiveQueueWorker(
-            QueueClientFactory queueFactory,
-            ImeMsiArchiver archiver,
-            ISessionRepository sessionRepo,
-            AdminConfigurationService adminConfigService,
-            ILogger<ImeMsiArchiveQueueWorker> logger)
-            : base(queueFactory, Constants.QueueNames.ImeMsiArchive, logger)
-        {
-            _archiver = archiver ?? throw new ArgumentNullException(nameof(archiver));
-            _sessionRepo = sessionRepo ?? throw new ArgumentNullException(nameof(sessionRepo));
-            _adminConfigService = adminConfigService ?? throw new ArgumentNullException(nameof(adminConfigService));
-        }
-
-        /// <summary>
-        /// Test seam: construct directly with mock <see cref="Azure.Storage.Queues.QueueClient"/>
-        /// instances so the poll loop can be exercised without Azurite. Mirrors
-        /// <see cref="Deletion.SessionDeletionWorker"/>'s internal test ctor.
-        /// </summary>
-        internal ImeMsiArchiveQueueWorker(
-            Azure.Storage.Queues.QueueClient mainQueue,
-            Azure.Storage.Queues.QueueClient poisonQueue,
-            ImeMsiArchiver archiver,
-            ISessionRepository sessionRepo,
-            AdminConfigurationService adminConfigService,
-            ILogger<ImeMsiArchiveQueueWorker> logger,
-            TimeSpan? pollInterval = null)
-            : base(mainQueue, poisonQueue, logger, pollInterval)
-        {
-            _archiver = archiver ?? throw new ArgumentNullException(nameof(archiver));
-            _sessionRepo = sessionRepo ?? throw new ArgumentNullException(nameof(sessionRepo));
-            _adminConfigService = adminConfigService ?? throw new ArgumentNullException(nameof(adminConfigService));
-        }
-
-        /// <summary>Download can legitimately take minutes — extend visibility while working.</summary>
-        protected override bool UseHeartbeat => true;
-
-        /// <summary>
-        /// Paused while <see cref="AdminConfiguration.ImeMsiArchivingEnabled"/> is off —
-        /// messages stay parked (not dropped) and are processed when the flag returns.
-        /// Uses the cached config read: this is a pause, not a kill-switch, and the worker
-        /// only wakes for rare messages anyway.
-        /// </summary>
-        protected override async ValueTask<bool> ShouldPauseAsync(CancellationToken ct)
-        {
-            try
-            {
-                var config = await _adminConfigService.GetConfigurationAsync();
-                return !config.ImeMsiArchivingEnabled;
-            }
-            catch
-            {
-                return false; // config unreadable → keep working (archive is fail-soft anyway)
-            }
-        }
-
-        protected override bool TryValidate(ImeMsiArchiveEnvelope envelope)
-            => !string.IsNullOrEmpty(envelope.Version);
-
-        protected override string DescribeForLog(ImeMsiArchiveEnvelope envelope)
-            => $"version {envelope.Version}";
-
-        protected override async Task HandleAsync(ImeMsiArchiveEnvelope envelope, CancellationToken ct)
-        {
-            var result = await _archiver.ArchiveAsync(envelope, ct);
-
-            // Merge the outcome first — also for failures, so the row always tells the truth.
-            await _sessionRepo.UpdateImeVersionArchiveInfoAsync(
-                envelope.Version, result.Status, result.BlobPath, result.Sha256, result.SizeBytes, result.SourceUrl);
-
-            if (!result.Success && result.Retryable)
-            {
-                // Visibility-timeout retry → poison after MaxDequeueCount. The thrown message
-                // lands in the worker's handler-failure log line.
-                throw new InvalidOperationException(
-                    $"IME MSI archive attempt failed with {result.Status} for version {envelope.Version}");
-            }
-        }
-    }
-}

src/Backend/AutopilotMonitor.Functions/Services/Indexing/IndexReconcileQueueWorker.cs

@@ -1,34 +0,0 @@
-using System;
-using System.Threading;
-using System.Threading.Tasks;
-using AutopilotMonitor.Functions.Services.Queueing;
-using AutopilotMonitor.Shared;
-using AutopilotMonitor.Shared.Models;
-using Microsoft.Extensions.Logging;
-
-namespace AutopilotMonitor.Functions.Services.Indexing
-{
-    /// <summary>
-    /// Background worker for the <c>telemetry-index-reconcile</c> queue (Plan §2.8, §M5.d.3).
-    /// Replaces the earlier <c>IndexReconcileFunction</c> QueueTrigger (whose binding required a
-    /// Functions-host-specific connection app-setting the project's storage access does not need)
-    /// with the shared <see cref="QueuePollingWorker{TEnvelope}"/> poll loop. Vanilla defaults:
-    /// batch 32, 5-min visibility, 5 attempts → poison.
-    /// </summary>
-    public sealed class IndexReconcileQueueWorker : QueuePollingWorker<IndexReconcileEnvelope>
-    {
-        private readonly IndexReconcileHandler _handler;
-
-        public IndexReconcileQueueWorker(
-            QueueClientFactory queueFactory,
-            IndexReconcileHandler handler,
-            ILogger<IndexReconcileQueueWorker> logger)
-            : base(queueFactory, Constants.QueueNames.TelemetryIndexReconcile, logger)
-        {
-            _handler = handler ?? throw new ArgumentNullException(nameof(handler));
-        }
-
-        protected override Task HandleAsync(IndexReconcileEnvelope envelope, CancellationToken ct)
-            => _handler.HandleAsync(envelope, ct);
-    }
-}

src/Backend/AutopilotMonitor.Functions/Services/Notifications/AzureQueueNotificationDispatchProducer.cs

@@ -0,0 +1,89 @@
+using System;
+using System.Threading;
+using System.Threading.Tasks;
+using Azure.Storage.Queues;
+using AutopilotMonitor.Functions.Services.Queueing;
+using AutopilotMonitor.Shared;
+using AutopilotMonitor.Shared.Models.Notifications;
+using Microsoft.Extensions.Logging;
+using Newtonsoft.Json;
+
+namespace AutopilotMonitor.Functions.Services.Notifications
+{
+    /// <summary>
+    /// Azure Storage Queue implementation of <see cref="INotificationDispatchProducer"/>. Mirrors
+    /// <see cref="Analyze.AzureQueueAnalyzeOnEnrollmentEndProducer"/>: Managed Identity preferred,
+    /// connection-string fallback, Base64 encoding (matches the host's queue trigger default).
+    /// <para>
+    /// <b>Fault tolerance:</b> queue-side exceptions are logged but never rethrown — the agent's
+    /// HTTP 200 must not be blocked. A missed enqueue means the alert is not delivered; the
+    /// warning names tenant and event type so an operator can see it.
+    /// </para>
+    /// </summary>
+    public sealed class AzureQueueNotificationDispatchProducer : INotificationDispatchProducer
+    {
+        private readonly QueueClient _queueClient;
+        private readonly ILogger<AzureQueueNotificationDispatchProducer> _logger;
+
+        private int _queueEnsured; // 0 = not yet ensured, 1 = CreateIfNotExistsAsync has run
+
+        public AzureQueueNotificationDispatchProducer(
+            QueueClientFactory queueFactory,
+            ILogger<AzureQueueNotificationDispatchProducer> logger)
+        {
+            _logger = logger;
+            _queueClient = queueFactory.Create(Constants.QueueNames.NotificationDispatch);
+        }
+
+        /// <summary>Test seam: construct directly with a mock <see cref="QueueClient"/>.</summary>
+        internal AzureQueueNotificationDispatchProducer(
+            QueueClient queueClient,
+            ILogger<AzureQueueNotificationDispatchProducer> logger)
+        {
+            _queueClient = queueClient;
+            _logger = logger;
+        }
+
+        public async Task EnqueueAsync(NotificationDispatchEnvelope envelope, CancellationToken cancellationToken = default)
+        {
+            if (envelope is null) return;
+            if (string.IsNullOrEmpty(envelope.TenantId) || envelope.Alert is null || envelope.ChannelIds.Count == 0)
+            {
+                _logger.LogWarning(
+                    "NotificationDispatch enqueue skipped — missing TenantId/Alert/ChannelIds (tenant={Tenant} event={EventType})",
+                    envelope.TenantId, envelope.Alert?.EventType);
+                return;
+            }
+
+            await EnsureQueueExistsAsync(cancellationToken).ConfigureAwait(false);
+
+            try
+            {
+                var body = JsonConvert.SerializeObject(envelope);
+                await _queueClient.SendMessageAsync(body, cancellationToken: cancellationToken).ConfigureAwait(false);
+            }
+            catch (Exception ex)
+            {
+                _logger.LogWarning(ex,
+                    "NotificationDispatch enqueue failed (tenant={Tenant} session={Session} event={EventType} channels={Channels}) — alert not delivered",
+                    envelope.TenantId, envelope.SessionId, envelope.Alert.EventType, envelope.ChannelIds.Count);
+            }
+        }
+
+        private async Task EnsureQueueExistsAsync(CancellationToken cancellationToken)
+        {
+            if (_queueEnsured == 1) return;
+
+            try
+            {
+                await _queueClient.CreateIfNotExistsAsync(cancellationToken: cancellationToken).ConfigureAwait(false);
+                Interlocked.Exchange(ref _queueEnsured, 1);
+            }
+            catch (Exception ex)
+            {
+                // Leave _queueEnsured=0 so a later call retries.
+                _logger.LogWarning(ex, "NotificationDispatch queue CreateIfNotExists failed — will retry next enqueue");
+            }
+        }
+    }
+}

src/Backend/AutopilotMonitor.Functions/Services/Notifications/INotificationDispatchProducer.cs

@@ -0,0 +1,16 @@
+using System.Threading;
+using System.Threading.Tasks;
+using AutopilotMonitor.Shared.Models.Notifications;
+
+namespace AutopilotMonitor.Functions.Services.Notifications
+{
+    /// <summary>
+    /// Enqueues a channel notification for durable delivery by
+    /// <c>NotificationDispatchQueueFunction</c>. Implementations must be fail-soft: callers sit
+    /// on the agent's hot path (ingest, distress) and must never observe an exception.
+    /// </summary>
+    public interface INotificationDispatchProducer
+    {
+        Task EnqueueAsync(NotificationDispatchEnvelope envelope, CancellationToken cancellationToken = default);
+    }
+}

src/Backend/AutopilotMonitor.Functions/Services/Notifications/NotificationDispatchHandler.cs

@@ -0,0 +1,64 @@
+using System;
+using System.Collections.Generic;
+using System.Linq;
+using System.Threading;
+using System.Threading.Tasks;
+using AutopilotMonitor.Shared.Models.Notifications;
+using Microsoft.Extensions.Logging;
+
+namespace AutopilotMonitor.Functions.Services.Notifications
+{
+    /// <summary>
+    /// Consumer side of the <c>notification-dispatch</c> queue: resolves the envelope's channel
+    /// ids against the tenant's CURRENT configuration (so secrets never travel in the queue and
+    /// a channel disabled meanwhile is skipped) and hands the alert to
+    /// <see cref="NotificationChannelDispatcher"/>. The dispatcher never throws — a failing
+    /// destination only logs — so the queue guarantees the attempt, not the delivery; per-request
+    /// retries stay with the transport's resilience policy.
+    /// </summary>
+    public class NotificationDispatchHandler
+    {
+        private readonly TenantConfigurationService _configService;
+        private readonly NotificationChannelDispatcher _dispatcher;
+        private readonly ILogger<NotificationDispatchHandler> _logger;
+
+        public NotificationDispatchHandler(
+            TenantConfigurationService configService,
+            NotificationChannelDispatcher dispatcher,
+            ILogger<NotificationDispatchHandler> logger)
+        {
+            _configService = configService ?? throw new ArgumentNullException(nameof(configService));
+            _dispatcher = dispatcher ?? throw new ArgumentNullException(nameof(dispatcher));
+            _logger = logger ?? throw new ArgumentNullException(nameof(logger));
+        }
+
+        public virtual async Task HandleAsync(NotificationDispatchEnvelope envelope, CancellationToken cancellationToken)
+        {
+            if (envelope is null) throw new ArgumentNullException(nameof(envelope));
+
+            var (config, exists) = await _configService.TryGetConfigurationAsync(envelope.TenantId).ConfigureAwait(false);
+            if (!exists)
+            {
+                _logger.LogWarning(
+                    "NotificationDispatch: tenant {Tenant} has no configuration — alert {EventType} dropped",
+                    envelope.TenantId, envelope.Alert?.EventType);
+                return;
+            }
+
+            var wanted = new HashSet<string>(envelope.ChannelIds ?? new List<string>(), StringComparer.Ordinal);
+            var targets = config.GetNotificationChannels()
+                .Where(c => c.Enabled && wanted.Contains(c.Id))
+                .ToList();
+
+            if (targets.Count == 0)
+            {
+                _logger.LogWarning(
+                    "NotificationDispatch: none of {Requested} channel(s) is still enabled for tenant {Tenant} — alert {EventType} dropped",
+                    wanted.Count, envelope.TenantId, envelope.Alert?.EventType);
+                return;
+            }
+
+            await _dispatcher.SendToChannelsAsync(targets, envelope.Alert).ConfigureAwait(false);
+        }
+    }
+}

src/Backend/AutopilotMonitor.Functions/Services/Offboarding/TenantOffboardingWorker.cs

@@ -15,9 +15,9 @@ namespace AutopilotMonitor.Functions.Services.Offboarding
     /// Background poll-loop for the <c>tenant-offboarding</c> queue (Plan §7.3). Built on the
     /// shared <see cref="QueuePollingWorker{TEnvelope}"/>:
     /// <list type="bullet">
-    ///   <item><see cref="BatchSize"/> = <b>1</b> — one tenant cascade per receive bounds memory
-    ///       (per-tenant enumerate + per-session enqueue is long-running) and limits the blast
-    ///       radius of a poison.</item>
+    ///   <item>One tenant cascade per receive (the base default) bounds memory — per-tenant
+    ///       enumerate + per-session enqueue is long-running — and limits the blast radius of a
+    ///       poison.</item>
     ///   <item>Heartbeat (<see cref="UseHeartbeat"/>) extends visibility while the handler runs so
     ///       concurrent re-delivery can't spawn a parallel offboarding for the same tenant.</item>
     ///   <item>Max-dequeue → poison queue, but <see cref="BeforePoisonMoveAsync"/> persists the
@@ -57,8 +57,6 @@ namespace AutopilotMonitor.Functions.Services.Offboarding
             _opsEvents = opsEvents ?? throw new ArgumentNullException(nameof(opsEvents));
         }
 
-        protected override int BatchSize => 1;
-
         protected override bool UseHeartbeat => true;
 
         protected override bool TryValidate(TenantOffboardingEnvelope envelope)

src/Backend/AutopilotMonitor.Functions/Services/Queueing/QueueEnvelopeReader.cs

@@ -0,0 +1,67 @@
+using System;
+using Azure.Storage.Queues.Models;
+using Microsoft.Extensions.Logging;
+using Newtonsoft.Json;
+
+namespace AutopilotMonitor.Functions.Services.Queueing
+{
+    /// <summary>
+    /// Envelope decoding shared by the <c>[QueueTrigger]</c> functions under <c>Functions/Queue/</c>.
+    /// The functions bind the raw <see cref="QueueMessage"/> rather than a POCO so the producers'
+    /// Newtonsoft contract stays the single wire format and <see cref="QueueMessage.DequeueCount"/>
+    /// reaches the log line.
+    /// <para>
+    /// A <c>false</c> result means "permanently unusable": malformed JSON, a <c>null</c> literal,
+    /// or a failed validation. The caller returns normally, the host deletes the message — the
+    /// same drop the self-managed poll loop performs, without burning a dequeue attempt.
+    /// </para>
+    /// </summary>
+    public static class QueueEnvelopeReader
+    {
+        public static bool TryRead<TEnvelope>(
+            QueueMessage message,
+            ILogger logger,
+            out TEnvelope envelope,
+            Func<TEnvelope, bool>? validate = null)
+            where TEnvelope : class
+        {
+            if (message is null) throw new ArgumentNullException(nameof(message));
+            if (logger is null) throw new ArgumentNullException(nameof(logger));
+
+            envelope = null!;
+            var name = typeof(TEnvelope).Name;
+
+            TEnvelope? parsed;
+            try
+            {
+                parsed = JsonConvert.DeserializeObject<TEnvelope>(message.Body.ToString());
+            }
+            catch (JsonException ex)
+            {
+                logger.LogWarning(ex,
+                    "{Envelope}: malformed envelope JSON — dropping (msg {Id}, dequeue {N})",
+                    name, message.MessageId, message.DequeueCount);
+                return false;
+            }
+
+            if (parsed is null)
+            {
+                logger.LogWarning(
+                    "{Envelope}: null envelope after deserialization — dropping (msg {Id}, dequeue {N})",
+                    name, message.MessageId, message.DequeueCount);
+                return false;
+            }
+
+            if (validate is not null && !validate(parsed))
+            {
+                logger.LogWarning(
+                    "{Envelope}: envelope failed validation (missing required fields) — dropping (msg {Id}, dequeue {N})",
+                    name, message.MessageId, message.DequeueCount);
+                return false;
+            }
+
+            envelope = parsed;
+            return true;
+        }
+    }
+}

src/Backend/AutopilotMonitor.Functions/Services/Queueing/QueuePollingWorkerBase.cs

@@ -9,13 +9,20 @@ using Microsoft.Extensions.Logging;
 namespace AutopilotMonitor.Functions.Services.Queueing
 {
     /// <summary>
-    /// Shared poll-loop skeleton for the project's Storage-Queue <see cref="BackgroundService"/>
-    /// workers (code-quality audit 2026-05-29, finding D2). Replaces the QueueTrigger binding —
-    /// which needs a Functions-host-specific <c>&lt;Connection&gt;__queueServiceUri</c> app-setting
-    /// the rest of the project's storage access does not use — with a pure DI poll loop that
-    /// reproduces the platform's retry semantics:
+    /// Shared poll-loop skeleton for the project's <b>self-managed</b> Storage-Queue
+    /// <see cref="BackgroundService"/> workers (code-quality audit 2026-05-29, finding D2).
+    /// <para>
+    /// Two classes of queue consumer exist on purpose. Stateless "envelope in, handler, done"
+    /// queues are <c>[QueueTrigger]</c> functions under <c>Functions/Queue/</c>: the Functions host
+    /// owns lease renewal, poison-move and, on Flex Consumption, the scale signal that wakes an
+    /// instance for a message. This base serves the cascade workers whose per-message lifecycle
+    /// the host cannot express — lease/CAS state machines, a poison-move veto, a kill-switch pause
+    /// and an in-flight heartbeat. Their accepted limit: after a scale-in the cascade resumes on the
+    /// next instance, at the latest when the next timer wakes one. The loop reproduces the
+    /// platform's retry semantics:
+    /// </para>
     /// <list type="bullet">
-    ///   <item>Receive up to <see cref="BatchSize"/> messages with <see cref="VisibilityTimeout"/>.</item>
+    ///   <item>Receive <see cref="BatchSize"/> message(s) with <see cref="VisibilityTimeout"/>.</item>
     ///   <item>Handler success → <c>DeleteMessageAsync</c>.</item>
     ///   <item>Handler failure → no delete; the message reappears after the visibility timeout and
     ///     <c>DequeueCount</c> increments (identical to platform retry).</item>
@@ -75,8 +82,14 @@ namespace AutopilotMonitor.Functions.Services.Queueing
 
         // ── Tunables (override per worker) ───────────────────────────────────────
 
-        /// <summary>Max messages received per poll. Storage Queue caps batch-receive at 32.</summary>
-        protected virtual int BatchSize => 32;
+        /// <summary>
+        /// Messages received per poll. Default <b>1</b>: the loop processes a batch sequentially and
+        /// the heartbeat only covers the in-flight message, so every message received beyond the
+        /// first would sit invisible with its clock running — with 32 × a 10-second handler the last
+        /// one reappears for another worker before this loop reaches it (audit 2026-09-05, F02).
+        /// A non-empty receive is followed by an immediate re-receive, so 1 costs no throughput.
+        /// </summary>
+        protected virtual int BatchSize => 1;
 
         /// <summary>Visibility timeout per received message — must cover the slowest handler path.</summary>
         protected virtual TimeSpan VisibilityTimeout => TimeSpan.FromMinutes(5);

src/Backend/AutopilotMonitor.Functions/Services/RuleEngine.cs

@@ -60,7 +60,7 @@ namespace AutopilotMonitor.Functions.Services
         /// <b>Failure semantics:</b> storage-layer exceptions from rule loading, event reading
         /// or existing-results lookup propagate to the caller. The queue worker relies on this
         /// to leave its message un-deleted so a transient Table Storage failure can be retried
-        /// via visibility-timeout (<see cref="AutopilotMonitor.Functions.Services.Analyze.AnalyzeOnEnrollmentEndQueueWorker"/>).
+        /// via visibility-timeout (<c>AnalyzeOnEnrollmentEndQueueFunction</c>).
         /// The on-demand HTTP path (<see cref="AutopilotMonitor.Functions.Functions.Rules.GetRuleResultsFunction"/>)
         /// wraps the call in its own try/catch and logs failures as warnings — the user sees
         /// the previously-stored results until they re-trigger.

src/Backend/AutopilotMonitor.Functions/Services/TenantApprovalService.cs

@@ -7,7 +7,7 @@ namespace AutopilotMonitor.Functions.Services;
 /// Shared tenant-activation path: whitelist approval plus the follow-up side effects
 /// (auto-promote of the onboarding user to TenantAdmin, welcome email). Used by both
 /// the Global Admin approve endpoint (<c>PreviewWhitelistFunction</c>) and the
-/// auto-approve queue worker (<c>TenantAutoApproveQueueWorker</c>) so the two
+/// auto-approve queue worker (<c>TenantAutoApproveQueueFunction</c>) so the two
 /// activation routes cannot drift.
 /// </summary>
 public class TenantApprovalService

src/Backend/AutopilotMonitor.Functions/Services/Vulnerability/VulnerabilityCorrelateQueueWorker.cs

@@ -1,35 +0,0 @@
-using System;
-using System.Threading;
-using System.Threading.Tasks;
-using AutopilotMonitor.Functions.Services.Queueing;
-using AutopilotMonitor.Shared;
-using AutopilotMonitor.Shared.Models;
-using Microsoft.Extensions.Logging;
-
-namespace AutopilotMonitor.Functions.Services.Vulnerability
-{
-    /// <summary>
-    /// Background worker for the <c>vulnerability-correlate</c> queue. Replaces the previous
-    /// in-function fire-and-forget <c>Task.Run</c> in <c>EventIngestProcessor</c> that ran
-    /// vulnerability correlation after a shutdown <c>software_inventory_analysis</c> event — that
-    /// pattern could be killed mid-flight by Azure Functions scale-in, leaving sessions without a
-    /// stored report. Uses the shared <see cref="QueuePollingWorker{TEnvelope}"/> poll loop with
-    /// vanilla defaults (batch 32, 5-min visibility, 5 attempts → poison).
-    /// </summary>
-    public sealed class VulnerabilityCorrelateQueueWorker : QueuePollingWorker<VulnerabilityCorrelateEnvelope>
-    {
-        private readonly VulnerabilityCorrelateHandler _handler;
-
-        public VulnerabilityCorrelateQueueWorker(
-            QueueClientFactory queueFactory,
-            VulnerabilityCorrelateHandler handler,
-            ILogger<VulnerabilityCorrelateQueueWorker> logger)
-            : base(queueFactory, Constants.QueueNames.VulnerabilityCorrelate, logger)
-        {
-            _handler = handler ?? throw new ArgumentNullException(nameof(handler));
-        }
-
-        protected override Task HandleAsync(VulnerabilityCorrelateEnvelope envelope, CancellationToken ct)
-            => _handler.HandleAsync(envelope, ct);
-    }
-}

src/Backend/AutopilotMonitor.Functions/host.json

@@ -27,6 +27,14 @@
     "extensions": {
         "http": {
             "routePrefix": "api"
+        },
+        "queues": {
+            "messageEncoding": "base64",
+            "batchSize": 4,
+            "newBatchThreshold": 2,
+            "maxDequeueCount": 5,
+            "visibilityTimeout": "00:01:00",
+            "maxPollingInterval": "00:00:10"
         }
     }
 }

src/Backend/AutopilotMonitor.Functions/local.settings.json.template

@@ -6,6 +6,9 @@
     "AzureTableStorageConnectionString": "UseDevelopmentStorage=true",
     "AzureBlobStorageConnectionString": "UseDevelopmentStorage=true",
 
+    "// DataStorage": "Connection the [QueueTrigger] functions under Functions/Queue bind to (Constants.QueueNames.TriggerConnection). Locally a connection string; in Azure the identity form 'DataStorage__queueServiceUri=https://<data-account>.queue.core.windows.net'. The Functions host resolves it, so it cannot be derived from AzureStorageAccountName in code.",
+    "DataStorage": "UseDevelopmentStorage=true",
+
     "// SignalR quota watcher": "Optional. Set to the full Azure resource id of the SignalR Service to enable hourly OpsEvents when the plan limits are approached (defaults: Standard S1 x1 unit = 1000 connections / 1M messages per day; override the limits when adding units or changing tier). Function App's managed identity needs 'Monitoring Reader' on this resource. Leave unset to skip the watcher.",
     "SignalRResourceId": "",
     "SignalRConnectionLimit": "1000",

src/Shared/AutopilotMonitor.Shared/Constants.cs

@@ -1566,6 +1566,27 @@ namespace AutopilotMonitor.Shared
         /// </summary>
         public static class QueueNames
         {
+            /// <summary>
+            /// Name of the host connection every <c>[QueueTrigger]</c> function binds to
+            /// (<c>Connection = TriggerConnection</c>). Resolved by the Functions HOST, not by the
+            /// worker's DI, so it is its own app setting: in Azure the identity form
+            /// <c>DataStorage__queueServiceUri</c> pointing at the data storage account (the
+            /// Managed Identity holds Storage Queue Data Contributor there); locally a plain
+            /// connection string under <c>DataStorage</c>. Producers keep using
+            /// <c>QueueClientFactory</c> — same account, same queues.
+            /// </summary>
+            public const string TriggerConnection = "DataStorage";
+
+            /// <summary>
+            /// Durable channel notifications from the agent hot paths (terminal enrollment alerts
+            /// in ingest, hardware rejection in distress). Producer = <c>INotificationDispatchProducer</c>;
+            /// consumer = <c>NotificationDispatchQueueFunction</c>, which re-resolves the envelope's
+            /// channel ids against the tenant config and sends. Replaced fire-and-forget sends the
+            /// host did not count as open work at shutdown. Poison suffix <c>-poison</c>,
+            /// max-dequeue 5.
+            /// </summary>
+            public const string NotificationDispatch = "notification-dispatch";
+
             /// <summary>
             /// V2 Decision Engine index-table fan-out (Plan §2.8, §M5.d). One message per
             /// committed primary row (Signal or DecisionTransition); consumer writes the
@@ -1621,7 +1642,7 @@ namespace AutopilotMonitor.Shared
             /// <summary>
             /// Delayed tenant auto-approve queue. Producer (AuthFunction first-login signup
             /// handler) enqueues one envelope per new tenant with a ~1-minute visibility delay;
-            /// consumer (<c>TenantAutoApproveQueueWorker</c>) activates the tenant via the shared
+            /// consumer (<c>TenantAutoApproveQueueFunction</c>) activates the tenant via the shared
             /// approval path if <c>AdminConfiguration.AutoApproveNewTenants</c> is enabled at
             /// processing time — otherwise the message is dropped and the tenant waits for
             /// manual approval. Poison suffix <c>-poison</c>, max-dequeue 5.
@@ -1631,8 +1652,8 @@ namespace AutopilotMonitor.Shared
             /// <summary>
             /// Critical-table backup job queue. Producer = HTTP trigger
             /// <c>/api/global/backups/trigger</c> (fail-hard); consumer =
-            /// <c>CriticalTableBackupQueueWorker</c> (BackgroundService, BatchSize=1,
-            /// VisibilityTimeout=60min with 25min PopReceipt-renewal). Carries a small
+            /// <c>CriticalTableBackupQueueWorker</c> (self-managed BackgroundService, one message
+            /// per receive, VisibilityTimeout=60min with 25min PopReceipt-renewal). Carries a small
             /// envelope <c>{ jobId }</c> — full state lives in <see cref="TableNames.BackupJobs"/>.
             /// Poison-suffix <c>-poison</c>, max-dequeue 5; Failed-state is persisted on
             /// Poison-Move (NOT on first throw — that would defeat retry).
@@ -1645,8 +1666,8 @@ namespace AutopilotMonitor.Shared
             /// <summary>
             /// Manual-trigger queue for the session-deletion maintenance run. Producer = HTTP
             /// trigger <c>/api/global/session-deletions/maintenance/trigger</c> (fail-hard);
-            /// consumer = <c>SessionDeletionMaintenanceQueueWorker</c> (BackgroundService,
-            /// BatchSize=1, VisibilityTimeout=60min — the run budget is 50min plus cushion).
+            /// consumer = <c>SessionDeletionMaintenanceQueueWorker</c> (self-managed
+            /// BackgroundService, VisibilityTimeout=60min — the run budget is 50min plus cushion).
             /// Carries a tiny envelope <c>{ triggeredBy }</c>; concurrency against the 12h timer
             /// is serialized by the session-deletion maintenance blob lease, not the queue.
             /// Poison-suffix <c>-poison</c>, max-dequeue 5.
@@ -1659,7 +1680,7 @@ namespace AutopilotMonitor.Shared
             /// <summary>
             /// IME-MSI archive fan-out. Producer = <c>EventIngestProcessor</c> when
             /// <c>RecordImeVersionAsync</c> reports a genuinely NEW IME version (first sighting
-            /// across the fleet — roughly monthly); consumer = <c>ImeMsiArchiveQueueWorker</c>,
+            /// across the fleet — roughly monthly); consumer = <c>ImeMsiArchiveQueueFunction</c>,
             /// which downloads the installer from the CSP-reported
             /// <c>msiDownloadUrl</c> (host-allowlisted, size-capped) into
             /// <see cref="BlobContainers.ImeArchive"/> and merges the outcome into the

src/Shared/AutopilotMonitor.Shared/Models/Notifications/NotificationDispatchEnvelope.cs

@@ -0,0 +1,38 @@
+using System;
+using System.Collections.Generic;
+
+namespace AutopilotMonitor.Shared.Models.Notifications
+{
+    /// <summary>
+    /// Queue-message envelope for the <c>notification-dispatch</c> queue: an enrollment or
+    /// hardware-rejection alert that must reach the tenant's channels even though the HTTP
+    /// request that produced it has already been answered. Replaces the fire-and-forget sends
+    /// in the ingest and distress paths, which the Functions host does not count as open work
+    /// at shutdown and which a Flex scale-in could therefore drop.
+    /// <para>
+    /// The envelope names the target channels by <see cref="ChannelIds"/> only. The consumer
+    /// re-resolves them from the tenant configuration at send time, so no webhook URL, custom
+    /// header or signing secret is ever copied into a queue message, and a channel disabled
+    /// between enqueue and send is skipped.
+    /// </para>
+    /// </summary>
+    public sealed class NotificationDispatchEnvelope
+    {
+        /// <summary>Schema version — bump on breaking envelope changes so consumers can reject or migrate.</summary>
+        public string EnvelopeVersion { get; set; } = "1";
+
+        public string TenantId { get; set; } = string.Empty;
+
+        /// <summary>Session the alert is about; null for alerts without one (hardware rejection).</summary>
+        public string? SessionId { get; set; }
+
+        /// <summary>Ids of the channels the producer selected (already filtered by their NotifyOn* toggle).</summary>
+        public List<string> ChannelIds { get; set; } = new();
+
+        /// <summary>The fully built, channel-agnostic alert; renderers run in the consumer.</summary>
+        public NotificationAlert Alert { get; set; } = default!;
+
+        /// <summary>UTC time the producer enqueued the message; useful for measuring queue lag.</summary>
+        public DateTime EnqueuedAt { get; set; }
+    }
+}