feat(backend): M5.d.4 index-reconcile timer + release-gate — M5 complete

Closes the M5.d dual-write loop with a 2h safety-net timer and the
per-index-table PK/RK consistency release-gate tests required by
Plan §M5-Pflicht.

Components:
- ISignalRepository.QueryByTimestampAtOrAfterAsync +
  IDecisionTransitionRepository.QueryByTimestampAtOrAfterAsync: new
  cross-partition scans filtered on the Azure Tables system-managed
  `Timestamp` property. Bounded to maxResults=50000 with a warn-log
  on cap-reached so silent truncation can't masquerade as success.
- IndexReconcileTimer: TimerTrigger on "0 0 */2 * * *" (same cadence
  as MaintenanceFunction). Scans the last 4h (window deliberately 2×
  the cadence so every primary row gets re-checked at least twice
  before falling out of range), builds envelopes via the existing
  IndexReconcileEnvelopeFactory, and hands them to the producer —
  reusing the M5.d.2 flag-gate + queue-failure-swallow semantics
  instead of duplicating them.
- Testable core: RunReconcileAsync(cutoffUtc, CT) split out from the
  TimerTrigger entry so unit tests can drive with deterministic
  cutoffs without an Azure Functions host.
- AdminConfigurationService.GetConfigurationAsync is now virtual —
  enables Moq-based mocking in timer flag-gate tests. Low-risk prod
  change (no existing caller relies on final).

Release gate (Plan §M5-Pflicht line 1395): IndexDualWriteConsistencyTests
verifies per-index-table that the captured index row's back-ref
fields (TenantId, SessionId, StepIndex / SessionSignalOrdinal) plug
directly into the primary repos' key builders to reconstruct the
primary (PartitionKey, RowKey). The resubmit-idempotency test simulates
a timer-triggered replay of the same primaries and asserts all
duplicated rows collapse onto identical Index (PK, RK) pairs — which
is exactly what the real TableIndexRepository's UpsertReplace does
at the storage layer.

Left deferred (not blocking M5 close):
- SessionsByStage stale-row pruning: stage-X→Y transitions still leave
  an orphan row in {tenantId}_X. Either query-side dedupe (per
  SessionId, max-Ticks RK) or a follow-up pruner handles it.
- Actual EnableIndexDualWrite=true flip: stays default-false until
  post-deploy verification runs. Producer + timer both no-op in the
  meantime, so the change is zero-risk on merge.
- Cross-partition-scan perf: fine for current primary-table sizes;
  revisit if prod metrics after pilot show degradation.

Tests (+10 in Functions):
- 4 IndexReconcileTimerTests: flag-off short-circuits queries, flag-on
  with primary rows enqueues all envelopes, empty-window skips enqueue,
  ReconcileWindow=4h invariant.
- 6 IndexDualWriteConsistencyTests: 5 per-index-table PK/RK back-ref
  (SignalsByKind + all 4 transition-derived indices) + 1 resubmit
  idempotency ensuring all duplicated rows collapse onto the same
  index keys.

M5 status: M5.a + M5.b (+ 3 sub-commits) + M5.c (+ 3 sub-commits) +
M5.d (+ 4 sub-commits) all complete. M5.d loop:
  Primary commit (M5.a) → Envelope enqueue (M5.d.2, flag-gated)
  → Queue → Handler (M5.d.3) → Index-Tables (M5.d.1)
  + 2h safety-net Timer (M5.d.4) reconciles missed envelopes.

Full Functions suite: 692 green (prev 682 + 10 new). Full solution
build green.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

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

