backend: remove the V2 index tables and their dual-write pipeline (D-196)

The five cross-session index tables (SessionsByTerminal, SessionsByStage,
DeadEndsByReason, ClassifierVerdictsByIdLevel, SignalsByKind) were write-only
since M5.d: no endpoint, MCP tool or web page ever read them, while the
dual-write cost a serial queue send per signal/transition on the ingest path,
a 5-13 min full scan of Signals + DecisionTransitions every two hours and five
property-filter scans per session deletion (audit 2026-09-05 F05).

Removed: the tables from TableNames.All (schema hash changes, next cold start
runs a full pass without them), the telemetry-index-reconcile queue name,
producer, envelope, handler, queue function and 2h timer, the cross-partition
Timestamp scans on the signal/transition repositories, the EnableIndexDualWrite
flag (model, config mapping, admin UI, generated wire types), cascade steps
11-15 (later steps renumbered 11-18), the deletion wrappers and the offboarding
wipe entries. DiscriminatorPkProp stays as a step class so manifests written
before this change still resume.

Order of operations: deploy first, then delete the queue pair and the five
tables in the portal (B-88h). Deleting the tables before the deploy 500s every
delete preview/enqueue and poisons offboarding.

Submodules: .claude/commands (import_session.py), internal (D-196, docs, plan).

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
This commit is contained in:
Oliver Kieselbach
2026-09-05 04:27:48 +02:00
parent 56d4dd8b84
commit c8ce7cdb37
51 changed files with 50 additions and 2784 deletions
@@ -1,34 +0,0 @@
using AutopilotMonitor.Shared.Models;
namespace AutopilotMonitor.Functions.Tests;
/// <summary>
/// Default-value + round-trip tests for <see cref="AdminConfiguration.EnableIndexDualWrite"/>
/// (Plan §M5.d). The flag gates the index-table dual-write rollout — defaulting to false
/// keeps pre-M5.d behaviour bit-exact and lets operators flip it on deliberately.
/// </summary>
public class AdminConfigurationEnableIndexDualWriteTests
{
[Fact]
public void EnableIndexDualWrite_defaults_to_false_on_new_config()
{
var cfg = new AdminConfiguration();
Assert.False(cfg.EnableIndexDualWrite);
}
[Fact]
public void EnableIndexDualWrite_defaults_to_false_on_CreateDefault()
{
// CreateDefault() is called when no admin-config row exists yet — if it somehow
// enabled dual-write it would break the "default-off" contract immediately.
var cfg = AdminConfiguration.CreateDefault();
Assert.False(cfg.EnableIndexDualWrite);
}
[Fact]
public void EnableIndexDualWrite_true_persists_on_the_config()
{
var cfg = new AdminConfiguration { EnableIndexDualWrite = true };
Assert.True(cfg.EnableIndexDualWrite);
}
}
@@ -46,12 +46,7 @@ public class DeletionManifestBuilderTests
.WithSignals(15)
.WithDecisionTransitions(4)
.WithEventTypeIndex(5)
.WithCveIndex(2)
.WithSessionsByTerminal(1)
.WithSessionsByStage(3)
.WithDeadEndsByReason(0)
.WithClassifierVerdictsByIdLevel(2)
.WithSignalsByKind(11));
.WithCveIndex(2));
var manifest = await NewBuilder(reader).BuildAsync(
TenantId, SessionId, "admin_delete",
@@ -69,15 +64,10 @@ public class DeletionManifestBuilderTests
Assert.Equal(4, c["decisionTransitions"]);
Assert.Equal(5, c["eventTypeIndex"]);
Assert.Equal(2, c["cveIndex"]);
Assert.Equal(1, c["sessionsByTerminal"]);
Assert.Equal(3, c["sessionsByStage"]);
Assert.Equal(0, c["deadEndsByReason"]);
Assert.Equal(2, c["classifierVerdictsByIdLevel"]);
Assert.Equal(11, c["signalsByKind"]);
// 20 cascade tables (incl. SessionTimeBreakdowns F1 PR2, 3 SessionAnnotations lanes +
// SessionTenantLookup P6.1) + tombstone = 21 steps; no inventory steps when no contributions row.
Assert.Equal(21, manifest.Steps.Count);
// 15 cascade tables (incl. SessionTimeBreakdowns F1 PR2, 3 SessionAnnotations lanes +
// SessionTenantLookup P6.1) + tombstone = 16 steps; no inventory steps when no contributions row.
Assert.Equal(16, manifest.Steps.Count);
Assert.Equal(DeletionStepClass.Final, manifest.Steps.Last().Class);
}
@@ -99,11 +89,6 @@ public class DeletionManifestBuilderTests
.WithDecisionTransitions(1, propValue: "transitions_marker")
.WithEventTypeIndex(1, propValue: "eti_marker")
.WithCveIndex(1, propValue: "cve_marker")
.WithSessionsByTerminal(1, propValue: "terminal_marker")
.WithSessionsByStage(1, propValue: "stage_marker")
.WithDeadEndsByReason(1, propValue: "deadend_marker")
.WithClassifierVerdictsByIdLevel(1, propValue: "classifier_marker")
.WithSignalsByKind(1, propValue: "signalkind_marker")
.WithSessionTimeBreakdown(propValue: "breakdown_marker")
.WithSessionAnnotations(propValue: "annotation_marker")
.WithSessionTenantLookup(propValue: "lookup_marker"));
@@ -149,9 +134,9 @@ public class DeletionManifestBuilderTests
new DeletionRetentionContext { TenantRetentionDays = 90 });
var nonTombstoneSteps = manifest.Steps.Where(s => s.Class != DeletionStepClass.Final).ToList();
// Without a contributions row, neither inventory step is emitted → exactly 20 cascade-table steps
// (16 + the 3 SessionAnnotations lane steps + SessionTenantLookup).
Assert.Equal(20, nonTombstoneSteps.Count);
// Without a contributions row, neither inventory step is emitted → exactly 15 cascade-table steps
// (11 + the 3 SessionAnnotations lane steps + SessionTenantLookup).
Assert.Equal(15, nonTombstoneSteps.Count);
foreach (var step in nonTombstoneSteps)
{
Assert.Equal(0, step.RowCount);
@@ -180,8 +165,8 @@ public class DeletionManifestBuilderTests
new DeletionActor { Type = "admin", Actor = "alice@example.com" },
new DeletionRetentionContext { TenantRetentionDays = 90 });
// Expect: 20 cascade + 2 inventory + 1 tombstone = 23 steps.
Assert.Equal(23, manifest.Steps.Count);
// Expect: 15 cascade + 2 inventory + 1 tombstone = 18 steps.
Assert.Equal(18, manifest.Steps.Count);
var aggregate = manifest.Steps.Single(s => s.Class == DeletionStepClass.Aggregate);
Assert.Equal(DeletionStepNames.SoftwareInventoryDecrement, aggregate.Step);
@@ -310,8 +295,7 @@ public class DeletionManifestBuilderTests
.WithSessionsRow()
.WithSessionsIndexRow()
.WithEvents(3)
.WithSignals(5)
.WithSessionsByTerminal(1));
.WithSignals(5));
var first = await NewBuilder(reader).BuildAsync(TenantId, SessionId, "admin_delete",
new DeletionActor { Type = "admin", Actor = "alice@example.com" },
@@ -345,7 +329,6 @@ public class DeletionManifestBuilderTests
.WithSessionsIndexRow()
.WithEvents(2, propValue: "events_check")
.WithRuleResults(1, propValue: "rules_check")
.WithSessionsByTerminal(1, propValue: "terminal_check")
.WithSessionInventoryContributions(MakeContributionsRow(new List<DeletionDecrementKey>
{
new DeletionDecrementKey { Vendor = "Contoso", Name = "App", Version = "1.0" },
@@ -778,12 +761,6 @@ public class DeletionManifestBuilderTests
return this;
}
public ReaderSeed WithSessionsByTerminal(int n, string propValue = "term") => SeedDiscriminatorPkProp(Constants.TableNames.SessionsByTerminal, n, propValue, "Failed");
public ReaderSeed WithSessionsByStage(int n, string propValue = "stg") => SeedDiscriminatorPkProp(Constants.TableNames.SessionsByStage, n, propValue, "AccountSetup");
public ReaderSeed WithDeadEndsByReason(int n, string propValue = "de") => SeedDiscriminatorPkProp(Constants.TableNames.DeadEndsByReason, n, propValue, "Timeout");
public ReaderSeed WithClassifierVerdictsByIdLevel(int n, string propValue = "cls") => SeedDiscriminatorPkProp(Constants.TableNames.ClassifierVerdictsByIdLevel, n, propValue, "EspApps_High");
public ReaderSeed WithSignalsByKind(int n, string propValue = "sk") => SeedDiscriminatorPkProp(Constants.TableNames.SignalsByKind, n, propValue, "EspPhaseChanged");
public ReaderSeed WithSessionAnnotations(string propValue = "anno")
{
// One row per lane — the builder emits one exact-RK step per lane.
@@ -887,22 +864,5 @@ public class DeletionManifestBuilderTests
return this;
}
private ReaderSeed SeedDiscriminatorPkProp(string tableName, int n, string propValue, string discriminator)
{
// PK = {tenantId}_{discriminator}, server-side filter on PK prefix + SessionId == sessionId.
var rows = new List<TableEntity>(n);
for (int i = 0; i < n; i++)
{
var e = new TableEntity($"{TenantId}_{discriminator}", $"{tableName}_{i:D5}")
{
["Marker"] = propValue,
["SessionId"] = SessionId,
};
e.ETag = new ETag("0x" + tableName + i);
rows.Add(e);
}
_query[tableName] = rows;
return this;
}
}
}
@@ -75,7 +75,7 @@ public class HealthCheckServicePoisonQueueTests
var probe = new FakePoisonQueueProbe();
probe.Counts[Constants.QueueNames.AnalyzeOnEnrollmentEnd + "-poison"] = 3;
probe.Counts[Constants.QueueNames.VulnerabilityCorrelate + "-poison"] = 0;
probe.Counts[Constants.QueueNames.TelemetryIndexReconcile + "-poison"] = 0;
probe.Counts[Constants.QueueNames.TenantAutoApprove + "-poison"] = 0;
var svc = BuildService(probe);
var check = await svc.CheckPoisonQueuesAsync();
@@ -94,7 +94,7 @@ public class HealthCheckServicePoisonQueueTests
var probe = new FakePoisonQueueProbe();
probe.Counts[Constants.QueueNames.AnalyzeOnEnrollmentEnd + "-poison"] = 15;
probe.Counts[Constants.QueueNames.VulnerabilityCorrelate + "-poison"] = 0;
probe.Counts[Constants.QueueNames.TelemetryIndexReconcile + "-poison"] = 2;
probe.Counts[Constants.QueueNames.TenantAutoApprove + "-poison"] = 2;
var svc = BuildService(probe);
var check = await svc.CheckPoisonQueuesAsync();
@@ -117,7 +117,7 @@ public class HealthCheckServicePoisonQueueTests
}
};
probe.Counts[Constants.QueueNames.AnalyzeOnEnrollmentEnd + "-poison"] = 0;
probe.Counts[Constants.QueueNames.TelemetryIndexReconcile + "-poison"] = 0;
probe.Counts[Constants.QueueNames.TenantAutoApprove + "-poison"] = 0;
var svc = BuildService(probe);
var check = await svc.CheckPoisonQueuesAsync();
@@ -134,7 +134,7 @@ public class HealthCheckServicePoisonQueueTests
var probe = new FakePoisonQueueProbe();
probe.Counts[Constants.QueueNames.AnalyzeOnEnrollmentEnd + "-poison"] = 4;
probe.Counts[Constants.QueueNames.VulnerabilityCorrelate + "-poison"] = 0;
probe.Counts[Constants.QueueNames.TelemetryIndexReconcile + "-poison"] = 0;
probe.Counts[Constants.QueueNames.TenantAutoApprove + "-poison"] = 0;
// warn=5 → 4 messages now classifies as healthy (would be warning under defaults)
var svc = BuildService(probe, new Dictionary<string, string?>
@@ -153,7 +153,7 @@ public class HealthCheckServicePoisonQueueTests
var probe = new FakePoisonQueueProbe();
probe.Counts[Constants.QueueNames.AnalyzeOnEnrollmentEnd + "-poison"] = 1;
probe.Counts[Constants.QueueNames.VulnerabilityCorrelate + "-poison"] = 0;
probe.Counts[Constants.QueueNames.TelemetryIndexReconcile + "-poison"] = 0;
probe.Counts[Constants.QueueNames.TenantAutoApprove + "-poison"] = 0;
var svc = BuildService(probe);
var check = await svc.CheckPoisonQueuesAsync();
@@ -1,285 +0,0 @@
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); }
}
}
@@ -1,167 +0,0 @@
using AutopilotMonitor.Functions.Services.Indexing;
using AutopilotMonitor.Shared.Models;
namespace AutopilotMonitor.Functions.Tests;
/// <summary>
/// Pure tests for <see cref="IndexReconcileEnvelopeFactory"/> (Plan §2.8, §M5.d).
/// Verifies the shape-only projection from primary record → queue envelope — the
/// consumer (M5.d.3) relies on these fields being copied verbatim.
/// </summary>
public class IndexReconcileEnvelopeFactoryTests
{
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);
// ============================================================ FromSignal
[Fact]
public void FromSignal_copies_discriminators_and_back_ref()
{
var record = new SignalRecord
{
TenantId = TenantId,
SessionId = SessionId,
SessionSignalOrdinal = 42,
SessionTraceOrdinal = 117,
Kind = "EspTerminalFailure",
KindSchemaVersion = 2,
OccurredAtUtc = Occurred,
SourceOrigin = "EspAndHelloTrackerAdapter",
PayloadJson = "{}",
};
var envelope = IndexReconcileEnvelopeFactory.FromSignal(record);
Assert.Equal("1", envelope.EnvelopeVersion);
Assert.Equal("Signal", envelope.SourceKind);
Assert.Equal(TenantId, envelope.TenantId);
Assert.Equal(SessionId, envelope.SessionId);
Assert.Equal(Occurred, envelope.OccurredAtUtc);
Assert.Equal(42L, envelope.SessionSignalOrdinal);
Assert.Equal("EspTerminalFailure", envelope.SignalKind);
Assert.Equal("EspAndHelloTrackerAdapter", envelope.SourceOrigin);
// DecisionTransition-only fields stay null on Signal envelopes.
Assert.Null(envelope.StepIndex);
Assert.Null(envelope.FromStage);
Assert.Null(envelope.ToStage);
Assert.Null(envelope.Taken);
Assert.Null(envelope.IsTerminal);
Assert.Null(envelope.DeadEndReason);
Assert.Null(envelope.ClassifierVerdictId);
Assert.Null(envelope.ClassifierHypothesisLevel);
}
// ============================================================ FromDecisionTransition
[Fact]
public void FromDecisionTransition_taken_terminal_with_classifier_copies_all_discriminators()
{
var record = new DecisionTransitionRecord
{
TenantId = TenantId,
SessionId = SessionId,
StepIndex = 7,
SessionTraceOrdinal = 118,
SignalOrdinalRef = 42,
OccurredAtUtc = Occurred,
Trigger = "EspExiting",
FromStage = "EspInProgress",
ToStage = "WhiteGloveSealed",
Taken = true,
DeadEndReason = null,
ReducerVersion = "1.0.0",
IsTerminal = true,
ClassifierVerdictId = "whiteglove-sealing",
ClassifierHypothesisLevel = "Strong",
PayloadJson = "{}",
};
var envelope = IndexReconcileEnvelopeFactory.FromDecisionTransition(record);
Assert.Equal("1", envelope.EnvelopeVersion);
Assert.Equal("DecisionTransition", envelope.SourceKind);
Assert.Equal(TenantId, envelope.TenantId);
Assert.Equal(SessionId, envelope.SessionId);
Assert.Equal(Occurred, envelope.OccurredAtUtc);
Assert.Equal(7, envelope.StepIndex);
Assert.Equal("EspInProgress", envelope.FromStage);
Assert.Equal("WhiteGloveSealed", envelope.ToStage);
Assert.Equal(true, envelope.Taken);
Assert.Equal(true, envelope.IsTerminal);
Assert.Null(envelope.DeadEndReason);
Assert.Equal("whiteglove-sealing", envelope.ClassifierVerdictId);
Assert.Equal("Strong", envelope.ClassifierHypothesisLevel);
// Signal-only fields stay null on DecisionTransition envelopes.
Assert.Null(envelope.SessionSignalOrdinal);
Assert.Null(envelope.SignalKind);
Assert.Null(envelope.SourceOrigin);
}
[Fact]
public void FromDecisionTransition_dead_end_carries_reason_without_classifier_fields()
{
var record = new DecisionTransitionRecord
{
TenantId = TenantId,
SessionId = SessionId,
StepIndex = 3,
OccurredAtUtc = Occurred,
Trigger = "EspExiting",
FromStage = "EspInProgress",
ToStage = "EspInProgress",
Taken = false,
DeadEndReason = "hybrid_reboot_gate_blocking",
ReducerVersion = "1.0.0",
IsTerminal = false,
PayloadJson = "{}",
};
var envelope = IndexReconcileEnvelopeFactory.FromDecisionTransition(record);
Assert.Equal(false, envelope.Taken);
Assert.Equal(false, envelope.IsTerminal);
Assert.Equal("hybrid_reboot_gate_blocking", envelope.DeadEndReason);
Assert.Null(envelope.ClassifierVerdictId);
Assert.Null(envelope.ClassifierHypothesisLevel);
}
// ============================================================ BuildBatch
[Fact]
public void BuildBatch_preserves_input_order_and_kind_tags()
{
var signals = new List<SignalRecord>
{
new() { TenantId = TenantId, SessionId = SessionId, SessionSignalOrdinal = 1, Kind = "EspPhaseChanged", OccurredAtUtc = Occurred },
new() { TenantId = TenantId, SessionId = SessionId, SessionSignalOrdinal = 2, Kind = "DesktopArrived", OccurredAtUtc = Occurred },
};
var transitions = new List<DecisionTransitionRecord>
{
new() { TenantId = TenantId, SessionId = SessionId, StepIndex = 1, FromStage = "A", ToStage = "B", Taken = true, OccurredAtUtc = Occurred },
};
var envelopes = IndexReconcileEnvelopeFactory.BuildBatch(signals, transitions);
Assert.Equal(3, envelopes.Count);
Assert.Equal("Signal", envelopes[0].SourceKind);
Assert.Equal(1L, envelopes[0].SessionSignalOrdinal);
Assert.Equal("Signal", envelopes[1].SourceKind);
Assert.Equal(2L, envelopes[1].SessionSignalOrdinal);
Assert.Equal("DecisionTransition", envelopes[2].SourceKind);
Assert.Equal(1, envelopes[2].StepIndex);
}
[Fact]
public void BuildBatch_returns_empty_for_null_or_empty_inputs()
{
Assert.Empty(IndexReconcileEnvelopeFactory.BuildBatch(
new List<SignalRecord>(), new List<DecisionTransitionRecord>()));
Assert.Empty(IndexReconcileEnvelopeFactory.BuildBatch(null!, null!));
}
}
@@ -1,102 +0,0 @@
using AutopilotMonitor.Shared.Models;
using Newtonsoft.Json;
namespace AutopilotMonitor.Functions.Tests;
/// <summary>
/// JSON round-trip tests for <see cref="IndexReconcileEnvelope"/> (Plan §2.8, §M5.d).
/// The producer (M5.d.2) serializes with Newtonsoft; the queue-triggered consumer
/// (M5.d.3) must deserialize the same shape back — verified here so the contract
/// doesn't drift silently as fields are added.
/// </summary>
public class IndexReconcileEnvelopeSerializationTests
{
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);
[Fact]
public void Signal_envelope_round_trips_through_newtonsoft()
{
var original = new IndexReconcileEnvelope
{
EnvelopeVersion = "1",
SourceKind = "Signal",
TenantId = TenantId,
SessionId = SessionId,
OccurredAtUtc = Occurred,
SessionSignalOrdinal = 42,
SignalKind = "EspTerminalFailure",
SourceOrigin = "EspAndHelloTrackerAdapter",
};
var json = JsonConvert.SerializeObject(original);
var restored = JsonConvert.DeserializeObject<IndexReconcileEnvelope>(json);
Assert.NotNull(restored);
Assert.Equal("1", restored!.EnvelopeVersion);
Assert.Equal("Signal", restored.SourceKind);
Assert.Equal(TenantId, restored.TenantId);
Assert.Equal(SessionId, restored.SessionId);
Assert.Equal(Occurred, restored.OccurredAtUtc.ToUniversalTime());
Assert.Equal(42L, restored.SessionSignalOrdinal);
Assert.Equal("EspTerminalFailure", restored.SignalKind);
Assert.Equal("EspAndHelloTrackerAdapter", restored.SourceOrigin);
Assert.Null(restored.StepIndex);
Assert.Null(restored.ClassifierVerdictId);
}
[Fact]
public void DecisionTransition_envelope_round_trips_all_discriminators()
{
var original = new IndexReconcileEnvelope
{
EnvelopeVersion = "1",
SourceKind = "DecisionTransition",
TenantId = TenantId,
SessionId = SessionId,
OccurredAtUtc = Occurred,
StepIndex = 7,
FromStage = "EspInProgress",
ToStage = "WhiteGloveSealed",
Taken = true,
IsTerminal = true,
DeadEndReason = null,
ClassifierVerdictId = "whiteglove-sealing",
ClassifierHypothesisLevel = "Strong",
};
var json = JsonConvert.SerializeObject(original);
var restored = JsonConvert.DeserializeObject<IndexReconcileEnvelope>(json);
Assert.NotNull(restored);
Assert.Equal("DecisionTransition", restored!.SourceKind);
Assert.Equal(7, restored.StepIndex);
Assert.Equal("EspInProgress", restored.FromStage);
Assert.Equal("WhiteGloveSealed", restored.ToStage);
Assert.Equal(true, restored.Taken);
Assert.Equal(true, restored.IsTerminal);
Assert.Null(restored.DeadEndReason);
Assert.Equal("whiteglove-sealing", restored.ClassifierVerdictId);
Assert.Equal("Strong", restored.ClassifierHypothesisLevel);
Assert.Null(restored.SessionSignalOrdinal);
}
[Fact]
public void Deserialization_tolerates_missing_optional_fields()
{
// A minimal envelope from a consumer-compatible-but-older writer should still parse.
const string minimal =
"{\"envelopeVersion\":\"1\",\"sourceKind\":\"Signal\",\"tenantId\":\"t\",\"sessionId\":\"s\",\"occurredAtUtc\":\"2026-04-21T14:07:33Z\"}";
var restored = JsonConvert.DeserializeObject<IndexReconcileEnvelope>(minimal);
Assert.NotNull(restored);
Assert.Equal("Signal", restored!.SourceKind);
Assert.Null(restored.SessionSignalOrdinal);
Assert.Null(restored.SignalKind);
Assert.Null(restored.StepIndex);
}
}
@@ -1,321 +0,0 @@
using AutopilotMonitor.Functions.Services.Indexing;
using AutopilotMonitor.Shared.DataAccess;
using AutopilotMonitor.Shared.Models;
using Microsoft.Extensions.Logging.Abstractions;
namespace AutopilotMonitor.Functions.Tests;
/// <summary>
/// Routing tests for <see cref="IndexReconcileHandler"/> (Plan §2.8, §M5.d.3). Verifies
/// the 0–3 fan-out decisions per envelope shape using a capture-only fake repo —
/// the real <c>TableIndexRepository</c> mapping is covered by <see cref="TableIndexRepositoryMappingTests"/>.
/// </summary>
public class IndexReconcileHandlerTests
{
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);
}
// ============================================================ Signal routing
[Fact]
public async Task Signal_envelope_writes_exactly_one_SignalsByKind_row()
{
var (handler, repo) = Sut();
await handler.HandleAsync(new IndexReconcileEnvelope
{
SourceKind = "Signal",
TenantId = TenantId,
SessionId = SessionId,
OccurredAtUtc = Occurred,
SessionSignalOrdinal = 42,
SignalKind = "EspTerminalFailure",
SourceOrigin = "EspAndHelloTrackerAdapter",
});
Assert.Single(repo.SignalsByKind);
var row = repo.SignalsByKind[0];
Assert.Equal(TenantId, row.TenantId);
Assert.Equal(SessionId, row.SessionId);
Assert.Equal("EspTerminalFailure", row.SignalKind);
Assert.Equal(42L, row.SessionSignalOrdinal);
Assert.Equal("EspAndHelloTrackerAdapter", row.SourceOrigin);
Assert.Equal(Occurred, row.OccurredAtUtc);
// Other tables untouched.
Assert.Empty(repo.SessionsByTerminal);
Assert.Empty(repo.SessionsByStage);
Assert.Empty(repo.DeadEndsByReason);
Assert.Empty(repo.ClassifierVerdictsByIdLevel);
}
[Fact]
public async Task Signal_envelope_missing_ordinal_is_dropped()
{
var (handler, repo) = Sut();
await handler.HandleAsync(new IndexReconcileEnvelope
{
SourceKind = "Signal",
TenantId = TenantId,
SessionId = SessionId,
OccurredAtUtc = Occurred,
SessionSignalOrdinal = null,
SignalKind = "EspPhaseChanged",
});
Assert.Empty(repo.SignalsByKind);
}
[Fact]
public async Task Signal_envelope_missing_kind_is_dropped()
{
var (handler, repo) = Sut();
await handler.HandleAsync(new IndexReconcileEnvelope
{
SourceKind = "Signal",
TenantId = TenantId,
SessionId = SessionId,
OccurredAtUtc = Occurred,
SessionSignalOrdinal = 1,
SignalKind = string.Empty,
});
Assert.Empty(repo.SignalsByKind);
}
// ============================================================ DecisionTransition routing — all 4 paths
[Fact]
public async Task DecisionTransition_terminal_taken_with_classifier_writes_three_index_rows()
{
var (handler, repo) = Sut();
// WhiteGlove-sealing: Taken=true, IsTerminal=true, classifier verdict attached.
// Expected: SessionsByTerminal + SessionsByStage + ClassifierVerdictsByIdLevel (3 rows).
await handler.HandleAsync(new IndexReconcileEnvelope
{
SourceKind = "DecisionTransition",
TenantId = TenantId,
SessionId = SessionId,
OccurredAtUtc = Occurred,
StepIndex = 7,
FromStage = "EspInProgress",
ToStage = "WhiteGloveSealed",
Taken = true,
IsTerminal = true,
ClassifierVerdictId = "whiteglove-sealing",
ClassifierHypothesisLevel = "Strong",
});
Assert.Single(repo.SessionsByTerminal);
Assert.Equal("WhiteGloveSealed", repo.SessionsByTerminal[0].TerminalStage);
Assert.Equal(7, repo.SessionsByTerminal[0].StepIndex);
Assert.Single(repo.SessionsByStage);
Assert.Equal("WhiteGloveSealed", repo.SessionsByStage[0].Stage);
Assert.Equal(Occurred, repo.SessionsByStage[0].LastUpdatedUtc);
Assert.Single(repo.ClassifierVerdictsByIdLevel);
Assert.Equal("whiteglove-sealing", repo.ClassifierVerdictsByIdLevel[0].ClassifierId);
Assert.Equal("Strong", repo.ClassifierVerdictsByIdLevel[0].HypothesisLevel);
// Dead-end path not triggered for Taken=true.
Assert.Empty(repo.DeadEndsByReason);
// Signal path never touched on a DecisionTransition envelope.
Assert.Empty(repo.SignalsByKind);
}
[Fact]
public async Task DecisionTransition_taken_non_terminal_no_classifier_writes_only_SessionsByStage()
{
var (handler, repo) = Sut();
await handler.HandleAsync(new IndexReconcileEnvelope
{
SourceKind = "DecisionTransition",
TenantId = TenantId,
SessionId = SessionId,
OccurredAtUtc = Occurred,
StepIndex = 3,
FromStage = "AwaitingEsp",
ToStage = "EspInProgress",
Taken = true,
IsTerminal = false,
});
Assert.Empty(repo.SessionsByTerminal);
Assert.Single(repo.SessionsByStage);
Assert.Empty(repo.DeadEndsByReason);
Assert.Empty(repo.ClassifierVerdictsByIdLevel);
}
[Fact]
public async Task DecisionTransition_dead_end_with_classifier_writes_DeadEndsByReason_and_verdict()
{
var (handler, repo) = Sut();
await handler.HandleAsync(new IndexReconcileEnvelope
{
SourceKind = "DecisionTransition",
TenantId = TenantId,
SessionId = SessionId,
OccurredAtUtc = Occurred,
StepIndex = 5,
FromStage = "EspInProgress",
ToStage = "EspInProgress",
Taken = false,
IsTerminal = false,
DeadEndReason = "hybrid_reboot_gate_blocking",
ClassifierVerdictId = "whiteglove-sealing",
ClassifierHypothesisLevel = "Weak",
});
Assert.Single(repo.DeadEndsByReason);
Assert.Equal("hybrid_reboot_gate_blocking", repo.DeadEndsByReason[0].DeadEndReason);
Assert.Equal("EspInProgress", repo.DeadEndsByReason[0].FromStage);
Assert.Equal("EspInProgress", repo.DeadEndsByReason[0].AttemptedToStage);
Assert.Single(repo.ClassifierVerdictsByIdLevel);
Assert.Equal("Weak", repo.ClassifierVerdictsByIdLevel[0].HypothesisLevel);
// Taken=false → SessionsByStage is NOT updated (no stage entry).
Assert.Empty(repo.SessionsByStage);
Assert.Empty(repo.SessionsByTerminal);
}
[Fact]
public async Task DecisionTransition_dead_end_without_reason_writes_nothing_to_DeadEnds()
{
// Defense in depth: an envelope with Taken=false but no reason should not create a
// DeadEnd row (empty reason would collide into a single PK across unrelated blocks).
var (handler, repo) = Sut();
await handler.HandleAsync(new IndexReconcileEnvelope
{
SourceKind = "DecisionTransition",
TenantId = TenantId,
SessionId = SessionId,
OccurredAtUtc = Occurred,
StepIndex = 1,
FromStage = "A",
ToStage = "A",
Taken = false,
DeadEndReason = null,
});
Assert.Empty(repo.DeadEndsByReason);
}
[Fact]
public async Task DecisionTransition_missing_stepIndex_is_dropped_entirely()
{
var (handler, repo) = Sut();
await handler.HandleAsync(new IndexReconcileEnvelope
{
SourceKind = "DecisionTransition",
TenantId = TenantId,
SessionId = SessionId,
OccurredAtUtc = Occurred,
StepIndex = null,
ToStage = "WhiteGloveSealed",
Taken = true,
IsTerminal = true,
});
Assert.Empty(repo.SessionsByTerminal);
Assert.Empty(repo.SessionsByStage);
Assert.Empty(repo.DeadEndsByReason);
Assert.Empty(repo.ClassifierVerdictsByIdLevel);
}
// ============================================================ Misc
[Fact]
public async Task Unknown_SourceKind_is_logged_and_dropped()
{
var (handler, repo) = Sut();
await handler.HandleAsync(new IndexReconcileEnvelope
{
SourceKind = "FutureThingWeDontKnowAbout",
TenantId = TenantId,
SessionId = SessionId,
OccurredAtUtc = Occurred,
});
Assert.Empty(repo.SignalsByKind);
Assert.Empty(repo.SessionsByTerminal);
Assert.Empty(repo.SessionsByStage);
Assert.Empty(repo.DeadEndsByReason);
Assert.Empty(repo.ClassifierVerdictsByIdLevel);
}
[Fact]
public async Task Null_envelope_is_logged_and_dropped()
{
var (handler, repo) = Sut();
await handler.HandleAsync(null!);
Assert.Empty(repo.SignalsByKind);
}
// ============================================================ FakeIndexTableRepository
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 cancellationToken = default)
{
SessionsByTerminal.AddRange(records);
return Task.FromResult(records.Count);
}
public Task<int> StoreSessionsByStageAsync(
IReadOnlyList<SessionsByStageRecord> records, CancellationToken cancellationToken = default)
{
SessionsByStage.AddRange(records);
return Task.FromResult(records.Count);
}
public Task<int> StoreDeadEndsByReasonAsync(
IReadOnlyList<DeadEndsByReasonRecord> records, CancellationToken cancellationToken = default)
{
DeadEndsByReason.AddRange(records);
return Task.FromResult(records.Count);
}
public Task<int> StoreClassifierVerdictsByIdLevelAsync(
IReadOnlyList<ClassifierVerdictsByIdLevelRecord> records, CancellationToken cancellationToken = default)
{
ClassifierVerdictsByIdLevel.AddRange(records);
return Task.FromResult(records.Count);
}
public Task<int> StoreSignalsByKindAsync(
IReadOnlyList<SignalsByKindRecord> records, CancellationToken cancellationToken = default)
{
SignalsByKind.AddRange(records);
return Task.FromResult(records.Count);
}
}
}
@@ -1,156 +0,0 @@
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;
}
}
@@ -1,153 +0,0 @@
using AutopilotMonitor.Functions.DataAccess.TableStorage;
namespace AutopilotMonitor.Functions.Tests;
/// <summary>
/// Pure PK/RK format tests for <see cref="IndexRowKeys"/> (Plan §2.8 query matrix, §M5.d).
/// These verify the string shape contract the queue-driven index writer + the reconciliation
/// timer both depend on — any drift here silently breaks cross-session queries.
/// </summary>
public class IndexRowKeysTests
{
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);
// ============================================================ SessionsByTerminal
[Fact]
public void SessionsByTerminal_pk_is_tenant_underscore_terminalStage()
{
Assert.Equal(
$"{TenantId}_WhiteGloveSealed",
IndexRowKeys.BuildSessionsByTerminalPk(TenantId, "WhiteGloveSealed"));
}
[Fact]
public void SessionsByTerminal_rk_is_yyyyMMddHHmmss_underscore_sessionId()
{
Assert.Equal(
$"20260421140733_{SessionId}",
IndexRowKeys.BuildSessionsByTerminalRk(Occurred, SessionId));
}
// ============================================================ SessionsByStage
[Fact]
public void SessionsByStage_pk_is_tenant_underscore_stage()
{
Assert.Equal(
$"{TenantId}_EspInProgress",
IndexRowKeys.BuildSessionsByStagePk(TenantId, "EspInProgress"));
}
[Fact]
public void SessionsByStage_rk_pads_ticks_to_19_digits_for_lex_ordering()
{
var rk = IndexRowKeys.BuildSessionsByStageRk(Occurred, SessionId);
// 19-digit zero-padding is the contract — string comparison then matches numeric time order.
var parts = rk.Split('_');
Assert.Equal(19, parts[0].Length);
Assert.Equal(Occurred.Ticks.ToString("D19"), parts[0]);
Assert.Equal(SessionId, parts[1]);
}
[Fact]
public void SessionsByStage_rk_orders_lexicographically_matching_chronology()
{
var earlier = IndexRowKeys.BuildSessionsByStageRk(Occurred, SessionId);
var later = IndexRowKeys.BuildSessionsByStageRk(Occurred.AddSeconds(1), SessionId);
Assert.True(string.CompareOrdinal(earlier, later) < 0);
}
// ============================================================ DeadEndsByReason
[Fact]
public void DeadEndsByReason_pk_is_tenant_underscore_reason()
{
Assert.Equal(
$"{TenantId}_hybrid_reboot_gate_blocking",
IndexRowKeys.BuildDeadEndsByReasonPk(TenantId, "hybrid_reboot_gate_blocking"));
}
[Fact]
public void DeadEndsByReason_rk_combines_timestamp_session_stepIndex_padded_to_6()
{
Assert.Equal(
$"20260421140733_{SessionId}_000042",
IndexRowKeys.BuildDeadEndsByReasonRk(Occurred, SessionId, 42));
}
// ============================================================ ClassifierVerdictsByIdLevel
[Fact]
public void ClassifierVerdictsByIdLevel_pk_is_tenant_underscore_classifierId_underscore_level()
{
Assert.Equal(
$"{TenantId}_whiteglove-sealing_Weak",
IndexRowKeys.BuildClassifierVerdictsByIdLevelPk(TenantId, "whiteglove-sealing", "Weak"));
}
[Fact]
public void ClassifierVerdictsByIdLevel_rk_combines_timestamp_session_stepIndex_padded_to_6()
{
Assert.Equal(
$"20260421140733_{SessionId}_000007",
IndexRowKeys.BuildClassifierVerdictsByIdLevelRk(Occurred, SessionId, 7));
}
// ============================================================ SignalsByKind
[Fact]
public void SignalsByKind_pk_is_tenant_underscore_signalKind()
{
Assert.Equal(
$"{TenantId}_EspTerminalFailure",
IndexRowKeys.BuildSignalsByKindPk(TenantId, "EspTerminalFailure"));
}
[Fact]
public void SignalsByKind_rk_combines_timestamp_session_ordinal_padded_to_10()
{
Assert.Equal(
$"20260421140733_{SessionId}_0000000123",
IndexRowKeys.BuildSignalsByKindRk(Occurred, SessionId, 123));
}
// ============================================================ UTC coercion + sanitizing
[Fact]
public void Timestamp_always_normalizes_to_UTC()
{
// If a caller passes Local-kind DateTime, conversion must still produce the UTC wall clock.
var localKind = DateTime.SpecifyKind(Occurred, DateTimeKind.Local);
var rk = IndexRowKeys.BuildSessionsByTerminalRk(localKind, SessionId);
Assert.Equal($"{localKind.ToUniversalTime():yyyyMMddHHmmss}_{SessionId}", rk);
}
[Theory]
[InlineData("foo/bar", "foo_bar")]
[InlineData("foo\\bar", "foo_bar")]
[InlineData("foo#bar", "foo_bar")]
[InlineData("foo?bar", "foo_bar")]
[InlineData("clean_name", "clean_name")]
[InlineData("", "")]
public void Sanitize_replaces_forbidden_key_chars_with_underscore(string input, string expected)
{
Assert.Equal(expected, IndexRowKeys.Sanitize(input));
}
[Fact]
public void Sanitize_preserves_enum_style_identifiers_unchanged()
{
// Discriminators are overwhelmingly enum names — sanitize must not mangle them.
Assert.Equal("WhiteGloveSealed", IndexRowKeys.Sanitize("WhiteGloveSealed"));
Assert.Equal("EspTerminalFailure", IndexRowKeys.Sanitize("EspTerminalFailure"));
Assert.Equal("hybrid_reboot_gate_blocking", IndexRowKeys.Sanitize("hybrid_reboot_gate_blocking"));
}
}
@@ -106,10 +106,10 @@ public class SessionDeletionGuardTests
.ReturnsAsync(row);
var ex = await Assert.ThrowsAsync<SessionDeletionLockedException>(
() => guard.EnsureWritableAsync(TenantId, SessionId, callerContext: "IndexReconcile"));
() => guard.EnsureWritableAsync(TenantId, SessionId, callerContext: "AnalyzeQueue"));
Assert.Equal(SessionDeletionState.Running, ex.CurrentState);
Assert.Equal("IndexReconcile", ex.CallerContext);
Assert.Equal("AnalyzeQueue", ex.CallerContext);
Assert.Equal("MANIFEST-ABC", ex.ManifestId);
}
@@ -752,15 +752,15 @@ public class SessionDeletionHandlerTests
Rows = new List<DeletionRowDump> { new DeletionRowDump { Pk = $"{TenantId}_install_failed", Rk = $"6299999999999999_{SessionId}" } } },
new DeletionStep { Order = 10, Table = Constants.TableNames.CveIndex, Class = DeletionStepClass.DiscriminatorPkRkExact, RowCount = 1,
Rows = new List<DeletionRowDump> { new DeletionRowDump { Pk = $"{TenantId}_CVE-2024-0001", Rk = SessionId } } },
new DeletionStep { Order = 11, Table = Constants.TableNames.SessionsByTerminal, Class = DeletionStepClass.DiscriminatorPkProp, RowCount = 1,
new DeletionStep { Order = 11, Table = "SessionsByTerminal", Class = DeletionStepClass.DiscriminatorPkProp, RowCount = 1,
Rows = new List<DeletionRowDump> { new DeletionRowDump { Pk = $"{TenantId}_failed", Rk = "rk-1" } } },
new DeletionStep { Order = 12, Table = Constants.TableNames.SessionsByStage, Class = DeletionStepClass.DiscriminatorPkProp, RowCount = 1,
new DeletionStep { Order = 12, Table = "SessionsByStage", Class = DeletionStepClass.DiscriminatorPkProp, RowCount = 1,
Rows = new List<DeletionRowDump> { new DeletionRowDump { Pk = $"{TenantId}_AccountSetup", Rk = "rk-2" } } },
new DeletionStep { Order = 13, Table = Constants.TableNames.DeadEndsByReason, Class = DeletionStepClass.DiscriminatorPkProp, RowCount = 0,
new DeletionStep { Order = 13, Table = "DeadEndsByReason", Class = DeletionStepClass.DiscriminatorPkProp, RowCount = 0,
Rows = new List<DeletionRowDump>() },
new DeletionStep { Order = 14, Table = Constants.TableNames.ClassifierVerdictsByIdLevel, Class = DeletionStepClass.DiscriminatorPkProp, RowCount = 1,
new DeletionStep { Order = 14, Table = "ClassifierVerdictsByIdLevel", Class = DeletionStepClass.DiscriminatorPkProp, RowCount = 1,
Rows = new List<DeletionRowDump> { new DeletionRowDump { Pk = $"{TenantId}_clf_warn", Rk = "rk-3" } } },
new DeletionStep { Order = 15, Table = Constants.TableNames.SignalsByKind, Class = DeletionStepClass.DiscriminatorPkProp, RowCount = 1,
new DeletionStep { Order = 15, Table = "SignalsByKind", Class = DeletionStepClass.DiscriminatorPkProp, RowCount = 1,
Rows = new List<DeletionRowDump> { new DeletionRowDump { Pk = $"{TenantId}_hello", Rk = "rk-4" } } },
new DeletionStep
{
@@ -1,167 +0,0 @@
using AutopilotMonitor.Functions.DataAccess.TableStorage;
using AutopilotMonitor.Shared.Models;
namespace AutopilotMonitor.Functions.Tests;
/// <summary>
/// Pure mapping tests for <see cref="TableIndexRepository"/>'s 5 <c>ToEntity</c> overloads
/// (Plan §2.8, §M5.d). Transaction-batching + Azurite round-trip are out of scope here;
/// this file verifies that every projected column and key lands on the entity as expected.
/// </summary>
public class TableIndexRepositoryMappingTests
{
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);
// ============================================================ SessionsByTerminal
[Fact]
public void SessionsByTerminal_ToEntity_projects_keys_and_typed_columns()
{
var record = new SessionsByTerminalRecord
{
TenantId = TenantId,
SessionId = SessionId,
TerminalStage = "WhiteGloveSealed",
OccurredAtUtc = Occurred,
StepIndex = 12,
};
var entity = TableIndexRepository.ToEntity(record);
Assert.Equal($"{TenantId}_WhiteGloveSealed", entity.PartitionKey);
Assert.Equal($"20260421140733_{SessionId}", entity.RowKey);
Assert.Equal(TenantId, entity.GetString("TenantId"));
Assert.Equal(SessionId, entity.GetString("SessionId"));
Assert.Equal("WhiteGloveSealed", entity.GetString("TerminalStage"));
Assert.Equal(Occurred, entity.GetDateTime("OccurredAtUtc"));
Assert.Equal(12, entity.GetInt32("StepIndex"));
}
// ============================================================ SessionsByStage
[Fact]
public void SessionsByStage_ToEntity_uses_padded_ticks_rowkey_for_lex_ordering()
{
var record = new SessionsByStageRecord
{
TenantId = TenantId,
SessionId = SessionId,
Stage = "EspInProgress",
LastUpdatedUtc = Occurred,
StepIndex = 4,
};
var entity = TableIndexRepository.ToEntity(record);
Assert.Equal($"{TenantId}_EspInProgress", entity.PartitionKey);
Assert.Equal($"{Occurred.Ticks:D19}_{SessionId}", entity.RowKey);
Assert.Equal("EspInProgress", entity.GetString("Stage"));
Assert.Equal(Occurred, entity.GetDateTime("LastUpdatedUtc"));
Assert.Equal(4, entity.GetInt32("StepIndex"));
}
// ============================================================ DeadEndsByReason
[Fact]
public void DeadEndsByReason_ToEntity_includes_attempted_stage_and_from_stage()
{
var record = new DeadEndsByReasonRecord
{
TenantId = TenantId,
SessionId = SessionId,
DeadEndReason = "hybrid_reboot_gate_blocking",
StepIndex = 7,
FromStage = "EspInProgress",
AttemptedToStage = "WhiteGloveSealed",
OccurredAtUtc = Occurred,
};
var entity = TableIndexRepository.ToEntity(record);
Assert.Equal($"{TenantId}_hybrid_reboot_gate_blocking", entity.PartitionKey);
Assert.Equal($"20260421140733_{SessionId}_000007", entity.RowKey);
Assert.Equal("hybrid_reboot_gate_blocking", entity.GetString("DeadEndReason"));
Assert.Equal(7, entity.GetInt32("StepIndex"));
Assert.Equal("EspInProgress", entity.GetString("FromStage"));
Assert.Equal("WhiteGloveSealed", entity.GetString("AttemptedToStage"));
Assert.Equal(Occurred, entity.GetDateTime("OccurredAtUtc"));
}
// ============================================================ ClassifierVerdictsByIdLevel
[Fact]
public void ClassifierVerdictsByIdLevel_ToEntity_combines_classifierId_and_level_in_pk()
{
var record = new ClassifierVerdictsByIdLevelRecord
{
TenantId = TenantId,
SessionId = SessionId,
ClassifierId = "whiteglove-sealing",
HypothesisLevel = "Weak",
StepIndex = 5,
OccurredAtUtc = Occurred,
};
var entity = TableIndexRepository.ToEntity(record);
Assert.Equal($"{TenantId}_whiteglove-sealing_Weak", entity.PartitionKey);
Assert.Equal($"20260421140733_{SessionId}_000005", entity.RowKey);
Assert.Equal("whiteglove-sealing", entity.GetString("ClassifierId"));
Assert.Equal("Weak", entity.GetString("HypothesisLevel"));
Assert.Equal(5, entity.GetInt32("StepIndex"));
Assert.Equal(Occurred, entity.GetDateTime("OccurredAtUtc"));
}
// ============================================================ SignalsByKind
[Fact]
public void SignalsByKind_ToEntity_carries_ordinal_backref_for_primary_row_hop()
{
var record = new SignalsByKindRecord
{
TenantId = TenantId,
SessionId = SessionId,
SignalKind = "EspTerminalFailure",
SessionSignalOrdinal = 987L,
OccurredAtUtc = Occurred,
SourceOrigin = "EspAndHelloTrackerAdapter",
};
var entity = TableIndexRepository.ToEntity(record);
Assert.Equal($"{TenantId}_EspTerminalFailure", entity.PartitionKey);
Assert.Equal($"20260421140733_{SessionId}_0000000987", entity.RowKey);
Assert.Equal("EspTerminalFailure", entity.GetString("SignalKind"));
Assert.Equal(987L, entity.GetInt64("SessionSignalOrdinal"));
Assert.Equal(Occurred, entity.GetDateTime("OccurredAtUtc"));
Assert.Equal("EspAndHelloTrackerAdapter", entity.GetString("SourceOrigin"));
}
// ============================================================ Null-tolerance on optional string columns
[Fact]
public void DeadEndsByReason_ToEntity_tolerates_null_strings_by_defaulting_to_empty()
{
// The repo is called from a queue worker; a malformed upstream message shouldn't NRE the mapper.
var record = new DeadEndsByReasonRecord
{
TenantId = TenantId,
SessionId = SessionId,
DeadEndReason = null!,
StepIndex = 1,
FromStage = null!,
AttemptedToStage = null!,
OccurredAtUtc = Occurred,
};
var entity = TableIndexRepository.ToEntity(record);
Assert.Equal(string.Empty, entity.GetString("DeadEndReason"));
Assert.Equal(string.Empty, entity.GetString("FromStage"));
Assert.Equal(string.Empty, entity.GetString("AttemptedToStage"));
}
}
@@ -46,7 +46,6 @@ namespace AutopilotMonitor.Functions.DataAccess
// V2 Decision Engine repositories (Plan §M5).
services.AddSingleton<ISignalRepository, TableSignalRepository>();
services.AddSingleton<IDecisionTransitionRepository, TableDecisionTransitionRepository>();
services.AddSingleton<IIndexTableRepository, TableIndexRepository>();
// Tenant-offboarding audit (Marker / History / ByTenant pointer).
services.AddSingleton<IOffboardingAuditRepository, TableOffboardingAuditRepository>();
@@ -1,87 +0,0 @@
using System;
using System.Globalization;
namespace AutopilotMonitor.Functions.DataAccess.TableStorage
{
/// <summary>
/// Pure PK/RK builders for the 5 V2 Decision Engine index tables
/// (Plan §2.8 query matrix, §M5.d). State-less + zero-IO so every key shape is
/// unit-testable against canned inputs without an Azurite harness.
/// </summary>
internal static class IndexRowKeys
{
private const string TimestampFormat = "yyyyMMddHHmmss";
// Azure Tables forbids these characters in PK/RK values. Discriminators are typically
// enum names (safe ASCII); the sanitize step is defensive cover for free-form strings
// like DeadEndReason that could in principle include a slash.
private static readonly char[] ForbiddenKeyChars = { '/', '\\', '#', '?' };
// ============================================================ SessionsByTerminal
public static string BuildSessionsByTerminalPk(string tenantId, string terminalStage)
=> $"{tenantId}_{Sanitize(terminalStage)}";
public static string BuildSessionsByTerminalRk(DateTime occurredAtUtc, string sessionId)
=> $"{FormatTimestamp(occurredAtUtc)}_{sessionId}";
// ============================================================ SessionsByStage
public static string BuildSessionsByStagePk(string tenantId, string stage)
=> $"{tenantId}_{Sanitize(stage)}";
/// <summary>
/// Ticks are zero-padded to 19 digits (DateTime.MaxValue has 19-digit ticks) so
/// lexicographic RK ordering matches numeric time order.
/// </summary>
public static string BuildSessionsByStageRk(DateTime lastUpdatedUtc, string sessionId)
=> $"{lastUpdatedUtc.ToUniversalTime().Ticks.ToString("D19", CultureInfo.InvariantCulture)}_{sessionId}";
// ============================================================ DeadEndsByReason
public static string BuildDeadEndsByReasonPk(string tenantId, string deadEndReason)
=> $"{tenantId}_{Sanitize(deadEndReason)}";
public static string BuildDeadEndsByReasonRk(DateTime occurredAtUtc, string sessionId, int stepIndex)
=> $"{FormatTimestamp(occurredAtUtc)}_{sessionId}_{stepIndex.ToString("D6", CultureInfo.InvariantCulture)}";
// ============================================================ ClassifierVerdictsByIdLevel
public static string BuildClassifierVerdictsByIdLevelPk(string tenantId, string classifierId, string hypothesisLevel)
=> $"{tenantId}_{Sanitize(classifierId)}_{Sanitize(hypothesisLevel)}";
public static string BuildClassifierVerdictsByIdLevelRk(DateTime occurredAtUtc, string sessionId, int stepIndex)
=> $"{FormatTimestamp(occurredAtUtc)}_{sessionId}_{stepIndex.ToString("D6", CultureInfo.InvariantCulture)}";
// ============================================================ SignalsByKind
public static string BuildSignalsByKindPk(string tenantId, string signalKind)
=> $"{tenantId}_{Sanitize(signalKind)}";
public static string BuildSignalsByKindRk(DateTime occurredAtUtc, string sessionId, long sessionSignalOrdinal)
=> $"{FormatTimestamp(occurredAtUtc)}_{sessionId}_{sessionSignalOrdinal.ToString("D10", CultureInfo.InvariantCulture)}";
// ============================================================ Helpers
private static string FormatTimestamp(DateTime utc)
=> utc.ToUniversalTime().ToString(TimestampFormat, CultureInfo.InvariantCulture);
/// <summary>
/// Replaces characters Azure Tables disallows in PK/RK values with <c>_</c>. Enum-derived
/// discriminators never trigger this; it's defensive cover for free-form strings.
/// </summary>
internal static string Sanitize(string? value)
{
if (string.IsNullOrEmpty(value)) return string.Empty;
if (value.IndexOfAny(ForbiddenKeyChars) < 0) return value;
var buffer = new char[value.Length];
for (var i = 0; i < value.Length; i++)
{
var c = value[i];
buffer[i] = Array.IndexOf(ForbiddenKeyChars, c) >= 0 ? '_' : c;
}
return new string(buffer);
}
}
}
@@ -901,7 +901,6 @@ namespace AutopilotMonitor.Functions.DataAccess.TableStorage
{ "WhiteGloveSealingPatternIdsJson", config.WhiteGloveSealingPatternIdsJson ?? string.Empty },
{ "NvdApiKey", config.NvdApiKey },
{ "VulnerabilityCorrelationEnabled", config.VulnerabilityCorrelationEnabled },
{ "EnableIndexDualWrite", config.EnableIndexDualWrite },
{ "AutoApproveNewTenants", config.AutoApproveNewTenants },
{ "ImeMsiArchivingEnabled", config.ImeMsiArchivingEnabled },
{ "MaxImeMsiDownloadSizeMB", config.MaxImeMsiDownloadSizeMB },
@@ -973,7 +972,6 @@ namespace AutopilotMonitor.Functions.DataAccess.TableStorage
WhiteGloveSealingPatternIdsJson = entity.GetString("WhiteGloveSealingPatternIdsJson"),
NvdApiKey = entity.GetString("NvdApiKey"),
VulnerabilityCorrelationEnabled = entity.GetBoolean("VulnerabilityCorrelationEnabled") ?? true,
EnableIndexDualWrite = entity.GetBoolean("EnableIndexDualWrite") ?? false,
AutoApproveNewTenants = entity.GetBoolean("AutoApproveNewTenants") ?? false,
ImeMsiArchivingEnabled = entity.GetBoolean("ImeMsiArchivingEnabled") ?? true,
MaxImeMsiDownloadSizeMB = entity.GetInt32("MaxImeMsiDownloadSizeMB") ?? 250,
@@ -107,33 +107,6 @@ 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.
@@ -1,175 +0,0 @@
using System;
using System.Collections.Generic;
using System.Linq;
using System.Threading;
using System.Threading.Tasks;
using Azure.Data.Tables;
using AutopilotMonitor.Functions.Security;
using AutopilotMonitor.Functions.Services;
using AutopilotMonitor.Shared;
using AutopilotMonitor.Shared.DataAccess;
using AutopilotMonitor.Shared.Models;
using Microsoft.Extensions.Logging;
namespace AutopilotMonitor.Functions.DataAccess.TableStorage
{
/// <summary>
/// Azure Tables implementation of <see cref="IIndexTableRepository"/> (Plan §2.8, §M5.d).
/// One class covers all 5 index tables — each table's projection lives in a dedicated
/// <c>ToEntity</c> overload, while the shared batched-transaction write path
/// (group-by-PK, 100-op chunk) mirrors <see cref="TableSignalRepository"/>.
/// </summary>
public sealed class TableIndexRepository : IIndexTableRepository
{
/// <summary>Azure Table Storage limit per entity-group-transaction.</summary>
internal const int TransactionChunkSize = 100;
private readonly TableStorageService _storage;
private readonly ILogger<TableIndexRepository> _logger;
public TableIndexRepository(TableStorageService storage, ILogger<TableIndexRepository> logger)
{
_storage = storage;
_logger = logger;
}
// ============================================================ public Store methods
public Task<int> StoreSessionsByTerminalAsync(
IReadOnlyList<SessionsByTerminalRecord> records,
CancellationToken cancellationToken = default)
=> StoreAsync(Constants.TableNames.SessionsByTerminal, records, ToEntity, cancellationToken);
public Task<int> StoreSessionsByStageAsync(
IReadOnlyList<SessionsByStageRecord> records,
CancellationToken cancellationToken = default)
=> StoreAsync(Constants.TableNames.SessionsByStage, records, ToEntity, cancellationToken);
public Task<int> StoreDeadEndsByReasonAsync(
IReadOnlyList<DeadEndsByReasonRecord> records,
CancellationToken cancellationToken = default)
=> StoreAsync(Constants.TableNames.DeadEndsByReason, records, ToEntity, cancellationToken);
public Task<int> StoreClassifierVerdictsByIdLevelAsync(
IReadOnlyList<ClassifierVerdictsByIdLevelRecord> records,
CancellationToken cancellationToken = default)
=> StoreAsync(Constants.TableNames.ClassifierVerdictsByIdLevel, records, ToEntity, cancellationToken);
public Task<int> StoreSignalsByKindAsync(
IReadOnlyList<SignalsByKindRecord> records,
CancellationToken cancellationToken = default)
=> StoreAsync(Constants.TableNames.SignalsByKind, records, ToEntity, cancellationToken);
// ============================================================ shared write path
private async Task<int> StoreAsync<T>(
string tableName,
IReadOnlyList<T> records,
Func<T, TableEntity> toEntity,
CancellationToken cancellationToken)
{
if (records is null || records.Count == 0) return 0;
var entities = new List<TableEntity>(records.Count);
foreach (var r in records)
{
var entity = toEntity(r);
SecurityValidator.EnsureValidGuid(entity.GetString("TenantId"), "TenantId");
SecurityValidator.EnsureValidGuid(entity.GetString("SessionId"), "SessionId");
entities.Add(entity);
}
var table = _storage.GetTableClient(tableName);
var committed = 0;
foreach (var group in entities.GroupBy(e => e.PartitionKey))
{
// Collapse duplicate RowKeys before building the transaction — Azure Tables
// rejects the whole batch with InvalidDuplicateRow otherwise. Dedup also yields
// RowKey ordering, replacing the previous explicit OrderBy.
var (deduped, dropped) = TableBatchDedup.ByRowKey(group);
if (dropped > 0)
_logger.LogWarning(
"{Table}: dropped {Dropped} duplicate-RowKey index row(s) (last-wins) for partition {Pk}",
tableName, dropped, group.Key);
for (var offset = 0; offset < deduped.Count; offset += TransactionChunkSize)
{
cancellationToken.ThrowIfCancellationRequested();
var chunk = deduped.Skip(offset).Take(TransactionChunkSize).ToList();
var actions = chunk
.Select(e => new TableTransactionAction(TableTransactionActionType.UpsertReplace, e))
.ToList();
await table.SubmitTransactionAsync(actions, cancellationToken).ConfigureAwait(false);
committed += chunk.Count;
}
}
_logger.LogDebug("{Table}: committed {Count} index rows", tableName, committed);
return committed;
}
// ============================================================ per-record entity mapping
internal static TableEntity ToEntity(SessionsByTerminalRecord r) => new(
IndexRowKeys.BuildSessionsByTerminalPk(r.TenantId, r.TerminalStage),
IndexRowKeys.BuildSessionsByTerminalRk(r.OccurredAtUtc, r.SessionId))
{
["TenantId"] = r.TenantId,
["SessionId"] = r.SessionId,
["TerminalStage"] = r.TerminalStage ?? string.Empty,
["OccurredAtUtc"] = r.OccurredAtUtc,
["StepIndex"] = r.StepIndex,
};
internal static TableEntity ToEntity(SessionsByStageRecord r) => new(
IndexRowKeys.BuildSessionsByStagePk(r.TenantId, r.Stage),
IndexRowKeys.BuildSessionsByStageRk(r.LastUpdatedUtc, r.SessionId))
{
["TenantId"] = r.TenantId,
["SessionId"] = r.SessionId,
["Stage"] = r.Stage ?? string.Empty,
["LastUpdatedUtc"] = r.LastUpdatedUtc,
["StepIndex"] = r.StepIndex,
};
internal static TableEntity ToEntity(DeadEndsByReasonRecord r) => new(
IndexRowKeys.BuildDeadEndsByReasonPk(r.TenantId, r.DeadEndReason),
IndexRowKeys.BuildDeadEndsByReasonRk(r.OccurredAtUtc, r.SessionId, r.StepIndex))
{
["TenantId"] = r.TenantId,
["SessionId"] = r.SessionId,
["DeadEndReason"] = r.DeadEndReason ?? string.Empty,
["StepIndex"] = r.StepIndex,
["FromStage"] = r.FromStage ?? string.Empty,
["AttemptedToStage"] = r.AttemptedToStage ?? string.Empty,
["OccurredAtUtc"] = r.OccurredAtUtc,
};
internal static TableEntity ToEntity(ClassifierVerdictsByIdLevelRecord r) => new(
IndexRowKeys.BuildClassifierVerdictsByIdLevelPk(r.TenantId, r.ClassifierId, r.HypothesisLevel),
IndexRowKeys.BuildClassifierVerdictsByIdLevelRk(r.OccurredAtUtc, r.SessionId, r.StepIndex))
{
["TenantId"] = r.TenantId,
["SessionId"] = r.SessionId,
["ClassifierId"] = r.ClassifierId ?? string.Empty,
["HypothesisLevel"] = r.HypothesisLevel ?? string.Empty,
["StepIndex"] = r.StepIndex,
["OccurredAtUtc"] = r.OccurredAtUtc,
};
internal static TableEntity ToEntity(SignalsByKindRecord r) => new(
IndexRowKeys.BuildSignalsByKindPk(r.TenantId, r.SignalKind),
IndexRowKeys.BuildSignalsByKindRk(r.OccurredAtUtc, r.SessionId, r.SessionSignalOrdinal))
{
["TenantId"] = r.TenantId,
["SessionId"] = r.SessionId,
["SignalKind"] = r.SignalKind ?? string.Empty,
["SessionSignalOrdinal"] = r.SessionSignalOrdinal,
["OccurredAtUtc"] = r.OccurredAtUtc,
["SourceOrigin"] = r.SourceOrigin ?? string.Empty,
};
}
}
@@ -116,35 +116,6 @@ 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.
@@ -1,100 +0,0 @@
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);
}
}
}
@@ -4,7 +4,6 @@ using AutopilotMonitor.Functions.Helpers;
using AutopilotMonitor.Functions.Security;
using AutopilotMonitor.Functions.Services;
using AutopilotMonitor.Functions.Services.Deletion;
using AutopilotMonitor.Functions.Services.Indexing;
using AutopilotMonitor.Shared.DataAccess;
using AutopilotMonitor.Shared.Models;
using Microsoft.Azure.Functions.Worker;
@@ -42,7 +41,6 @@ namespace AutopilotMonitor.Functions.Functions.Ingest
private readonly ISessionRepository _sessionRepo;
private readonly ISignalRepository _signalRepo;
private readonly IDecisionTransitionRepository _transitionRepo;
private readonly IIndexReconcileProducer _indexReconcileProducer;
private readonly EventIngestProcessor _eventProcessor;
private readonly TenantConfigurationService _configService;
private readonly AdminConfigurationService _adminConfigService;
@@ -62,7 +60,6 @@ namespace AutopilotMonitor.Functions.Functions.Ingest
ISessionRepository sessionRepo,
ISignalRepository signalRepo,
IDecisionTransitionRepository transitionRepo,
IIndexReconcileProducer indexReconcileProducer,
EventIngestProcessor eventProcessor,
TenantConfigurationService configService,
AdminConfigurationService adminConfigService,
@@ -81,7 +78,6 @@ namespace AutopilotMonitor.Functions.Functions.Ingest
_sessionRepo = sessionRepo;
_signalRepo = signalRepo;
_transitionRepo = transitionRepo;
_indexReconcileProducer = indexReconcileProducer;
_eventProcessor = eventProcessor;
_configService = configService;
_adminConfigService = adminConfigService;
@@ -454,16 +450,6 @@ namespace AutopilotMonitor.Functions.Functions.Ingest
var signalCount = await _signalRepo.StoreBatchAsync(signals);
var transitionCount = await _transitionRepo.StoreBatchAsync(transitions);
// M5.d.2: after the primary commit, fan out one envelope per row onto the
// telemetry-index-reconcile queue for the index-table writer (M5.d.3).
// The producer is feature-flag-gated (AdminConfiguration.EnableIndexDualWrite,
// default off) and never rethrows — see IIndexReconcileProducer contract.
if (signals.Count > 0 || transitions.Count > 0)
{
var envelopes = IndexReconcileEnvelopeFactory.BuildBatch(signals, transitions);
await _indexReconcileProducer.EnqueueBatchAsync(envelopes);
}
int eventCount;
string? adminAction;
List<ServerAction>? pendingActions;
@@ -50,7 +50,7 @@ namespace AutopilotMonitor.Functions.Functions.Maintenance
/// </summary>
public sealed class SessionDeletionMaintenanceFunction
{
// Plan §16 R14: every 12h. const per repo convention (see IndexReconcileTimer.cs:33).
// Plan §16 R14: every 12h. const per repo convention (see SessionSweepFunction.cs).
private const string Cron = "0 0 */12 * * *";
// Plan §10: Preparing-GC threshold. Producer wall-clock for manifest build of even a
@@ -25,7 +25,7 @@ namespace AutopilotMonitor.Functions.Functions.Maintenance
public class SessionSweepFunction
{
// Minute 30 — deliberately offset from the 2h Maintenance timer (minute 0) so the two
// never fire together. const per repo convention (see IndexReconcileTimer.cs).
// never fire together. const per repo convention (see MaintenanceFunction.cs).
private const string Cron = "0 30 * * * *";
private readonly MaintenanceService _maintenanceService;
@@ -1,44 +0,0 @@
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);
}
}
}
@@ -311,19 +311,8 @@ builder.Services.AddSingleton<AutopilotMonitor.Functions.Services.Diagnostics.Se
builder.Services.AddSingleton<SessionReportService>();
builder.Services.AddSingleton<BootstrapSessionService>();
// V2 Decision Engine index-table dual-write producer (Plan §2.8, §M5.d). Gated by
// AdminConfiguration.EnableIndexDualWrite (default false) inside the implementation.
builder.Services.AddSingleton<
AutopilotMonitor.Shared.DataAccess.IIndexReconcileProducer,
AutopilotMonitor.Functions.Services.Indexing.AzureQueueIndexReconcileProducer>();
// V2 Decision Engine index-table reconcile consumer (Plan §M5.d.3). Plain class, not
// interface-abstracted — Cosmos swap would reshape around IIndexTableRepository, not here.
builder.Services.AddSingleton<
AutopilotMonitor.Functions.Services.Indexing.IndexReconcileHandler>();
// Consumer: Functions/Queue/IndexReconcileQueueFunction — a [QueueTrigger] on the
// Constants.QueueNames.TriggerConnection connection (host-resolved app setting
// Queue producers + handlers. Consumers are [QueueTrigger] functions under Functions/Queue/
// 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.
@@ -331,8 +320,8 @@ builder.Services.AddSingleton<
// 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 +
// handler pattern as IndexReconcile above; consumer = Functions/Queue/AnalyzeOnEnrollmentEndQueueFunction.
// rule results un-persisted (manual "Analyze Now" was the only recovery).
// Consumer = Functions/Queue/AnalyzeOnEnrollmentEndQueueFunction.
builder.Services.AddSingleton<
AutopilotMonitor.Functions.Services.Analyze.IAnalyzeOnEnrollmentEndProducer,
AutopilotMonitor.Functions.Services.Analyze.AzureQueueAnalyzeOnEnrollmentEndProducer>();
@@ -33,7 +33,7 @@ namespace AutopilotMonitor.Functions.Services
/// 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).
/// flag values (e.g. queue-function flag-gate tests).
/// </para>
/// </summary>
public virtual async Task<AdminConfiguration> GetConfigurationAsync()
@@ -12,8 +12,8 @@ namespace AutopilotMonitor.Functions.Services.Analyze
{
/// <summary>
/// Azure Storage Queue implementation of <see cref="IAnalyzeOnEnrollmentEndProducer"/>.
/// Mirrors <c>AzureQueueIndexReconcileProducer</c>: Managed Identity preferred, connection
/// string fallback, Base64 encoding (matches the consumer worker).
/// Managed Identity preferred, connection string fallback, Base64 encoding (matches the
/// consumer function).
/// <para>
/// <b>Fault tolerance:</b> queue-side exceptions (network, throttling, queue missing) are
/// logged but never rethrown. The agent's HTTP 200 must not be blocked. A missed enqueue
@@ -98,7 +98,7 @@ namespace AutopilotMonitor.Functions.Services.Deletion
var safeSessionId = ODataSanitizer.EscapeValue(sessionId);
var compositeSessionPk = $"{tenantId}_{sessionId}";
// ---- Steps 1-15: cascade tables, in the §3 order. ----
// ---- Steps 1-10: cascade tables, in the §3 order. ----
await AddPkBySessionStepAsync(manifest, order: 1, table: Constants.TableNames.Events, compositeSessionPk, safeTenantId, safeSessionId, cancellationToken);
await AddPkBySessionStepAsync(manifest, order: 2, table: Constants.TableNames.RuleResults, compositeSessionPk, safeTenantId, safeSessionId, cancellationToken);
await AddPropTenantPkStepAsync(manifest, order: 3, table: Constants.TableNames.AppInstallSummaries, tenantId, sessionId, safeTenantId, safeSessionId, cancellationToken);
@@ -109,39 +109,34 @@ namespace AutopilotMonitor.Functions.Services.Deletion
await AddPkBySessionStepAsync(manifest, order: 8, table: Constants.TableNames.DecisionTransitions, compositeSessionPk, safeTenantId, safeSessionId, cancellationToken);
await AddDiscriminatorPkRkSuffixStepAsync(manifest, order: 9, table: Constants.TableNames.EventTypeIndex, tenantId, sessionId, safeTenantId, cancellationToken);
await AddDiscriminatorPkRkExactStepAsync(manifest, order: 10, table: Constants.TableNames.CveIndex, tenantId, sessionId, safeTenantId, cancellationToken);
await AddDiscriminatorPkPropStepAsync(manifest, order: 11, table: Constants.TableNames.SessionsByTerminal, tenantId, sessionId, safeTenantId, safeSessionId, cancellationToken);
await AddDiscriminatorPkPropStepAsync(manifest, order: 12, table: Constants.TableNames.SessionsByStage, tenantId, sessionId, safeTenantId, safeSessionId, cancellationToken);
await AddDiscriminatorPkPropStepAsync(manifest, order: 13, table: Constants.TableNames.DeadEndsByReason, tenantId, sessionId, safeTenantId, safeSessionId, cancellationToken);
await AddDiscriminatorPkPropStepAsync(manifest, order: 14, table: Constants.TableNames.ClassifierVerdictsByIdLevel, tenantId, sessionId, safeTenantId, safeSessionId, cancellationToken);
await AddDiscriminatorPkPropStepAsync(manifest, order: 15, table: Constants.TableNames.SignalsByKind, tenantId, sessionId, safeTenantId, safeSessionId, cancellationToken);
// ---- Step 16: F1 time-attribution breakdown (PK=tenant, RK=sessionId; PR2). ----
await AddPkRkExactStepAsync(manifest, order: 16, table: Constants.TableNames.SessionTimeBreakdowns, partitionKey: tenantId, rowKey: sessionId, cancellationToken);
// ---- Step 11: F1 time-attribution breakdown (PK=tenant, RK=sessionId; PR2). ----
await AddPkRkExactStepAsync(manifest, order: 11, table: Constants.TableNames.SessionTimeBreakdowns, partitionKey: tenantId, rowKey: sessionId, cancellationToken);
// ---- Steps 17-19: session annotations (PK=tenant, RK="{sessionId}_{lane}"). ----
// ---- Steps 12-14: session annotations (PK=tenant, RK="{sessionId}_{lane}"). ----
// Lanes are a closed set of 3, so exact-RK steps cover every possible row.
var annotationOrder = 17;
var annotationOrder = 12;
foreach (var lane in AnnotationLanes.All)
{
await AddPkRkExactStepAsync(manifest, order: annotationOrder++, table: Constants.TableNames.SessionAnnotations, partitionKey: tenantId, rowKey: $"{sessionId}_{lane}", cancellationToken);
}
// ---- Steps 20 + 21: SoftwareInventory side-row (omit both for pre-side-row sessions). ----
// ---- Steps 15 + 16: SoftwareInventory side-row (omit both for pre-side-row sessions). ----
var contributionsRow = await _reader.GetEntityOrNullAsync(
Constants.TableNames.SessionInventoryContributions, tenantId, sessionId, cancellationToken);
if (contributionsRow != null)
{
AddSoftwareInventoryDecrementStep(manifest, order: 20, contributionsRow);
AddContributionsRowStep(manifest, order: 21, contributionsRow);
AddSoftwareInventoryDecrementStep(manifest, order: 15, contributionsRow);
AddContributionsRowStep(manifest, order: 16, contributionsRow);
}
// ---- Step 22: sessionId → tenantId lookup row (PK=sessionId, RK="tenant"). ----
// ---- Step 17: sessionId → tenantId lookup row (PK=sessionId, RK="tenant"). ----
// Deleted just before the tombstone so cross-tenant resolution keeps working for
// as long as the session itself exists; absent for legacy sessions (0-row step).
await AddPkRkExactStepAsync(manifest, order: 22, table: Constants.TableNames.SessionTenantLookup, partitionKey: sessionId, rowKey: "tenant", cancellationToken);
await AddPkRkExactStepAsync(manifest, order: 17, table: Constants.TableNames.SessionTenantLookup, partitionKey: sessionId, rowKey: "tenant", cancellationToken);
// ---- Step 23: Tombstone (SessionsIndex first, then Sessions). ----
AddTombstoneStep(manifest, order: 23, sessionsIndexRow, sessionRow);
// ---- Step 18: Tombstone (SessionsIndex first, then Sessions). ----
AddTombstoneStep(manifest, order: 18, sessionsIndexRow, sessionRow);
// PreflightCounts derive from each step's RowCount, plus the AGGREGATE decrements length.
manifest.PreflightCounts = ComputePreflightCounts(manifest);
@@ -269,30 +264,6 @@ namespace AutopilotMonitor.Functions.Services.Deletion
});
}
private async Task AddDiscriminatorPkPropStepAsync(
DeletionManifest manifest, int order, string table,
string tenantId, string sessionId, string safeTenantId, string safeSessionId,
CancellationToken cancellationToken)
{
// Server-side: PK prefix scan AND SessionId eq '{sessionId}'. Eligible because every
// DISCRIMINATOR_PK_PROP table writes the SessionId property explicitly (see IndexRowKeys
// call sites; the writers project SessionId so the server-side filter is reliable).
var filter = $"PartitionKey ge '{safeTenantId}_' and PartitionKey lt '{safeTenantId}_~' and SessionId eq '{safeSessionId}'";
var rows = new List<DeletionRowDump>();
await foreach (var entity in _reader.QueryAsync(table, filter, cancellationToken))
{
rows.Add(TableEntityDumpConverter.MapEntityToDump(entity));
}
manifest.Steps.Add(new DeletionStep
{
Order = order,
Table = table,
Class = DeletionStepClass.DiscriminatorPkProp,
RowCount = rows.Count,
Rows = rows,
});
}
private void AddSoftwareInventoryDecrementStep(DeletionManifest manifest, int order, TableEntity contributionsRow)
{
var decrements = DecodeDecrementKeys(contributionsRow);
@@ -11,7 +11,7 @@ namespace AutopilotMonitor.Functions.Services.Deletion
/// existing producer is the production binding, tests substitute a fake.
/// <para>
/// Mirrors the sibling queue producers (<c>IAnalyzeOnEnrollmentEndProducer</c>,
/// <c>IVulnerabilityCorrelateProducer</c>, <c>IIndexReconcileProducer</c>): one method
/// <c>IVulnerabilityCorrelateProducer</c>): one method
/// returning an outcome enum so callers translate to HTTP statuses without coupling to
/// internal CAS / blob mechanics.
/// </para>
@@ -1,113 +0,0 @@
using System;
using System.Collections.Generic;
using System.Threading;
using System.Threading.Tasks;
using Azure.Storage.Queues;
using AutopilotMonitor.Functions.Services.Queueing;
using AutopilotMonitor.Shared;
using AutopilotMonitor.Shared.DataAccess;
using AutopilotMonitor.Shared.Models;
using Microsoft.Extensions.Logging;
using Newtonsoft.Json;
namespace AutopilotMonitor.Functions.Services.Indexing
{
/// <summary>
/// Azure Storage Queue implementation of <see cref="IIndexReconcileProducer"/>
/// (Plan §2.8, §M5.d). Writes one Base64-encoded JSON message per envelope onto
/// the <c>telemetry-index-reconcile</c> queue.
/// <para>
/// <b>Gate:</b> feature flag <c>AdminConfiguration.EnableIndexDualWrite</c> is
/// consulted via <see cref="AdminConfigurationService"/> (5-min memory cache).
/// When false, the producer no-ops — the primary commit has already happened
/// and pre-M5.d behaviour stays bit-exact.
/// </para>
/// <para>
/// <b>Fault tolerance:</b> queue-side exceptions (network, throttling, queue
/// missing) are logged but NOT rethrown. Primary rows are the source of truth;
/// missed envelopes get recovered by the 2h <c>IndexReconcileTimer</c> (M5.d.4).
/// </para>
/// </summary>
public sealed class AzureQueueIndexReconcileProducer : IIndexReconcileProducer
{
private readonly QueueClient _queueClient;
private readonly AdminConfigurationService _adminConfig;
private readonly ILogger<AzureQueueIndexReconcileProducer> _logger;
private int _queueEnsured; // 0 = not yet ensured, 1 = CreateIfNotExistsAsync has run
public AzureQueueIndexReconcileProducer(
QueueClientFactory queueFactory,
AdminConfigurationService adminConfig,
ILogger<AzureQueueIndexReconcileProducer> logger)
{
_adminConfig = adminConfig;
_logger = logger;
// Base64 matches the Azure Functions QueueTrigger default so the M5.d.3 consumer
// binding can decode our messages without an explicit override.
_queueClient = queueFactory.Create(Constants.QueueNames.TelemetryIndexReconcile);
}
public async Task<int> EnqueueBatchAsync(
IReadOnlyList<IndexReconcileEnvelope> envelopes,
CancellationToken cancellationToken = default)
{
if (envelopes is null || envelopes.Count == 0) return 0;
// Flag check — cached 5min inside AdminConfigurationService, cheap per call.
var config = await _adminConfig.GetConfigurationAsync().ConfigureAwait(false);
if (!config.EnableIndexDualWrite)
{
return 0;
}
await EnsureQueueExistsAsync(cancellationToken).ConfigureAwait(false);
var sent = 0;
foreach (var envelope in envelopes)
{
cancellationToken.ThrowIfCancellationRequested();
try
{
var body = JsonConvert.SerializeObject(envelope);
await _queueClient.SendMessageAsync(body, cancellationToken).ConfigureAwait(false);
sent++;
}
catch (Exception ex)
{
_logger.LogWarning(
ex,
"IndexReconcile enqueue failed (source={Source} tenant={Tenant} session={Session}) — timer reconciliation will recover",
envelope.SourceKind, envelope.TenantId, envelope.SessionId);
}
}
if (sent > 0)
{
_logger.LogDebug("IndexReconcile: enqueued {Count} envelope(s)", sent);
}
return sent;
}
private async Task EnsureQueueExistsAsync(CancellationToken cancellationToken)
{
// Cheap double-check: if CreateIfNotExistsAsync has already succeeded once we skip
// the round trip. Interlocked flag keeps concurrent first-callers exclusive without
// introducing a lock on the fast path.
if (_queueEnsured == 1) return;
try
{
await _queueClient.CreateIfNotExistsAsync(cancellationToken: cancellationToken).ConfigureAwait(false);
System.Threading.Interlocked.Exchange(ref _queueEnsured, 1);
}
catch (Exception ex)
{
// Leave _queueEnsured=0 so a later call retries — we'll still attempt SendMessageAsync
// below; if the queue genuinely doesn't exist that call fails + gets logged per-envelope.
_logger.LogWarning(ex, "IndexReconcile queue CreateIfNotExists failed — will retry next batch");
}
}
}
}
@@ -1,75 +0,0 @@
using System.Collections.Generic;
using AutopilotMonitor.Shared.Models;
namespace AutopilotMonitor.Functions.Services.Indexing
{
/// <summary>
/// Pure factory that lifts committed primary rows into <see cref="IndexReconcileEnvelope"/>s
/// for the <c>telemetry-index-reconcile</c> queue (Plan §2.8, §M5.d). State-less + zero-IO
/// so callers can unit-test mapping behaviour against canned inputs without a live queue.
/// <para>
/// One envelope per primary row. The handler (M5.d.3) decides which of the 0–3 index
/// tables to write, so the decision here is shape-only: copy discriminators + back-refs.
/// </para>
/// </summary>
internal static class IndexReconcileEnvelopeFactory
{
public static IndexReconcileEnvelope FromSignal(SignalRecord record)
{
return new IndexReconcileEnvelope
{
EnvelopeVersion = "1",
SourceKind = "Signal",
TenantId = record.TenantId,
SessionId = record.SessionId,
OccurredAtUtc = record.OccurredAtUtc,
SessionSignalOrdinal = record.SessionSignalOrdinal,
SignalKind = record.Kind,
SourceOrigin = record.SourceOrigin,
};
}
public static IndexReconcileEnvelope FromDecisionTransition(DecisionTransitionRecord record)
{
return new IndexReconcileEnvelope
{
EnvelopeVersion = "1",
SourceKind = "DecisionTransition",
TenantId = record.TenantId,
SessionId = record.SessionId,
OccurredAtUtc = record.OccurredAtUtc,
StepIndex = record.StepIndex,
FromStage = record.FromStage,
ToStage = record.ToStage,
Taken = record.Taken,
IsTerminal = record.IsTerminal,
DeadEndReason = record.DeadEndReason,
ClassifierVerdictId = record.ClassifierVerdictId,
ClassifierHypothesisLevel = record.ClassifierHypothesisLevel,
};
}
/// <summary>
/// Builds envelopes for both kinds in one pass, preserving input order. Convenience for
/// <see cref="AutopilotMonitor.Functions.Functions.Ingest.IngestTelemetryFunction"/>,
/// whose per-request batch mixes signals + transitions.
/// </summary>
public static List<IndexReconcileEnvelope> BuildBatch(
IReadOnlyList<SignalRecord> signals,
IReadOnlyList<DecisionTransitionRecord> transitions)
{
var result = new List<IndexReconcileEnvelope>(
(signals?.Count ?? 0) + (transitions?.Count ?? 0));
if (signals != null)
{
foreach (var s in signals) result.Add(FromSignal(s));
}
if (transitions != null)
{
foreach (var t in transitions) result.Add(FromDecisionTransition(t));
}
return result;
}
}
}
@@ -1,203 +0,0 @@
using System.Threading;
using System.Threading.Tasks;
using AutopilotMonitor.Shared.DataAccess;
using AutopilotMonitor.Shared.Models;
using Microsoft.Extensions.Logging;
namespace AutopilotMonitor.Functions.Services.Indexing
{
/// <summary>
/// Fan-out logic for <see cref="IndexReconcileEnvelope"/>s consumed from the
/// <c>telemetry-index-reconcile</c> queue (Plan §2.8, §M5.d.3). Projects a single
/// envelope into the 0–3 applicable index-table rows via <see cref="IIndexTableRepository"/>.
/// <para>
/// <b>Routing:</b>
/// <list type="bullet">
/// <item><c>Signal</c> → <c>SignalsByKind</c> (always, 1 row).</item>
/// <item><c>DecisionTransition</c>, <c>IsTerminal=true</c> → <c>SessionsByTerminal</c>.</item>
/// <item><c>DecisionTransition</c>, <c>Taken=true</c> → <c>SessionsByStage</c>
/// (session entered a new stage; older rows for the same SessionId in other stage
/// PKs remain until the 2h reconcile timer prunes or the next stage entry supersedes
/// them in query-time ordering).</item>
/// <item><c>DecisionTransition</c>, <c>Taken=false</c> + non-null <c>DeadEndReason</c>
/// → <c>DeadEndsByReason</c>.</item>
/// <item><c>DecisionTransition</c> with non-null <c>ClassifierVerdictId</c>
/// → <c>ClassifierVerdictsByIdLevel</c>.</item>
/// </list>
/// </para>
/// <para>
/// <b>Failure semantics:</b> malformed envelopes (unknown <see cref="IndexReconcileEnvelope.SourceKind"/>,
/// missing required discriminators) are logged + dropped — retrying them against a poison
/// queue won't help. Transient Table-Storage errors rethrow, so the QueueTrigger platform
/// retry + poison-queue machinery kicks in.
/// </para>
/// <para>
/// <b>Plain class (no <c>I</c>-interface):</b> repo layer is already DAL-abstracted via
/// <see cref="IIndexTableRepository"/>; wrapping the handler too would be Cosmos-swap bloat.
/// Testability comes from constructor-injecting a fake repo.
/// </para>
/// </summary>
public class IndexReconcileHandler
{
private readonly IIndexTableRepository _indexRepo;
private readonly ILogger<IndexReconcileHandler> _logger;
public IndexReconcileHandler(
IIndexTableRepository indexRepo,
ILogger<IndexReconcileHandler> logger)
{
_indexRepo = indexRepo;
_logger = logger;
}
public Task HandleAsync(IndexReconcileEnvelope envelope, CancellationToken cancellationToken = default)
{
if (envelope is null)
{
_logger.LogWarning("IndexReconcile: null envelope — dropping");
return Task.CompletedTask;
}
return envelope.SourceKind switch
{
"Signal" => HandleSignalAsync(envelope, cancellationToken),
"DecisionTransition" => HandleDecisionTransitionAsync(envelope, cancellationToken),
_ => LogAndDrop(envelope),
};
}
private Task LogAndDrop(IndexReconcileEnvelope envelope)
{
_logger.LogWarning(
"IndexReconcile: unknown SourceKind '{Kind}' (envelopeVersion={Version}) — dropping",
envelope.SourceKind, envelope.EnvelopeVersion);
return Task.CompletedTask;
}
// ============================================================ Signal path
private async Task HandleSignalAsync(IndexReconcileEnvelope e, CancellationToken ct)
{
if (e.SessionSignalOrdinal is not long ordinal || string.IsNullOrEmpty(e.SignalKind))
{
_logger.LogWarning(
"IndexReconcile Signal envelope missing SessionSignalOrdinal or SignalKind (tenant={Tenant} session={Session}) — dropping",
e.TenantId, e.SessionId);
return;
}
await _indexRepo.StoreSignalsByKindAsync(
new[]
{
new SignalsByKindRecord
{
TenantId = e.TenantId,
SessionId = e.SessionId,
SignalKind = e.SignalKind,
SessionSignalOrdinal = ordinal,
OccurredAtUtc = e.OccurredAtUtc,
SourceOrigin = e.SourceOrigin ?? string.Empty,
}
},
ct).ConfigureAwait(false);
}
// ============================================================ DecisionTransition path
private async Task HandleDecisionTransitionAsync(IndexReconcileEnvelope e, CancellationToken ct)
{
if (e.StepIndex is not int stepIndex)
{
_logger.LogWarning(
"IndexReconcile DecisionTransition envelope missing StepIndex (tenant={Tenant} session={Session}) — dropping",
e.TenantId, e.SessionId);
return;
}
var writes = 0;
// SessionsByTerminal — only if a terminal stage was reached.
if (e.IsTerminal == true && !string.IsNullOrEmpty(e.ToStage))
{
await _indexRepo.StoreSessionsByTerminalAsync(
new[]
{
new SessionsByTerminalRecord
{
TenantId = e.TenantId,
SessionId = e.SessionId,
TerminalStage = e.ToStage!,
OccurredAtUtc = e.OccurredAtUtc,
StepIndex = stepIndex,
}
},
ct).ConfigureAwait(false);
writes++;
}
// SessionsByStage — every Taken transition advances the "current stage" row.
if (e.Taken == true && !string.IsNullOrEmpty(e.ToStage))
{
await _indexRepo.StoreSessionsByStageAsync(
new[]
{
new SessionsByStageRecord
{
TenantId = e.TenantId,
SessionId = e.SessionId,
Stage = e.ToStage!,
LastUpdatedUtc = e.OccurredAtUtc,
StepIndex = stepIndex,
}
},
ct).ConfigureAwait(false);
writes++;
}
// DeadEndsByReason — blocked transitions with an explained reason.
if (e.Taken == false && !string.IsNullOrEmpty(e.DeadEndReason))
{
await _indexRepo.StoreDeadEndsByReasonAsync(
new[]
{
new DeadEndsByReasonRecord
{
TenantId = e.TenantId,
SessionId = e.SessionId,
DeadEndReason = e.DeadEndReason!,
StepIndex = stepIndex,
FromStage = e.FromStage ?? string.Empty,
AttemptedToStage = e.ToStage ?? string.Empty,
OccurredAtUtc = e.OccurredAtUtc,
}
},
ct).ConfigureAwait(false);
writes++;
}
// ClassifierVerdictsByIdLevel — any transition step that recorded a classifier verdict.
if (!string.IsNullOrEmpty(e.ClassifierVerdictId))
{
await _indexRepo.StoreClassifierVerdictsByIdLevelAsync(
new[]
{
new ClassifierVerdictsByIdLevelRecord
{
TenantId = e.TenantId,
SessionId = e.SessionId,
ClassifierId = e.ClassifierVerdictId!,
HypothesisLevel = e.ClassifierHypothesisLevel ?? string.Empty,
StepIndex = stepIndex,
OccurredAtUtc = e.OccurredAtUtc,
}
},
ct).ConfigureAwait(false);
writes++;
}
_logger.LogDebug(
"IndexReconcile DecisionTransition handled (tenant={Tenant} session={Session} step={Step}): {Writes} index row(s)",
e.TenantId, e.SessionId, stepIndex, writes);
}
}
}
@@ -127,11 +127,6 @@ namespace AutopilotMonitor.Functions.Services.Offboarding
Constants.TableNames.DecisionTransitions,
Constants.TableNames.EventTypeIndex,
Constants.TableNames.CveIndex,
Constants.TableNames.SessionsByTerminal,
Constants.TableNames.SessionsByStage,
Constants.TableNames.DeadEndsByReason,
Constants.TableNames.ClassifierVerdictsByIdLevel,
Constants.TableNames.SignalsByKind,
};
// Plan §6.3.1 — Variant B (Discriminator + TenantId property).
@@ -379,28 +379,6 @@ namespace AutopilotMonitor.Functions.Services
return DeleteByExactKeysInBatchesAsync(Shared.Constants.TableNames.CveIndex, keys, ct);
}
// ---------------- DISCRIMINATOR_PK_PROP -----------------------------------------------
public Task<DeletionBatchResult> DeleteSessionsByTerminalAsync(
IReadOnlyList<(string Pk, string Rk)> keys, CancellationToken ct = default)
=> DeleteByExactKeysInBatchesAsync(Shared.Constants.TableNames.SessionsByTerminal, keys, ct);
public Task<DeletionBatchResult> DeleteSessionsByStageAsync(
IReadOnlyList<(string Pk, string Rk)> keys, CancellationToken ct = default)
=> DeleteByExactKeysInBatchesAsync(Shared.Constants.TableNames.SessionsByStage, keys, ct);
public Task<DeletionBatchResult> DeleteDeadEndsByReasonAsync(
IReadOnlyList<(string Pk, string Rk)> keys, CancellationToken ct = default)
=> DeleteByExactKeysInBatchesAsync(Shared.Constants.TableNames.DeadEndsByReason, keys, ct);
public Task<DeletionBatchResult> DeleteClassifierVerdictsByIdLevelAsync(
IReadOnlyList<(string Pk, string Rk)> keys, CancellationToken ct = default)
=> DeleteByExactKeysInBatchesAsync(Shared.Constants.TableNames.ClassifierVerdictsByIdLevel, keys, ct);
public Task<DeletionBatchResult> DeleteSignalsByKindAsync(
IReadOnlyList<(string Pk, string Rk)> keys, CancellationToken ct = default)
=> DeleteByExactKeysInBatchesAsync(Shared.Constants.TableNames.SignalsByKind, keys, ct);
// ---------------- PK_RK_EXACT (single-row) --------------------------------------------
public async Task DeleteDeviceSnapshotAsync(string tenantId, string sessionId, CancellationToken ct = default)
@@ -135,8 +135,6 @@ export interface AdminConfiguration {
whiteGloveSealingPatternIdsJson: string;
/** Whether vulnerability correlation is globally enabled. When false, agents still collect inventory but backend skips correlation. Default: true */
vulnerabilityCorrelationEnabled: boolean;
/** Feature flag for V2 Decision Engine index-table dual-write (Plan §M5.d). When true, IngestTelemetryFunction enqueues telemetry-index-reconcile envelopes after committing each primary Signals / DecisionTransitions row; a queue-triggered handler (M5.d.3) then writes the 0–3 applicable index rows. Default: false — enables controlled rollout. The 2h reconcile timer (M5.d.4) is the safety-net for queue failures even once the flag is on. */
enableIndexDualWrite: boolean;
/** When true, new tenant signups are activated automatically ~1 minute after first sign-in (tenant-auto-approve queue worker). When false, every signup waits for a manual Global Admin approval in Tenant Management. Default false — the operator opts into auto-activation and can flip it off at any time (e.g. on abuse) to return to manual vetting; messages already in the queue are then dropped, not parked. Round-tripped via the 4-file web chain (memory feedback_admin_config_ui_roundtrip). */
autoApproveNewTenants: boolean;
/** When true, the first fleet-wide sighting of a new IME version enqueues an archive job (ime-msi-archive queue) that downloads the installer from the CSP-reported URL into the ime-archive blob container. Roughly one download per Microsoft IME release (~monthly). Default true — flip off to pause archiving (queued messages then wait, they are not dropped). Round-tripped via the 4-file web chain (memory feedback_admin_config_ui_roundtrip). */
@@ -1323,16 +1323,6 @@ namespace AutopilotMonitor.Shared
public const string Signals = "Signals";
public const string DecisionTransitions = "DecisionTransitions";
// V2 Decision Engine index tables (Plan §2.8 query matrix + §M5.d). Secondary
// projections written asynchronously via the `telemetry-index-reconcile` queue
// after the primary Signals / DecisionTransitions row has committed. Eventual
// consistency; a 2h timer re-indexes the last 4h as a safety-net on queue failures.
public const string SessionsByTerminal = "SessionsByTerminal";
public const string SessionsByStage = "SessionsByStage";
public const string DeadEndsByReason = "DeadEndsByReason";
public const string ClassifierVerdictsByIdLevel = "ClassifierVerdictsByIdLevel";
public const string SignalsByKind = "SignalsByKind";
// Tenant offboarding audit (cascade-worker plan, Rev 9). Houses three
// PartitionKey patterns side-by-side:
// - "OffboardingMarker" (1 row/tenant, RK=tenantId) active 403-gate marker
@@ -1458,11 +1448,6 @@ namespace AutopilotMonitor.Shared
SessionTombstones,
Signals,
DecisionTransitions,
SessionsByTerminal,
SessionsByStage,
DeadEndsByReason,
ClassifierVerdictsByIdLevel,
SignalsByKind,
OffboardingAudit,
TenantOffboardingCustomsArchive,
ScriptNameCache,
@@ -1587,14 +1572,6 @@ namespace AutopilotMonitor.Shared
/// </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
/// 0–3 applicable index rows. Eventual consistency; the 2h reconcile timer is
/// the safety net on queue failures.
/// </summary>
public const string TelemetryIndexReconcile = "telemetry-index-reconcile";
/// <summary>
/// Auto-analyze fan-out at session end (enrollment_complete / enrollment_failed)
/// and after async vulnerability correlation. Replaces the in-function
@@ -28,19 +28,5 @@ 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);
}
}
@@ -1,29 +0,0 @@
using System.Collections.Generic;
using System.Threading;
using System.Threading.Tasks;
using AutopilotMonitor.Shared.Models;
namespace AutopilotMonitor.Shared.DataAccess
{
/// <summary>
/// Enqueues <see cref="IndexReconcileEnvelope"/>s onto the
/// <c>telemetry-index-reconcile</c> queue (Plan §2.8, §M5.d). Implementations MUST
/// no-op when the <c>AdminConfiguration.EnableIndexDualWrite</c> flag is false —
/// this is the rollout gate that keeps pre-M5.d behaviour bit-exact by default.
/// <para>
/// Exceptions from the underlying queue should NOT propagate back to the ingest
/// path. The primary-row write is the source of truth; index rows are eventually
/// consistent and the 2h reconcile timer (M5.d.4) is the safety net for failures.
/// </para>
/// </summary>
public interface IIndexReconcileProducer
{
/// <summary>
/// Enqueues one queue message per envelope. Returns the number of messages actually
/// sent (0 if the feature flag is off or the batch is empty).
/// </summary>
Task<int> EnqueueBatchAsync(
IReadOnlyList<IndexReconcileEnvelope> envelopes,
CancellationToken cancellationToken = default);
}
}
@@ -1,43 +0,0 @@
using System.Collections.Generic;
using System.Threading;
using System.Threading.Tasks;
using AutopilotMonitor.Shared.Models;
namespace AutopilotMonitor.Shared.DataAccess
{
/// <summary>
/// Write-only repository for the 5 V2 Decision Engine index tables
/// (Plan §2.8 query matrix, §M5.d). Drives async fan-out from the primary
/// <c>Signals</c> / <c>DecisionTransitions</c> rows after they've committed, via the
/// <c>telemetry-index-reconcile</c> queue and the 2h safety-net timer.
/// <para>
/// All writes are idempotent via <c>(PartitionKey, RowKey)</c>: a retried reconcile
/// replays without duplicating rows. <see cref="Azure.Data.Tables.TableTransactionAction"/>
/// batches require a shared PartitionKey, so the implementation groups + chunks input
/// per the 100-op transaction limit. Callers should partition by target table before
/// invoking the corresponding <c>Store…Async</c> method.
/// </para>
/// </summary>
public interface IIndexTableRepository
{
Task<int> StoreSessionsByTerminalAsync(
IReadOnlyList<SessionsByTerminalRecord> records,
CancellationToken cancellationToken = default);
Task<int> StoreSessionsByStageAsync(
IReadOnlyList<SessionsByStageRecord> records,
CancellationToken cancellationToken = default);
Task<int> StoreDeadEndsByReasonAsync(
IReadOnlyList<DeadEndsByReasonRecord> records,
CancellationToken cancellationToken = default);
Task<int> StoreClassifierVerdictsByIdLevelAsync(
IReadOnlyList<ClassifierVerdictsByIdLevelRecord> records,
CancellationToken cancellationToken = default);
Task<int> StoreSignalsByKindAsync(
IReadOnlyList<SignalsByKindRecord> records,
CancellationToken cancellationToken = default);
}
}
@@ -70,19 +70,5 @@ namespace AutopilotMonitor.Shared.DataAccess
int maxResults = 1000,
CancellationToken cancellationToken = default,
long maxTotalPayloadChars = SignalQueryLimits.DefaultMaxTotalPayloadChars);
/// <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);
}
}
@@ -525,16 +525,6 @@ namespace AutopilotMonitor.Shared.Models
/// </summary>
public bool VulnerabilityCorrelationEnabled { get; set; } = true;
/// <summary>
/// Feature flag for V2 Decision Engine index-table dual-write (Plan §M5.d).
/// When true, <c>IngestTelemetryFunction</c> enqueues <c>telemetry-index-reconcile</c>
/// envelopes after committing each primary <c>Signals</c> / <c>DecisionTransitions</c>
/// row; a queue-triggered handler (M5.d.3) then writes the 0–3 applicable index rows.
/// Default: false — enables controlled rollout. The 2h reconcile timer (M5.d.4) is
/// the safety-net for queue failures even once the flag is on.
/// </summary>
public bool EnableIndexDualWrite { get; set; } = false;
/// <summary>
/// When true, new tenant signups are activated automatically ~1 minute after first
/// sign-in (tenant-auto-approve queue worker). When false, every signup waits for a
@@ -10,11 +10,9 @@ namespace AutopilotMonitor.Shared.Models
/// PK = <c>{TenantId}_{SessionId}</c>, RK = <c>{StepIndex:D10}</c>.
/// </para>
/// <para>
/// The index-discriminator columns (<see cref="IsTerminal"/>, <see cref="ClassifierVerdictId"/>,
/// <see cref="ClassifierHypothesisLevel"/>) are projected eagerly so index-table fan-out
/// (SessionsByTerminal / ClassifierVerdictsByIdLevel / …) can happen without re-parsing
/// <see cref="PayloadJson"/>. The primary write stores them inline; the queue-driven
/// index writer reads them straight off the primary row.
/// The discriminator columns (<see cref="IsTerminal"/>, <see cref="ClassifierVerdictId"/>,
/// <see cref="ClassifierHypothesisLevel"/>) are projected eagerly so readers such as the
/// decision-graph builder never re-parse <see cref="PayloadJson"/>.
/// </para>
/// </summary>
public sealed class DecisionTransitionRecord
@@ -1,56 +0,0 @@
using System;
namespace AutopilotMonitor.Shared.Models
{
/// <summary>
/// Queue-message envelope for the <c>telemetry-index-reconcile</c> queue
/// (Plan §2.8, §M5.d). Carries all discriminators needed for the queue-triggered
/// handler (M5.d.3) to fan-out one primary row into the 0–3 applicable index tables,
/// without needing to re-read the primary row from storage.
/// <para>
/// Flat shape (nullable fields per <see cref="SourceKind"/>) deliberately avoids
/// polymorphic Newtonsoft deserialization — see M5.c.3 finding that abstract-type
/// round-trips through JSON need bespoke type-handling configuration, which isn't
/// worth the complexity for a two-variant envelope.
/// </para>
/// </summary>
public sealed class IndexReconcileEnvelope
{
/// <summary>Schema version — bump on breaking envelope changes so consumers can reject or migrate.</summary>
public string EnvelopeVersion { get; set; } = "1";
/// <summary>Either <c>"Signal"</c> or <c>"DecisionTransition"</c>; drives which nullable fields carry data.</summary>
public string SourceKind { get; set; } = string.Empty;
public string TenantId { get; set; } = string.Empty;
public string SessionId { get; set; } = string.Empty;
/// <summary>When the source event happened on the agent clock; primary RK is derived from this.</summary>
public DateTime OccurredAtUtc { get; set; }
// ---------- SourceKind="Signal" fields ----------
/// <summary>Back-ref to <c>Signals</c> primary row (its RK).</summary>
public long? SessionSignalOrdinal { get; set; }
public string? SignalKind { get; set; }
public string? SourceOrigin { get; set; }
// ---------- SourceKind="DecisionTransition" fields ----------
/// <summary>Back-ref to <c>DecisionTransitions</c> primary row (its RK).</summary>
public int? StepIndex { get; set; }
public string? FromStage { get; set; }
public string? ToStage { get; set; }
public bool? Taken { get; set; }
public bool? IsTerminal { get; set; }
public string? DeadEndReason { get; set; }
public string? ClassifierVerdictId { get; set; }
public string? ClassifierHypothesisLevel { get; set; }
}
}
@@ -1,125 +0,0 @@
using System;
namespace AutopilotMonitor.Shared.Models
{
/// <summary>
/// Index-table record projecting a terminal <see cref="DecisionTransitionRecord"/> onto the
/// <c>SessionsByTerminal</c> table (Plan §2.8 query matrix, §M5.d). Written async via the
/// <c>telemetry-index-reconcile</c> queue after the primary <c>DecisionTransitions</c>
/// row has committed; enables "which sessions ended in {TerminalStage}?" queries.
/// <para>
/// PK = <c>{TenantId}_{TerminalStage}</c>, RK = <c>{OccurredAtUtc:yyyyMMddHHmmss}_{SessionId}</c>.
/// </para>
/// </summary>
public sealed class SessionsByTerminalRecord
{
public string TenantId { get; set; } = string.Empty;
public string SessionId { get; set; } = string.Empty;
/// <summary>The terminal <c>SessionStage</c> name (e.g. <c>Completed</c>, <c>Failed</c>, <c>WhiteGloveSealed</c>).</summary>
public string TerminalStage { get; set; } = string.Empty;
public DateTime OccurredAtUtc { get; set; }
/// <summary>Back-ref into the primary <c>DecisionTransitions</c> row so queries can hop to full detail.</summary>
public int StepIndex { get; set; }
}
/// <summary>
/// Index-table record tracking a session's stage history on the <c>SessionsByStage</c> table
/// (Plan §2.8 query matrix, §M5.d). Written per stage-entering transition; older stage rows
/// for the same session remain until queue-side or reconciliation logic removes them.
/// <para>
/// PK = <c>{TenantId}_{Stage}</c>, RK = <c>{LastUpdatedUtcTicks:D19}_{SessionId}</c> —
/// ticks zero-padded to 19 digits so lexicographic RK ordering matches numeric time order.
/// </para>
/// </summary>
public sealed class SessionsByStageRecord
{
public string TenantId { get; set; } = string.Empty;
public string SessionId { get; set; } = string.Empty;
/// <summary>The <c>SessionStage</c> name the session entered with this row.</summary>
public string Stage { get; set; } = string.Empty;
public DateTime LastUpdatedUtc { get; set; }
/// <summary>Back-ref into the primary <c>DecisionTransitions</c> row that moved the session into this stage.</summary>
public int StepIndex { get; set; }
}
/// <summary>
/// Index-table record for dead-end transitions on the <c>DeadEndsByReason</c> table
/// (Plan §2.8 query matrix, §M5.d). Written for <see cref="DecisionTransitionRecord"/>
/// rows with <c>Taken=false</c> and non-null <c>DeadEndReason</c>.
/// <para>
/// PK = <c>{TenantId}_{DeadEndReason}</c>, RK = <c>{OccurredAtUtc:yyyyMMddHHmmss}_{SessionId}_{StepIndex:D6}</c>.
/// </para>
/// </summary>
public sealed class DeadEndsByReasonRecord
{
public string TenantId { get; set; } = string.Empty;
public string SessionId { get; set; } = string.Empty;
public string DeadEndReason { get; set; } = string.Empty;
/// <summary>Back-ref into the primary <c>DecisionTransitions</c> row.</summary>
public int StepIndex { get; set; }
public string FromStage { get; set; } = string.Empty;
/// <summary>The <c>ToStage</c> the transition attempted but was blocked from reaching.</summary>
public string AttemptedToStage { get; set; } = string.Empty;
public DateTime OccurredAtUtc { get; set; }
}
/// <summary>
/// Index-table record projecting classifier verdicts onto the
/// <c>ClassifierVerdictsByIdLevel</c> table (Plan §2.8 query matrix, §M5.d). Written for
/// <see cref="DecisionTransitionRecord"/> rows carrying a non-null <c>ClassifierVerdictId</c>.
/// <para>
/// PK = <c>{TenantId}_{ClassifierId}_{HypothesisLevel}</c>,
/// RK = <c>{OccurredAtUtc:yyyyMMddHHmmss}_{SessionId}_{StepIndex:D6}</c>.
/// </para>
/// </summary>
public sealed class ClassifierVerdictsByIdLevelRecord
{
public string TenantId { get; set; } = string.Empty;
public string SessionId { get; set; } = string.Empty;
public string ClassifierId { get; set; } = string.Empty;
/// <summary>Hypothesis level name (e.g. <c>Strong</c>, <c>Weak</c>, <c>None</c>, <c>Inconclusive</c>).</summary>
public string HypothesisLevel { get; set; } = string.Empty;
/// <summary>Back-ref into the primary <c>DecisionTransitions</c> row.</summary>
public int StepIndex { get; set; }
public DateTime OccurredAtUtc { get; set; }
}
/// <summary>
/// Index-table record projecting signals onto the <c>SignalsByKind</c> table
/// (Plan §2.8 query matrix, §M5.d). Written per <see cref="SignalRecord"/>; enables
/// cross-session "all signals of kind {SignalKind} for tenant {TenantId}" queries.
/// <para>
/// PK = <c>{TenantId}_{SignalKind}</c>,
/// RK = <c>{OccurredAtUtc:yyyyMMddHHmmss}_{SessionId}_{SessionSignalOrdinal:D10}</c>.
/// </para>
/// </summary>
public sealed class SignalsByKindRecord
{
public string TenantId { get; set; } = string.Empty;
public string SessionId { get; set; } = string.Empty;
public string SignalKind { get; set; } = string.Empty;
/// <summary>Back-ref into the primary <c>Signals</c> row.</summary>
public long SessionSignalOrdinal { get; set; }
public DateTime OccurredAtUtc { get; set; }
public string SourceOrigin { get; set; } = string.Empty;
}
}
@@ -58,8 +58,6 @@ interface AdminConfigContextValue {
setAllowAgentDowngrade: (value: boolean) => void;
modernDeploymentHarmlessEventIds: string;
setModernDeploymentHarmlessEventIds: (value: string) => void;
enableIndexDualWrite: boolean;
setEnableIndexDualWrite: (value: boolean) => void;
sessionDeletionKillSwitch: boolean;
setSessionDeletionKillSwitch: (value: boolean) => void;
autoApproveNewTenants: boolean;
@@ -167,7 +165,6 @@ export function AdminConfigProvider({ children }: { children: React.ReactNode })
const [slaNotificationCooldownHours, setSlaNotificationCooldownHours] = useState(24);
const [allowAgentDowngrade, setAllowAgentDowngrade] = useState(false);
const [modernDeploymentHarmlessEventIds, setModernDeploymentHarmlessEventIds] = useState("100, 1005, 1010");
const [enableIndexDualWrite, setEnableIndexDualWrite] = useState(false);
const [sessionDeletionKillSwitch, setSessionDeletionKillSwitch] = useState(false);
const [autoApproveNewTenants, setAutoApproveNewTenants] = useState(false);
const [selfServiceAppHomingEnabled, setSelfServiceAppHomingEnabled] = useState(false);
@@ -240,7 +237,6 @@ export function AdminConfigProvider({ children }: { children: React.ReactNode })
setSlaNotificationCooldownHours(data.slaNotificationCooldownHours ?? 24);
setAllowAgentDowngrade(data.allowAgentDowngrade ?? false);
setModernDeploymentHarmlessEventIds(parseHarmlessEventIdsJson(data.modernDeploymentHarmlessEventIdsJson));
setEnableIndexDualWrite(data.enableIndexDualWrite ?? false);
setSessionDeletionKillSwitch(data.sessionDeletionKillSwitch ?? false);
setAutoApproveNewTenants(data.autoApproveNewTenants ?? false);
setSelfServiceAppHomingEnabled(data.selfServiceAppHomingEnabled ?? false);
@@ -375,7 +371,6 @@ export function AdminConfigProvider({ children }: { children: React.ReactNode })
slaNotificationCooldownHours,
allowAgentDowngrade,
modernDeploymentHarmlessEventIdsJson: serializeHarmlessEventIds(modernDeploymentHarmlessEventIds),
enableIndexDualWrite,
sessionDeletionKillSwitch,
autoApproveNewTenants,
selfServiceAppHomingEnabled,
@@ -407,7 +402,7 @@ export function AdminConfigProvider({ children }: { children: React.ReactNode })
} finally {
setSavingConfig(false);
}
}, [isGlobalAdmin, adminConfig, globalRateLimit, userRateLimit, globalAdminRateLimit, platformStatsBlobSasUrl, agentMigrateApiBaseUrl, agentMigrateTenantOverridesJson, collectorIdleTimeoutMinutes, desktopDetectorNoCandidateTimeoutMinutes, opsEventRetentionDays, slaNotificationCooldownHours, allowAgentDowngrade, modernDeploymentHarmlessEventIds, enableIndexDualWrite, sessionDeletionKillSwitch, autoApproveNewTenants, selfServiceAppHomingEnabled, imeMsiArchivingEnabled, maxImeMsiDownloadSizeMB, getAccessToken]);
}, [isGlobalAdmin, adminConfig, globalRateLimit, userRateLimit, globalAdminRateLimit, platformStatsBlobSasUrl, agentMigrateApiBaseUrl, agentMigrateTenantOverridesJson, collectorIdleTimeoutMinutes, desktopDetectorNoCandidateTimeoutMinutes, opsEventRetentionDays, slaNotificationCooldownHours, allowAgentDowngrade, modernDeploymentHarmlessEventIds, sessionDeletionKillSwitch, autoApproveNewTenants, selfServiceAppHomingEnabled, imeMsiArchivingEnabled, maxImeMsiDownloadSizeMB, getAccessToken]);
// Reset admin config
const handleResetAdminConfig = useCallback(() => {
@@ -424,7 +419,6 @@ export function AdminConfigProvider({ children }: { children: React.ReactNode })
setSlaNotificationCooldownHours(adminConfig.slaNotificationCooldownHours ?? 24);
setAllowAgentDowngrade(adminConfig.allowAgentDowngrade ?? false);
setModernDeploymentHarmlessEventIds(parseHarmlessEventIdsJson(adminConfig.modernDeploymentHarmlessEventIdsJson));
setEnableIndexDualWrite(adminConfig.enableIndexDualWrite ?? false);
setSessionDeletionKillSwitch(adminConfig.sessionDeletionKillSwitch ?? false);
setAutoApproveNewTenants(adminConfig.autoApproveNewTenants ?? false);
setSelfServiceAppHomingEnabled(adminConfig.selfServiceAppHomingEnabled ?? false);
@@ -608,7 +602,6 @@ export function AdminConfigProvider({ children }: { children: React.ReactNode })
slaNotificationCooldownHours, setSlaNotificationCooldownHours,
allowAgentDowngrade, setAllowAgentDowngrade,
modernDeploymentHarmlessEventIds, setModernDeploymentHarmlessEventIds,
enableIndexDualWrite, setEnableIndexDualWrite,
sessionDeletionKillSwitch, setSessionDeletionKillSwitch,
autoApproveNewTenants, setAutoApproveNewTenants,
selfServiceAppHomingEnabled, setSelfServiceAppHomingEnabled,
@@ -29,8 +29,6 @@ interface AdminConfigSettingsSectionProps {
setAllowAgentDowngrade: (value: boolean) => void;
modernDeploymentHarmlessEventIds: string;
setModernDeploymentHarmlessEventIds: (value: string) => void;
enableIndexDualWrite: boolean;
setEnableIndexDualWrite: (value: boolean) => void;
sessionDeletionKillSwitch: boolean;
setSessionDeletionKillSwitch: (value: boolean) => void;
autoApproveNewTenants: boolean;
@@ -71,8 +69,6 @@ export function AdminConfigSettingsSection({
setAllowAgentDowngrade,
modernDeploymentHarmlessEventIds,
setModernDeploymentHarmlessEventIds,
enableIndexDualWrite,
setEnableIndexDualWrite,
sessionDeletionKillSwitch,
setSessionDeletionKillSwitch,
autoApproveNewTenants,
@@ -306,34 +302,6 @@ export function AdminConfigSettingsSection({
</label>
</div>
<div>
<label className="flex items-start space-x-3 cursor-pointer">
<input
type="checkbox"
checked={enableIndexDualWrite}
onChange={(e) => setEnableIndexDualWrite(e.target.checked)}
className="mt-1 h-5 w-5 rounded border-indigo-300 dark:border-indigo-600 text-green-600 focus:ring-green-500"
/>
<span>
<span className="text-indigo-900 dark:text-indigo-100 font-medium">V2 Decision Engine: Enable index-table dual-write</span>
<p className="text-sm text-indigo-800 dark:text-gray-300 mt-1">
Feature flag for the M5.d secondary-index fan-out. When <strong>off</strong> (default), the V2 telemetry
ingest path writes only the primary <code className="text-xs bg-indigo-100 dark:bg-indigo-900 dark:text-indigo-200 px-1 rounded">Signals</code> and
<code className="ml-1 text-xs bg-indigo-100 dark:bg-indigo-900 dark:text-indigo-200 px-1 rounded">DecisionTransitions</code> tables. When
<strong> on</strong>, each committed primary row is also enqueued onto the <code className="text-xs bg-indigo-100 dark:bg-indigo-900 dark:text-indigo-200 px-1 rounded">telemetry-index-reconcile</code> queue
and fanned out into the 5 cross-session index tables
(<code className="text-xs bg-indigo-100 dark:bg-indigo-900 dark:text-indigo-200 px-1 rounded">SessionsByTerminal</code>,
<code className="ml-1 text-xs bg-indigo-100 dark:bg-indigo-900 dark:text-indigo-200 px-1 rounded">SessionsByStage</code>,
<code className="ml-1 text-xs bg-indigo-100 dark:bg-indigo-900 dark:text-indigo-200 px-1 rounded">DeadEndsByReason</code>,
<code className="ml-1 text-xs bg-indigo-100 dark:bg-indigo-900 dark:text-indigo-200 px-1 rounded">ClassifierVerdictsByIdLevel</code>,
<code className="ml-1 text-xs bg-indigo-100 dark:bg-indigo-900 dark:text-indigo-200 px-1 rounded">SignalsByKind</code>).
A 2h timer re-scans the last 4h as a safety net against queue failures. Flip this on only after the
M5.d release-gate tests have been verified in the target environment.
</p>
</span>
</label>
</div>
<div>
<label className="flex items-start space-x-3 cursor-pointer">
<input
@@ -20,7 +20,6 @@ export function SectionGlobalSettings() {
slaNotificationCooldownHours, setSlaNotificationCooldownHours,
allowAgentDowngrade, setAllowAgentDowngrade,
modernDeploymentHarmlessEventIds, setModernDeploymentHarmlessEventIds,
enableIndexDualWrite, setEnableIndexDualWrite,
sessionDeletionKillSwitch, setSessionDeletionKillSwitch,
autoApproveNewTenants, setAutoApproveNewTenants,
selfServiceAppHomingEnabled, setSelfServiceAppHomingEnabled,
@@ -60,8 +59,6 @@ export function SectionGlobalSettings() {
setAllowAgentDowngrade={setAllowAgentDowngrade}
modernDeploymentHarmlessEventIds={modernDeploymentHarmlessEventIds}
setModernDeploymentHarmlessEventIds={setModernDeploymentHarmlessEventIds}
enableIndexDualWrite={enableIndexDualWrite}
setEnableIndexDualWrite={setEnableIndexDualWrite}
sessionDeletionKillSwitch={sessionDeletionKillSwitch}
setSessionDeletionKillSwitch={setSessionDeletionKillSwitch}
autoApproveNewTenants={autoApproveNewTenants}
@@ -50,7 +50,6 @@ export const SHARED_MANIFEST = {
"modernDeploymentHarmlessEventIdsJson",
"whiteGloveSealingPatternIdsJson",
"vulnerabilityCorrelationEnabled",
"enableIndexDualWrite",
"autoApproveNewTenants",
"imeMsiArchivingEnabled",
"maxImeMsiDownloadSizeMB",
@@ -46,7 +46,6 @@
"modernDeploymentHarmlessEventIdsJson",
"whiteGloveSealingPatternIdsJson",
"vulnerabilityCorrelationEnabled",
"enableIndexDualWrite",
"autoApproveNewTenants",
"imeMsiArchivingEnabled",
"maxImeMsiDownloadSizeMB",
@@ -1091,14 +1090,6 @@
},
"doc": "Whether vulnerability correlation is globally enabled. When false, agents still collect inventory but backend skips correlation. Default: true"
},
{
"name": "enableIndexDualWrite",
"type": {
"kind": "primitive",
"name": "boolean"
},
"doc": "Feature flag for V2 Decision Engine index-table dual-write (Plan §M5.d). When true, IngestTelemetryFunction enqueues telemetry-index-reconcile envelopes after committing each primary Signals / DecisionTransitions row; a queue-triggered handler (M5.d.3) then writes the 0–3 applicable index rows. Default: false — enables controlled rollout. The 2h reconcile timer (M5.d.4) is the safety-net for queue failures even once the flag is on."
},
{
"name": "autoApproveNewTenants",
"type": {
@@ -135,8 +135,6 @@ export interface AdminConfiguration {
whiteGloveSealingPatternIdsJson: string;
/** Whether vulnerability correlation is globally enabled. When false, agents still collect inventory but backend skips correlation. Default: true */
vulnerabilityCorrelationEnabled: boolean;
/** Feature flag for V2 Decision Engine index-table dual-write (Plan §M5.d). When true, IngestTelemetryFunction enqueues telemetry-index-reconcile envelopes after committing each primary Signals / DecisionTransitions row; a queue-triggered handler (M5.d.3) then writes the 0–3 applicable index rows. Default: false — enables controlled rollout. The 2h reconcile timer (M5.d.4) is the safety-net for queue failures even once the flag is on. */
enableIndexDualWrite: boolean;
/** When true, new tenant signups are activated automatically ~1 minute after first sign-in (tenant-auto-approve queue worker). When false, every signup waits for a manual Global Admin approval in Tenant Management. Default false — the operator opts into auto-activation and can flip it off at any time (e.g. on abuse) to return to manual vetting; messages already in the queue are then dropped, not parked. Round-tripped via the 4-file web chain (memory feedback_admin_config_ui_roundtrip). */
autoApproveNewTenants: boolean;
/** When true, the first fleet-wide sighting of a new IME version enqueues an archive job (ime-msi-archive queue) that downloads the installer from the CSP-reported URL into the ime-archive blob container. Roughly one download per Microsoft IME release (~monthly). Default true — flip off to pause archiving (queued messages then wait, they are not dropped). Round-tripped via the 4-file web chain (memory feedback_admin_config_ui_roundtrip). */