diff --git a/src/Backend/AutopilotMonitor.Functions.Tests/IndexDualWriteConsistencyTests.cs b/src/Backend/AutopilotMonitor.Functions.Tests/IndexDualWriteConsistencyTests.cs
new file mode 100644
index 00000000..36be3e64
--- /dev/null
+++ b/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;
+
+///
+/// M5 release-gate (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 (PartitionKey, RowKey) via /
+/// key builders.
+///
+/// Also asserts pipeline idempotency: running the same primary row through twice
+/// produces identical upsert keys — the consumer's UpsertReplace batches stay
+/// deduplicating even under timer-triggered resubmits.
+///
+///
+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.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 SessionsByTerminal { get; } = new();
+ public List SessionsByStage { get; } = new();
+ public List DeadEndsByReason { get; } = new();
+ public List ClassifierVerdictsByIdLevel { get; } = new();
+ public List SignalsByKind { get; } = new();
+
+ public Task StoreSessionsByTerminalAsync(IReadOnlyList records, CancellationToken ct = default)
+ { SessionsByTerminal.AddRange(records); return Task.FromResult(records.Count); }
+ public Task StoreSessionsByStageAsync(IReadOnlyList records, CancellationToken ct = default)
+ { SessionsByStage.AddRange(records); return Task.FromResult(records.Count); }
+ public Task StoreDeadEndsByReasonAsync(IReadOnlyList records, CancellationToken ct = default)
+ { DeadEndsByReason.AddRange(records); return Task.FromResult(records.Count); }
+ public Task StoreClassifierVerdictsByIdLevelAsync(IReadOnlyList records, CancellationToken ct = default)
+ { ClassifierVerdictsByIdLevel.AddRange(records); return Task.FromResult(records.Count); }
+ public Task StoreSignalsByKindAsync(IReadOnlyList records, CancellationToken ct = default)
+ { SignalsByKind.AddRange(records); return Task.FromResult(records.Count); }
+ }
+}
diff --git a/src/Backend/AutopilotMonitor.Functions.Tests/IndexReconcileTimerTests.cs b/src/Backend/AutopilotMonitor.Functions.Tests/IndexReconcileTimerTests.cs
new file mode 100644
index 00000000..61e0e993
--- /dev/null
+++ b/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;
+
+///
+/// Tests for the 2h safety-net (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.
+///
+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(MockBehavior.Strict);
+ var transitionRepo = new Mock(MockBehavior.Strict);
+ var producer = new Mock(MockBehavior.Strict);
+
+ var timer = new IndexReconcileTimer(
+ adminConfigSvc,
+ signalRepo.Object,
+ transitionRepo.Object,
+ producer.Object,
+ NullLogger.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
+ {
+ 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
+ {
+ new() { TenantId = TenantId, SessionId = SessionId, StepIndex = 1, Taken = true, FromStage = "A", ToStage = "B", OccurredAtUtc = Cutoff.AddMinutes(7) },
+ };
+
+ var signalRepo = new Mock();
+ signalRepo
+ .Setup(r => r.QueryByTimestampAtOrAfterAsync(Cutoff, It.IsAny(), It.IsAny()))
+ .ReturnsAsync(signals);
+ var transitionRepo = new Mock();
+ transitionRepo
+ .Setup(r => r.QueryByTimestampAtOrAfterAsync(Cutoff, It.IsAny(), It.IsAny()))
+ .ReturnsAsync(transitions);
+
+ List? captured = null;
+ var producer = new Mock();
+ producer
+ .Setup(p => p.EnqueueBatchAsync(It.IsAny>(), It.IsAny()))
+ .Callback, CancellationToken>((e, _) => captured = e.ToList())
+ .ReturnsAsync(3);
+
+ var timer = new IndexReconcileTimer(
+ adminConfigSvc,
+ signalRepo.Object,
+ transitionRepo.Object,
+ producer.Object,
+ NullLogger.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();
+ signalRepo
+ .Setup(r => r.QueryByTimestampAtOrAfterAsync(Cutoff, It.IsAny(), It.IsAny()))
+ .ReturnsAsync(new List());
+ var transitionRepo = new Mock();
+ transitionRepo
+ .Setup(r => r.QueryByTimestampAtOrAfterAsync(Cutoff, It.IsAny(), It.IsAny()))
+ .ReturnsAsync(new List());
+
+ var producer = new Mock(MockBehavior.Strict);
+
+ var timer = new IndexReconcileTimer(
+ adminConfigSvc,
+ signalRepo.Object,
+ transitionRepo.Object,
+ producer.Object,
+ NullLogger.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(
+ Mock.Of(),
+ NullLogger.Instance,
+ new MemoryCache(new MemoryCacheOptions()));
+
+ mock.Setup(s => s.GetConfigurationAsync())
+ .ReturnsAsync(new AdminConfiguration { EnableIndexDualWrite = enableIndexDualWrite });
+
+ return mock.Object;
+ }
+}
diff --git a/src/Backend/AutopilotMonitor.Functions/DataAccess/TableStorage/TableDecisionTransitionRepository.cs b/src/Backend/AutopilotMonitor.Functions/DataAccess/TableStorage/TableDecisionTransitionRepository.cs
index 13874e60..44c091c0 100644
--- a/src/Backend/AutopilotMonitor.Functions/DataAccess/TableStorage/TableDecisionTransitionRepository.cs
+++ b/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> 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(capacity: 128);
+ var pages = table.QueryAsync(
+ 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;
+ }
+
///
/// Projects an Azure back into a ,
/// reassembling chunked PayloadJson if present. Internal for mapping tests.
diff --git a/src/Backend/AutopilotMonitor.Functions/DataAccess/TableStorage/TableSignalRepository.cs b/src/Backend/AutopilotMonitor.Functions/DataAccess/TableStorage/TableSignalRepository.cs
index 298d4e7f..c35ffc60 100644
--- a/src/Backend/AutopilotMonitor.Functions/DataAccess/TableStorage/TableSignalRepository.cs
+++ b/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> 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(capacity: 128);
+ var pages = table.QueryAsync(
+ 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;
+ }
+
///
/// Projects an Azure back into a ,
/// reassembling chunked PayloadJson if present. Internal for mapping tests.
diff --git a/src/Backend/AutopilotMonitor.Functions/Functions/Admin/IndexReconcileTimer.cs b/src/Backend/AutopilotMonitor.Functions/Functions/Admin/IndexReconcileTimer.cs
new file mode 100644
index 00000000..96af808d
--- /dev/null
+++ b/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
+{
+ ///
+ /// 2h safety-net timer (Plan §2.8, §M5.d.4) that recovers missed index-table writes.
+ /// Scans the last 4h of primary Signals + DecisionTransitions rows and
+ /// re-enqueues envelopes onto the telemetry-index-reconcile queue via the
+ /// same producer the hot ingest path uses. The queue-triggered handler (M5.d.3)
+ /// then upserts index rows idempotently.
+ ///
+ /// Why 2h / 4h: 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.
+ ///
+ ///
+ /// Feature gate: .
+ /// When false, the scan is short-circuited — no queries, no enqueues.
+ ///
+ ///
+ 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 * * *";
+
+ /// Safety-net window: scan primaries committed in the last N hours.
+ 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 _logger;
+
+ public IndexReconcileTimer(
+ AdminConfigurationService adminConfig,
+ ISignalRepository signalRepo,
+ IDecisionTransitionRepository transitionRepo,
+ IIndexReconcileProducer producer,
+ ILogger 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);
+ }
+
+ ///
+ /// Testable core. Separated from the TimerTrigger entry so unit tests can drive it
+ /// with a deterministic cutoff without an Azure Functions host.
+ ///
+ 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);
+ }
+ }
+}
diff --git a/src/Backend/AutopilotMonitor.Functions/Services/AdminConfigurationService.cs b/src/Backend/AutopilotMonitor.Functions/Services/AdminConfigurationService.cs
index 9994e904..c55af9db 100644
--- a/src/Backend/AutopilotMonitor.Functions/Services/AdminConfigurationService.cs
+++ b/src/Backend/AutopilotMonitor.Functions/Services/AdminConfigurationService.cs
@@ -30,9 +30,13 @@ namespace AutopilotMonitor.Functions.Services
}
///
- /// Gets global admin configuration (uses cache with 5-minute TTL)
+ /// Gets global admin configuration (uses cache with 5-minute TTL).
+ ///
+ /// Virtual so it can be mocked via Moq in consumer tests that need controlled
+ /// flag values (e.g. IndexReconcileTimer flag-gate tests).
+ ///
///
- public async Task GetConfigurationAsync()
+ public virtual async Task GetConfigurationAsync()
{
if (_cache.TryGetValue(CacheKey, out AdminConfiguration? cachedConfig) && cachedConfig != null)
{
diff --git a/src/Shared/AutopilotMonitor.Shared/DataAccess/IDecisionTransitionRepository.cs b/src/Shared/AutopilotMonitor.Shared/DataAccess/IDecisionTransitionRepository.cs
index 1abdf10a..77a0b766 100644
--- a/src/Shared/AutopilotMonitor.Shared/DataAccess/IDecisionTransitionRepository.cs
+++ b/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
///
Task> QueryBySessionAsync(
string tenantId, string sessionId, int maxResults = 1000, CancellationToken cancellationToken = default);
+
+ ///
+ /// Cross-partition scan of all decision transitions whose Azure Tables server-side
+ /// Timestamp is at or after . Used by the
+ /// IndexReconcileTimer (Plan §M5.d.4) to re-enqueue the last 4h of primary
+ /// rows as a safety-net against queue failures.
+ ///
+ /// Perf note: Azure Tables has no PK-bound filter here, so this is a scan;
+ /// bounds memory. If the cap is reached the caller
+ /// should log + narrow the window rather than trust completeness.
+ ///
+ ///
+ Task> QueryByTimestampAtOrAfterAsync(
+ DateTime cutoffUtc, int maxResults = 50_000, CancellationToken cancellationToken = default);
}
}
diff --git a/src/Shared/AutopilotMonitor.Shared/DataAccess/ISignalRepository.cs b/src/Shared/AutopilotMonitor.Shared/DataAccess/ISignalRepository.cs
index 86107880..edf88c8f 100644
--- a/src/Shared/AutopilotMonitor.Shared/DataAccess/ISignalRepository.cs
+++ b/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
///
Task> QueryBySessionAsync(
string tenantId, string sessionId, int maxResults = 1000, CancellationToken cancellationToken = default);
+
+ ///
+ /// Cross-partition scan of all signals whose Azure Tables server-side Timestamp
+ /// is at or after . Used by the
+ /// IndexReconcileTimer (Plan §M5.d.4) to re-enqueue the last 4h of primary
+ /// rows as a safety-net against queue failures.
+ ///
+ /// Perf note: Azure Tables has no PK-bound filter here, so this is a scan;
+ /// bounds memory. If the cap is reached the caller
+ /// should log + narrow the window rather than trust completeness.
+ ///
+ ///
+ Task> QueryByTimestampAtOrAfterAsync(
+ DateTime cutoffUtc, int maxResults = 50_000, CancellationToken cancellationToken = default);
}
}