@@ -0,0 +1,285 @@
+using AutopilotMonitor.Functions.DataAccess.TableStorage;
+using AutopilotMonitor.Functions.Services.Indexing;
+using AutopilotMonitor.Shared.DataAccess;
+using AutopilotMonitor.Shared.Models;
+using Microsoft.Extensions.Logging.Abstractions;
+
+namespace AutopilotMonitor.Functions.Tests;
+
+/// <summary>
+/// <b>M5 release-gate</b> (Plan §M5-Pflicht line 1395): per-index-table PK/RK consistency
+/// between primary rows and their derived index rows. For each index table, walk a
+/// canonical primary row through the Factory + Handler pipeline and verify that the
+/// captured index record carries enough back-ref information to reconstruct the
+/// primary's <c>(PartitionKey, RowKey)</c> via <see cref="TableSignalRepository"/> /
+/// <see cref="TableDecisionTransitionRepository"/> key builders.
+/// <para>
+/// Also asserts pipeline idempotency: running the same primary row through twice
+/// produces identical upsert keys — the consumer's <c>UpsertReplace</c> batches stay
+/// deduplicating even under timer-triggered resubmits.
+/// </para>
+/// </summary>
+public class IndexDualWriteConsistencyTests
+{
+    private const string TenantId  = "a1b2c3d4-e5f6-7890-abcd-ef1234567890";
+    private const string SessionId = "b2c3d4e5-f6a7-8901-bcde-f12345678901";
+
+    private static readonly DateTime Occurred =
+        new(2026, 4, 21, 14, 7, 33, DateTimeKind.Utc);
+
+    private static (IndexReconcileHandler handler, FakeIndexTableRepository repo) Sut()
+    {
+        var repo = new FakeIndexTableRepository();
+        var handler = new IndexReconcileHandler(repo, NullLogger<IndexReconcileHandler>.Instance);
+        return (handler, repo);
+    }
+
+    // ============================================================ SignalsByKind ↔ Signals
+
+    [Fact]
+    public async Task SignalsByKind_row_carries_back_ref_to_primary_Signals_row()
+    {
+        var primary = new SignalRecord
+        {
+            TenantId             = TenantId,
+            SessionId            = SessionId,
+            SessionSignalOrdinal = 42,
+            Kind                 = "EspTerminalFailure",
+            OccurredAtUtc        = Occurred,
+            SourceOrigin         = "EspAndHelloTrackerAdapter",
+        };
+
+        var (handler, repo) = Sut();
+        await handler.HandleAsync(IndexReconcileEnvelopeFactory.FromSignal(primary));
+
+        var index = Assert.Single(repo.SignalsByKind);
+
+        // Back-ref fields on the index row must reconstruct the primary PK/RK.
+        var reconstructedPk = TableSignalRepository.BuildPartitionKey(index.TenantId, index.SessionId);
+        var reconstructedRk = TableSignalRepository.BuildRowKey(index.SessionSignalOrdinal);
+
+        Assert.Equal(
+            TableSignalRepository.BuildPartitionKey(primary.TenantId, primary.SessionId),
+            reconstructedPk);
+        Assert.Equal(
+            TableSignalRepository.BuildRowKey(primary.SessionSignalOrdinal),
+            reconstructedRk);
+    }
+
+    // ============================================================ SessionsByTerminal ↔ DecisionTransitions
+
+    [Fact]
+    public async Task SessionsByTerminal_row_carries_back_ref_to_primary_DecisionTransitions_row()
+    {
+        var primary = TerminalTransition();
+
+        var (handler, repo) = Sut();
+        await handler.HandleAsync(IndexReconcileEnvelopeFactory.FromDecisionTransition(primary));
+
+        var index = Assert.Single(repo.SessionsByTerminal);
+
+        var reconstructedPk = TableDecisionTransitionRepository.BuildPartitionKey(index.TenantId, index.SessionId);
+        var reconstructedRk = TableDecisionTransitionRepository.BuildRowKey(index.StepIndex);
+
+        Assert.Equal(
+            TableDecisionTransitionRepository.BuildPartitionKey(primary.TenantId, primary.SessionId),
+            reconstructedPk);
+        Assert.Equal(
+            TableDecisionTransitionRepository.BuildRowKey(primary.StepIndex),
+            reconstructedRk);
+    }
+
+    [Fact]
+    public async Task SessionsByStage_row_carries_back_ref_to_primary_DecisionTransitions_row()
+    {
+        var primary = new DecisionTransitionRecord
+        {
+            TenantId      = TenantId,
+            SessionId     = SessionId,
+            StepIndex     = 3,
+            FromStage     = "AwaitingEsp",
+            ToStage       = "EspInProgress",
+            Taken         = true,
+            IsTerminal    = false,
+            OccurredAtUtc = Occurred,
+        };
+
+        var (handler, repo) = Sut();
+        await handler.HandleAsync(IndexReconcileEnvelopeFactory.FromDecisionTransition(primary));
+
+        var index = Assert.Single(repo.SessionsByStage);
+
+        var reconstructedPk = TableDecisionTransitionRepository.BuildPartitionKey(index.TenantId, index.SessionId);
+        var reconstructedRk = TableDecisionTransitionRepository.BuildRowKey(index.StepIndex);
+
+        Assert.Equal(
+            TableDecisionTransitionRepository.BuildPartitionKey(primary.TenantId, primary.SessionId),
+            reconstructedPk);
+        Assert.Equal(
+            TableDecisionTransitionRepository.BuildRowKey(primary.StepIndex),
+            reconstructedRk);
+    }
+
+    [Fact]
+    public async Task DeadEndsByReason_row_carries_back_ref_to_primary_DecisionTransitions_row()
+    {
+        var primary = new DecisionTransitionRecord
+        {
+            TenantId      = TenantId,
+            SessionId     = SessionId,
+            StepIndex     = 5,
+            FromStage     = "EspInProgress",
+            ToStage       = "EspInProgress",
+            Taken         = false,
+            IsTerminal    = false,
+            DeadEndReason = "hybrid_reboot_gate_blocking",
+            OccurredAtUtc = Occurred,
+        };
+
+        var (handler, repo) = Sut();
+        await handler.HandleAsync(IndexReconcileEnvelopeFactory.FromDecisionTransition(primary));
+
+        var index = Assert.Single(repo.DeadEndsByReason);
+
+        Assert.Equal(
+            TableDecisionTransitionRepository.BuildPartitionKey(primary.TenantId, primary.SessionId),
+            TableDecisionTransitionRepository.BuildPartitionKey(index.TenantId, index.SessionId));
+        Assert.Equal(
+            TableDecisionTransitionRepository.BuildRowKey(primary.StepIndex),
+            TableDecisionTransitionRepository.BuildRowKey(index.StepIndex));
+    }
+
+    [Fact]
+    public async Task ClassifierVerdictsByIdLevel_row_carries_back_ref_to_primary_DecisionTransitions_row()
+    {
+        var primary = new DecisionTransitionRecord
+        {
+            TenantId                  = TenantId,
+            SessionId                 = SessionId,
+            StepIndex                 = 7,
+            FromStage                 = "EspInProgress",
+            ToStage                   = "WhiteGloveSealed",
+            Taken                     = true,
+            IsTerminal                = true,
+            ClassifierVerdictId       = "whiteglove-sealing",
+            ClassifierHypothesisLevel = "Strong",
+            OccurredAtUtc             = Occurred,
+        };
+
+        var (handler, repo) = Sut();
+        await handler.HandleAsync(IndexReconcileEnvelopeFactory.FromDecisionTransition(primary));
+
+        var index = Assert.Single(repo.ClassifierVerdictsByIdLevel);
+
+        Assert.Equal(
+            TableDecisionTransitionRepository.BuildPartitionKey(primary.TenantId, primary.SessionId),
+            TableDecisionTransitionRepository.BuildPartitionKey(index.TenantId, index.SessionId));
+        Assert.Equal(
+            TableDecisionTransitionRepository.BuildRowKey(primary.StepIndex),
+            TableDecisionTransitionRepository.BuildRowKey(index.StepIndex));
+    }
+
+    // ============================================================ pipeline idempotency
+
+    [Fact]
+    public async Task Resubmitting_same_primary_rows_produces_same_index_keys()
+    {
+        // Simulates the M5.d.4 timer re-enqueueing primary rows it already handled —
+        // the handler's index-row keys must be key-stable so the real TableIndexRepository's
+        // UpsertReplace collapses duplicates rather than appending rows.
+        var (handler, repo) = Sut();
+
+        var signal = new SignalRecord
+        {
+            TenantId             = TenantId,
+            SessionId            = SessionId,
+            SessionSignalOrdinal = 1,
+            Kind                 = "DesktopArrived",
+            OccurredAtUtc        = Occurred,
+        };
+        var transition = TerminalTransition();
+
+        // First pass: initial ingest fan-out.
+        await handler.HandleAsync(IndexReconcileEnvelopeFactory.FromSignal(signal));
+        await handler.HandleAsync(IndexReconcileEnvelopeFactory.FromDecisionTransition(transition));
+
+        // Second pass: 2h-timer reconcile resubmits the same primaries.
+        await handler.HandleAsync(IndexReconcileEnvelopeFactory.FromSignal(signal));
+        await handler.HandleAsync(IndexReconcileEnvelopeFactory.FromDecisionTransition(transition));
+
+        // The fake repo captures one row per call, but all duplicated rows must map to the
+        // same (PK, RK) — the real Azure Tables Upsert would collapse them.
+        Assert.Equal(2, repo.SignalsByKind.Count);
+        var signalKeys = repo.SignalsByKind
+            .Select(r => (
+                Pk: IndexRowKeys.BuildSignalsByKindPk(r.TenantId, r.SignalKind),
+                Rk: IndexRowKeys.BuildSignalsByKindRk(r.OccurredAtUtc, r.SessionId, r.SessionSignalOrdinal)))
+            .Distinct()
+            .ToList();
+        Assert.Single(signalKeys);
+
+        Assert.Equal(2, repo.SessionsByTerminal.Count);
+        var terminalKeys = repo.SessionsByTerminal
+            .Select(r => (
+                Pk: IndexRowKeys.BuildSessionsByTerminalPk(r.TenantId, r.TerminalStage),
+                Rk: IndexRowKeys.BuildSessionsByTerminalRk(r.OccurredAtUtc, r.SessionId)))
+            .Distinct()
+            .ToList();
+        Assert.Single(terminalKeys);
+
+        Assert.Equal(2, repo.SessionsByStage.Count);
+        var stageKeys = repo.SessionsByStage
+            .Select(r => (
+                Pk: IndexRowKeys.BuildSessionsByStagePk(r.TenantId, r.Stage),
+                Rk: IndexRowKeys.BuildSessionsByStageRk(r.LastUpdatedUtc, r.SessionId)))
+            .Distinct()
+            .ToList();
+        Assert.Single(stageKeys);
+
+        Assert.Equal(2, repo.ClassifierVerdictsByIdLevel.Count);
+        var verdictKeys = repo.ClassifierVerdictsByIdLevel
+            .Select(r => (
+                Pk: IndexRowKeys.BuildClassifierVerdictsByIdLevelPk(r.TenantId, r.ClassifierId, r.HypothesisLevel),
+                Rk: IndexRowKeys.BuildClassifierVerdictsByIdLevelRk(r.OccurredAtUtc, r.SessionId, r.StepIndex)))
+            .Distinct()
+            .ToList();
+        Assert.Single(verdictKeys);
+    }
+
+    // ============================================================ helpers
+
+    private static DecisionTransitionRecord TerminalTransition()
+        => new()
+        {
+            TenantId                  = TenantId,
+            SessionId                 = SessionId,
+            StepIndex                 = 7,
+            FromStage                 = "EspInProgress",
+            ToStage                   = "WhiteGloveSealed",
+            Taken                     = true,
+            IsTerminal                = true,
+            ClassifierVerdictId       = "whiteglove-sealing",
+            ClassifierHypothesisLevel = "Strong",
+            OccurredAtUtc             = Occurred,
+        };
+
+    private sealed class FakeIndexTableRepository : IIndexTableRepository
+    {
+        public List<SessionsByTerminalRecord>          SessionsByTerminal          { get; } = new();
+        public List<SessionsByStageRecord>             SessionsByStage             { get; } = new();
+        public List<DeadEndsByReasonRecord>            DeadEndsByReason            { get; } = new();
+        public List<ClassifierVerdictsByIdLevelRecord> ClassifierVerdictsByIdLevel { get; } = new();
+        public List<SignalsByKindRecord>               SignalsByKind               { get; } = new();
+
+        public Task<int> StoreSessionsByTerminalAsync(IReadOnlyList<SessionsByTerminalRecord> records, CancellationToken ct = default)
+        { SessionsByTerminal.AddRange(records); return Task.FromResult(records.Count); }
+        public Task<int> StoreSessionsByStageAsync(IReadOnlyList<SessionsByStageRecord> records, CancellationToken ct = default)
+        { SessionsByStage.AddRange(records); return Task.FromResult(records.Count); }
+        public Task<int> StoreDeadEndsByReasonAsync(IReadOnlyList<DeadEndsByReasonRecord> records, CancellationToken ct = default)
+        { DeadEndsByReason.AddRange(records); return Task.FromResult(records.Count); }
+        public Task<int> StoreClassifierVerdictsByIdLevelAsync(IReadOnlyList<ClassifierVerdictsByIdLevelRecord> records, CancellationToken ct = default)
+        { ClassifierVerdictsByIdLevel.AddRange(records); return Task.FromResult(records.Count); }
+        public Task<int> StoreSignalsByKindAsync(IReadOnlyList<SignalsByKindRecord> records, CancellationToken ct = default)
+        { SignalsByKind.AddRange(records); return Task.FromResult(records.Count); }
+    }
+}

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

@@ -0,0 +1,156 @@
+using AutopilotMonitor.Functions.Functions.Admin;
+using AutopilotMonitor.Functions.Services;
+using AutopilotMonitor.Shared.DataAccess;
+using AutopilotMonitor.Shared.Models;
+using Microsoft.Extensions.Caching.Memory;
+using Microsoft.Extensions.Logging.Abstractions;
+using Moq;
+
+namespace AutopilotMonitor.Functions.Tests;
+
+/// <summary>
+/// Tests for the 2h safety-net <see cref="IndexReconcileTimer"/> (Plan §2.8, §M5.d.4).
+/// Covers the flag-gate short-circuit, the happy path (enqueue every primary row
+/// in the window), and the empty-window case.
+/// </summary>
+public class IndexReconcileTimerTests
+{
+    private const string TenantId  = "a1b2c3d4-e5f6-7890-abcd-ef1234567890";
+    private const string SessionId = "b2c3d4e5-f6a7-8901-bcde-f12345678901";
+
+    private static readonly DateTime Cutoff = new(2026, 4, 21, 10, 0, 0, DateTimeKind.Utc);
+
+    // ============================================================ flag-off short-circuit
+
+    [Fact]
+    public async Task Feature_flag_off_skips_queries_and_enqueue()
+    {
+        var adminConfigSvc = BuildAdminConfigService(enableIndexDualWrite: false);
+        var signalRepo     = new Mock<ISignalRepository>(MockBehavior.Strict);
+        var transitionRepo = new Mock<IDecisionTransitionRepository>(MockBehavior.Strict);
+        var producer       = new Mock<IIndexReconcileProducer>(MockBehavior.Strict);
+
+        var timer = new IndexReconcileTimer(
+            adminConfigSvc,
+            signalRepo.Object,
+            transitionRepo.Object,
+            producer.Object,
+            NullLogger<IndexReconcileTimer>.Instance);
+
+        await timer.RunReconcileAsync(Cutoff, default);
+
+        // Strict mocks throw if any unconfigured call happens — so zero setups + no throw
+        // proves the timer short-circuited before any query / enqueue.
+        signalRepo.VerifyNoOtherCalls();
+        transitionRepo.VerifyNoOtherCalls();
+        producer.VerifyNoOtherCalls();
+    }
+
+    // ============================================================ happy path
+
+    [Fact]
+    public async Task Flag_on_with_primary_rows_enqueues_all_envelopes()
+    {
+        var adminConfigSvc = BuildAdminConfigService(enableIndexDualWrite: true);
+
+        var signals = new List<SignalRecord>
+        {
+            new() { TenantId = TenantId, SessionId = SessionId, SessionSignalOrdinal = 1, Kind = "EspPhaseChanged", OccurredAtUtc = Cutoff.AddMinutes(5) },
+            new() { TenantId = TenantId, SessionId = SessionId, SessionSignalOrdinal = 2, Kind = "DesktopArrived",   OccurredAtUtc = Cutoff.AddMinutes(10) },
+        };
+        var transitions = new List<DecisionTransitionRecord>
+        {
+            new() { TenantId = TenantId, SessionId = SessionId, StepIndex = 1, Taken = true, FromStage = "A", ToStage = "B", OccurredAtUtc = Cutoff.AddMinutes(7) },
+        };
+
+        var signalRepo = new Mock<ISignalRepository>();
+        signalRepo
+            .Setup(r => r.QueryByTimestampAtOrAfterAsync(Cutoff, It.IsAny<int>(), It.IsAny<CancellationToken>()))
+            .ReturnsAsync(signals);
+        var transitionRepo = new Mock<IDecisionTransitionRepository>();
+        transitionRepo
+            .Setup(r => r.QueryByTimestampAtOrAfterAsync(Cutoff, It.IsAny<int>(), It.IsAny<CancellationToken>()))
+            .ReturnsAsync(transitions);
+
+        List<IndexReconcileEnvelope>? captured = null;
+        var producer = new Mock<IIndexReconcileProducer>();
+        producer
+            .Setup(p => p.EnqueueBatchAsync(It.IsAny<IReadOnlyList<IndexReconcileEnvelope>>(), It.IsAny<CancellationToken>()))
+            .Callback<IReadOnlyList<IndexReconcileEnvelope>, CancellationToken>((e, _) => captured = e.ToList())
+            .ReturnsAsync(3);
+
+        var timer = new IndexReconcileTimer(
+            adminConfigSvc,
+            signalRepo.Object,
+            transitionRepo.Object,
+            producer.Object,
+            NullLogger<IndexReconcileTimer>.Instance);
+
+        await timer.RunReconcileAsync(Cutoff, default);
+
+        Assert.NotNull(captured);
+        Assert.Equal(3, captured!.Count);
+        Assert.Equal("Signal",             captured[0].SourceKind);
+        Assert.Equal(1L,                   captured[0].SessionSignalOrdinal);
+        Assert.Equal("Signal",             captured[1].SourceKind);
+        Assert.Equal(2L,                   captured[1].SessionSignalOrdinal);
+        Assert.Equal("DecisionTransition", captured[2].SourceKind);
+        Assert.Equal(1,                    captured[2].StepIndex);
+    }
+
+    // ============================================================ empty-window: no enqueue
+
+    [Fact]
+    public async Task Flag_on_empty_window_skips_enqueue()
+    {
+        var adminConfigSvc = BuildAdminConfigService(enableIndexDualWrite: true);
+
+        var signalRepo = new Mock<ISignalRepository>();
+        signalRepo
+            .Setup(r => r.QueryByTimestampAtOrAfterAsync(Cutoff, It.IsAny<int>(), It.IsAny<CancellationToken>()))
+            .ReturnsAsync(new List<SignalRecord>());
+        var transitionRepo = new Mock<IDecisionTransitionRepository>();
+        transitionRepo
+            .Setup(r => r.QueryByTimestampAtOrAfterAsync(Cutoff, It.IsAny<int>(), It.IsAny<CancellationToken>()))
+            .ReturnsAsync(new List<DecisionTransitionRecord>());
+
+        var producer = new Mock<IIndexReconcileProducer>(MockBehavior.Strict);
+
+        var timer = new IndexReconcileTimer(
+            adminConfigSvc,
+            signalRepo.Object,
+            transitionRepo.Object,
+            producer.Object,
+            NullLogger<IndexReconcileTimer>.Instance);
+
+        await timer.RunReconcileAsync(Cutoff, default);
+
+        // Empty batch → no producer call (strict mock throws otherwise).
+        producer.VerifyNoOtherCalls();
+    }
+
+    // ============================================================ cutoff window invariant
+
+    [Fact]
+    public void ReconcileWindow_is_4_hours_and_larger_than_timer_cadence()
+    {
+        // The 2h cron fires every 2h but scans a 4h window — so every primary row is
+        // re-checked at least twice before falling out of range. Protect the invariant.
+        Assert.Equal(TimeSpan.FromHours(4), IndexReconcileTimer.ReconcileWindow);
+    }
+
+    // ============================================================ helpers
+
+    private static AdminConfigurationService BuildAdminConfigService(bool enableIndexDualWrite)
+    {
+        var mock = new Mock<AdminConfigurationService>(
+            Mock.Of<IConfigRepository>(),
+            NullLogger<AdminConfigurationService>.Instance,
+            new MemoryCache(new MemoryCacheOptions()));
+
+        mock.Setup(s => s.GetConfigurationAsync())
+            .ReturnsAsync(new AdminConfiguration { EnableIndexDualWrite = enableIndexDualWrite });
+
+        return mock.Object;
+    }
+}

src/Backend/AutopilotMonitor.Functions/DataAccess/TableStorage/TableDecisionTransitionRepository.cs

@@ -1,3 +1,4 @@
+using System;
 using System.Collections.Generic;
 using System.Linq;
 using System.Threading;
@@ -100,6 +101,33 @@ namespace AutopilotMonitor.Functions.DataAccess.TableStorage
             return results;
         }
 
+        public async Task<List<DecisionTransitionRecord>> QueryByTimestampAtOrAfterAsync(
+            DateTime cutoffUtc, int maxResults = 50_000, CancellationToken cancellationToken = default)
+        {
+            var table = _storage.GetTableClient(Constants.TableNames.DecisionTransitions);
+            var filter = $"Timestamp ge datetime'{cutoffUtc.ToUniversalTime():yyyy-MM-ddTHH:mm:ss.fffffffZ}'";
+
+            var results = new List<DecisionTransitionRecord>(capacity: 128);
+            var pages = table.QueryAsync<TableEntity>(
+                filter: filter,
+                maxPerPage: 1000,
+                cancellationToken: cancellationToken);
+
+            await foreach (var entity in pages.ConfigureAwait(false))
+            {
+                if (results.Count >= maxResults)
+                {
+                    _logger.LogWarning(
+                        "DecisionTransitions: time-range query reached maxResults cap {Cap} at cutoff {Cutoff:o} — narrow window or bump cap",
+                        maxResults, cutoffUtc);
+                    break;
+                }
+                results.Add(FromEntity(entity));
+            }
+
+            return results;
+        }
+
         /// <summary>
         /// Projects an Azure <see cref="TableEntity"/> back into a <see cref="DecisionTransitionRecord"/>,
         /// reassembling chunked <c>PayloadJson</c> if present. Internal for mapping tests.

src/Backend/AutopilotMonitor.Functions/DataAccess/TableStorage/TableSignalRepository.cs

@@ -1,3 +1,4 @@
+using System;
 using System.Collections.Generic;
 using System.Linq;
 using System.Threading;
@@ -101,6 +102,35 @@ namespace AutopilotMonitor.Functions.DataAccess.TableStorage
             return results;
         }
 
+        public async Task<List<SignalRecord>> QueryByTimestampAtOrAfterAsync(
+            DateTime cutoffUtc, int maxResults = 50_000, CancellationToken cancellationToken = default)
+        {
+            var table = _storage.GetTableClient(Constants.TableNames.Signals);
+            // Timestamp is the Azure Tables system-managed server-side property. Filter shape
+            // follows OData datetime literal format.
+            var filter = $"Timestamp ge datetime'{cutoffUtc.ToUniversalTime():yyyy-MM-ddTHH:mm:ss.fffffffZ}'";
+
+            var results = new List<SignalRecord>(capacity: 128);
+            var pages = table.QueryAsync<TableEntity>(
+                filter: filter,
+                maxPerPage: 1000,
+                cancellationToken: cancellationToken);
+
+            await foreach (var entity in pages.ConfigureAwait(false))
+            {
+                if (results.Count >= maxResults)
+                {
+                    _logger.LogWarning(
+                        "Signals: time-range query reached maxResults cap {Cap} at cutoff {Cutoff:o} — narrow window or bump cap",
+                        maxResults, cutoffUtc);
+                    break;
+                }
+                results.Add(FromEntity(entity));
+            }
+
+            return results;
+        }
+
         /// <summary>
         /// Projects an Azure <see cref="TableEntity"/> back into a <see cref="SignalRecord"/>,
         /// reassembling chunked <c>PayloadJson</c> if present. Internal for mapping tests.

src/Backend/AutopilotMonitor.Functions/Functions/Admin/IndexReconcileTimer.cs

@@ -0,0 +1,100 @@
+using System;
+using System.Threading;
+using System.Threading.Tasks;
+using AutopilotMonitor.Functions.Services;
+using AutopilotMonitor.Functions.Services.Indexing;
+using AutopilotMonitor.Shared.DataAccess;
+using Microsoft.Azure.Functions.Worker;
+using Microsoft.Extensions.Logging;
+
+namespace AutopilotMonitor.Functions.Functions.Admin
+{
+    /// <summary>
+    /// 2h safety-net timer (Plan §2.8, §M5.d.4) that recovers missed index-table writes.
+    /// Scans the last 4h of primary <c>Signals</c> + <c>DecisionTransitions</c> rows and
+    /// re-enqueues envelopes onto the <c>telemetry-index-reconcile</c> queue via the
+    /// same producer the hot ingest path uses. The queue-triggered handler (M5.d.3)
+    /// then upserts index rows idempotently.
+    /// <para>
+    /// <b>Why 2h / 4h:</b> window is deliberately larger than cadence so every row
+    /// gets re-checked at least twice before falling out of scan range. Queue failures
+    /// or pod restarts during ingest are recoverable within the overlap.
+    /// </para>
+    /// <para>
+    /// <b>Feature gate:</b> <see cref="Shared.Models.AdminConfiguration.EnableIndexDualWrite"/>.
+    /// When false, the scan is short-circuited — no queries, no enqueues.
+    /// </para>
+    /// </summary>
+    public sealed class IndexReconcileTimer
+    {
+        // Cron "0 0 */2 * * *" = top of every second hour (00:00, 02:00, 04:00 UTC, …)
+        // — matches the MaintenanceFunction cadence so ops only has two timer heartbeats
+        // to babysit.
+        private const string Cron = "0 0 */2 * * *";
+
+        /// <summary>Safety-net window: scan primaries committed in the last N hours.</summary>
+        internal static readonly TimeSpan ReconcileWindow = TimeSpan.FromHours(4);
+
+        private readonly AdminConfigurationService _adminConfig;
+        private readonly ISignalRepository _signalRepo;
+        private readonly IDecisionTransitionRepository _transitionRepo;
+        private readonly IIndexReconcileProducer _producer;
+        private readonly ILogger<IndexReconcileTimer> _logger;
+
+        public IndexReconcileTimer(
+            AdminConfigurationService adminConfig,
+            ISignalRepository signalRepo,
+            IDecisionTransitionRepository transitionRepo,
+            IIndexReconcileProducer producer,
+            ILogger<IndexReconcileTimer> logger)
+        {
+            _adminConfig = adminConfig;
+            _signalRepo = signalRepo;
+            _transitionRepo = transitionRepo;
+            _producer = producer;
+            _logger = logger;
+        }
+
+        [Function("IndexReconcileTimer")]
+        public Task Run([TimerTrigger(Cron)] object timer, CancellationToken cancellationToken)
+        {
+            var cutoff = DateTime.UtcNow - ReconcileWindow;
+            return RunReconcileAsync(cutoff, cancellationToken);
+        }
+
+        /// <summary>
+        /// Testable core. Separated from the TimerTrigger entry so unit tests can drive it
+        /// with a deterministic cutoff without an Azure Functions host.
+        /// </summary>
+        internal async Task RunReconcileAsync(DateTime cutoffUtc, CancellationToken cancellationToken)
+        {
+            var config = await _adminConfig.GetConfigurationAsync().ConfigureAwait(false);
+            if (!config.EnableIndexDualWrite)
+            {
+                _logger.LogInformation(
+                    "IndexReconcileTimer: EnableIndexDualWrite=false — skipping scan (cutoff {Cutoff:o})",
+                    cutoffUtc);
+                return;
+            }
+
+            var signals = await _signalRepo
+                .QueryByTimestampAtOrAfterAsync(cutoffUtc, cancellationToken: cancellationToken)
+                .ConfigureAwait(false);
+            var transitions = await _transitionRepo
+                .QueryByTimestampAtOrAfterAsync(cutoffUtc, cancellationToken: cancellationToken)
+                .ConfigureAwait(false);
+
+            var envelopes = IndexReconcileEnvelopeFactory.BuildBatch(signals, transitions);
+
+            _logger.LogInformation(
+                "IndexReconcileTimer: re-enqueueing {Count} envelope(s) ({S} signals + {T} transitions) since {Cutoff:o}",
+                envelopes.Count, signals.Count, transitions.Count, cutoffUtc);
+
+            if (envelopes.Count == 0) return;
+
+            // The producer is already flag-gated and swallows queue-side failures. We trust
+            // its contract here — timer never throws back into the host.
+            await _producer.EnqueueBatchAsync(envelopes, cancellationToken).ConfigureAwait(false);
+        }
+    }
+}

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

@@ -30,9 +30,13 @@ namespace AutopilotMonitor.Functions.Services
         }
 
         /// <summary>
-        /// Gets global admin configuration (uses cache with 5-minute TTL)
+        /// Gets global admin configuration (uses cache with 5-minute TTL).
+        /// <para>
+        /// Virtual so it can be mocked via Moq in consumer tests that need controlled
+        /// flag values (e.g. <c>IndexReconcileTimer</c> flag-gate tests).
+        /// </para>
         /// </summary>
-        public async Task<AdminConfiguration> GetConfigurationAsync()
+        public virtual async Task<AdminConfiguration> GetConfigurationAsync()
         {
             if (_cache.TryGetValue(CacheKey, out AdminConfiguration? cachedConfig) && cachedConfig != null)
             {

src/Shared/AutopilotMonitor.Shared/DataAccess/IDecisionTransitionRepository.cs

@@ -1,3 +1,4 @@
+using System;
 using System.Collections.Generic;
 using System.Threading;
 using System.Threading.Tasks;
@@ -27,5 +28,19 @@ namespace AutopilotMonitor.Shared.DataAccess
         /// </summary>
         Task<List<DecisionTransitionRecord>> QueryBySessionAsync(
             string tenantId, string sessionId, int maxResults = 1000, CancellationToken cancellationToken = default);
+
+        /// <summary>
+        /// Cross-partition scan of all decision transitions whose Azure Tables server-side
+        /// <c>Timestamp</c> is at or after <paramref name="cutoffUtc"/>. Used by the
+        /// <c>IndexReconcileTimer</c> (Plan §M5.d.4) to re-enqueue the last 4h of primary
+        /// rows as a safety-net against queue failures.
+        /// <para>
+        /// <b>Perf note:</b> Azure Tables has no PK-bound filter here, so this is a scan;
+        /// <paramref name="maxResults"/> bounds memory. If the cap is reached the caller
+        /// should log + narrow the window rather than trust completeness.
+        /// </para>
+        /// </summary>
+        Task<List<DecisionTransitionRecord>> QueryByTimestampAtOrAfterAsync(
+            DateTime cutoffUtc, int maxResults = 50_000, CancellationToken cancellationToken = default);
     }
 }

src/Shared/AutopilotMonitor.Shared/DataAccess/ISignalRepository.cs

@@ -1,3 +1,4 @@
+using System;
 using System.Collections.Generic;
 using System.Threading;
 using System.Threading.Tasks;
@@ -29,5 +30,19 @@ namespace AutopilotMonitor.Shared.DataAccess
         /// </summary>
         Task<List<SignalRecord>> QueryBySessionAsync(
             string tenantId, string sessionId, int maxResults = 1000, CancellationToken cancellationToken = default);
+
+        /// <summary>
+        /// Cross-partition scan of all signals whose Azure Tables server-side <c>Timestamp</c>
+        /// is at or after <paramref name="cutoffUtc"/>. Used by the
+        /// <c>IndexReconcileTimer</c> (Plan §M5.d.4) to re-enqueue the last 4h of primary
+        /// rows as a safety-net against queue failures.
+        /// <para>
+        /// <b>Perf note:</b> Azure Tables has no PK-bound filter here, so this is a scan;
+        /// <paramref name="maxResults"/> bounds memory. If the cap is reached the caller
+        /// should log + narrow the window rather than trust completeness.
+        /// </para>
+        /// </summary>
+        Task<List<SignalRecord>> QueryByTimestampAtOrAfterAsync(
+            DateTime cutoffUtc, int maxResults = 50_000, CancellationToken cancellationToken = default);
     }
 }