fix(backend): CMTrace skew tripwire judges only the current ingest era

Three CmTraceTimeSkewRegression fires on 2026-09-01 (one tenant, sessions
e797117b / c06d639d / d7c8032b, agent 2.0.1445, each exactly -60.0 min,
residual 0.0, 85-88 % grid-conformant) all came from the same stale leg: those
sessions are pre-provisioning, their Part 1 ran on 2026-08-20 under agent
2.0.1409 (before per-line anchoring) and left 23-26 IME samples at -60.00,
while the Part 2 leg that completed that morning is clean at -0.16. The
terminal reconcile scans the whole event partition, so the old leg supplied the
majority of the IME samples and fired against a build the device had
self-updated to minutes earlier. Real skew, 12 days old, already fixed - and
>=821 pending pre-provisioning sessions with agent < 1412 are queued to repeat
it.

Event rows carry no agent version, so the split is drawn on ReceivedAt:
ResolveEraStartUtc walks the batch stamps of BOTH sides back from the newest
one and ends the era at the first gap > EraGapHours (2.0) - above any in-leg
upload gap (self-metrics every few minutes, session_stalled at 60 min idle,
maintenance sweep at 2 h), below every pre-provisioning handover. Median, grid
residual, conformity AND the distinct-batch guard now all run on the same
era-filtered sample set; the batch counters used to come from the full scan.
SessionSkewScan carries ReceivedAt per sample and its 20k cap keeps the NEWEST
samples (chunked trim) - keeping the head would have blinded the detector on
exactly the sessions large enough to reach it. The ops event reports
eraStartUtc plus the excluded per-side counts, so an inherited leg stays
visible without being alarmed on.

Tests: 3 new era facts, mutation-proven (EraGapHours = 100000 turns exactly
those three red). The handover fact also asserts that the old leg judged on its
own still fires, so it is the window that silences it, not a moved threshold.
Backend 5334 green, Agent core 2805 green, solution builds with 0 warnings.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
This commit is contained in:
Oliver Kieselbach
2026-09-01 13:53:59 +02:00
parent bac26def52
commit 0a0c323f73
7 changed files with 313 additions and 53 deletions
+17 -1
View File
@@ -4,7 +4,7 @@ title: CMTrace Time Resolution — Per-Line Self-Anchoring
description: How the agent converts bias-less CMTrace local times to UTC without trusting any process's timezone belief — each provably fresh line anchors its own offset against the agent clock on the 15-minute grid; everything not provably fresh falls back uniformly and says so.
resource: src/Agent/AutopilotMonitor.Agent.V2.Core/Monitoring/Enrollment/Ime
tags: [agent, ime, cmtrace, timestamps, timezone, calibration, provenance]
timestamp: 2026-08-30
timestamp: 2026-09-01
---
# Problem
@@ -139,6 +139,22 @@ regression); `measuredWriterOffsetMinutes` is never consulted (sticky after era
flip-backs). Kill switch: app setting `CmTraceSkewTripwireDisabled`. Goal state: the event
never fires — any hit is an anchoring case the per-line design misses, or a detector bug.
Samples are windowed to the session's most recent INGEST ERA before any of that is
computed: the batch stamps of both sides are walked backwards from the newest one and the
era ends at the first `ReceivedAt` gap wider than 2 h. A session partition is not one agent
run — a pre-provisioning session's Part 1 is written by whatever agent build was current
weeks earlier, and its events stay in the partition forever. Field 2026-09-01 (sessions
`e797117b` / `c06d639d` / `d7c8032b`, one tenant): 26 IME samples at exactly −60 min from a
2026-08-20 technician leg under agent 2.0.1409 — before per-line anchoring — outnumbered the
3…9 clean samples of the user leg that completed that morning and fired the tripwire against
2.0.1445, a build the device had self-updated to minutes earlier. The skew was real, 12 days
old, and already fixed; no scan-wide statistic can tell those apart, because event rows carry
no agent version. The era boundary can. `eraStartUtc` and the excluded per-side sample counts
travel in the ops event, so an inherited leg stays visible without being alarmed on. The
2 h threshold sits above any in-leg upload gap (a live agent uploads self-metrics every few
minutes, reports `session_stalled` at 60 min idle, and the maintenance sweep classifies
silence at 2 h) and below every pre-provisioning handover.
Device-clock problems (the customer-actionable cousin) are covered separately by the
`clock_skew` analyze condition behind `ANALYZE-DEV-008`, which excludes IME-derived events
precisely so an anchoring regression can never surface as a customer finding.
+4
View File
@@ -1,5 +1,9 @@
# Log
## 2026-09-01 (1)
* **Update**: `agent/cmtrace-time-resolution.md` — the backend skew tripwire now judges only the session's most recent ingest era (batch stamps walked back from the newest, split at the first `ReceivedAt` gap > 2 h). Pre-provisioning partitions hold a Part 1 leg written weeks earlier by an older agent build: on 2026-09-01 three sessions of one tenant fired on 26 IME samples at −60 min left by agent 2.0.1409 on 2026-08-20 (pre per-line anchoring), against a device already self-updated to 2.0.1445. `SessionSkewScan` carries `ReceivedAt` per sample (distinct-batch counts derive from the era-filtered set; the cap keeps the newest samples), and the ops event reports `eraStartUtc` plus the excluded per-side counts.
## 2026-08-31 (2)
* **Update**: `drift-guards.md` — typisierung follow-up: the deliberately-untyped remainder (RuleDryRun trace, HealthCheck, GlobalNotificationDto, TenantConfigFieldSchema, maintenance run reports, MetricsSummary items, auth/me, auth/mcp) is typed end-to-end; a new object-slot reflection ratchet in `TypedResponseGuardTests` closes the class (baseline: `RuleDryRunCondition.Evidence` only). Deliberate wire change: global-admins / tenant-admins / preview-whitelist rows are flat read models (`GlobalAdminRow` / `TenantAdminRow` / `PreviewWhitelistTenantEntry`) without partitionKey/rowKey/eTag/timestamp, pinned by exact-JSON shape tests.
@@ -6,31 +6,44 @@ namespace AutopilotMonitor.Functions.Tests;
/// <summary>
/// Pins the CMTrace time-skew regression tripwire (CmTraceSkewTripwire.Evaluate): fires only
/// when the IME-vs-other median delta divergence sits on the 15-minute offset grid (residual
/// &lt; 2 min) AND both sides have enough samples over enough distinct upload batches. The
/// tripwire's goal state is to never fire, so the must-stay-silent directions are the primary
/// cases: off-grid replays, sub-grid divergence, thin samples, batch-composition artifacts,
/// and common-mode spool latency must all pass without a hit.
/// &lt; 2 min) AND both sides have enough samples over enough distinct upload batches, judged
/// over the session's most recent ingest era. The tripwire's goal state is to never fire, so
/// the must-stay-silent directions are the primary cases: off-grid replays, sub-grid
/// divergence, thin samples, batch-composition artifacts, common-mode spool latency, and the
/// skewed-but-ancient leg of a resumed pre-provisioning session must all pass without a hit.
/// </summary>
public class CmTraceSkewTripwireTests
{
/// <summary>Reference ingest era used by every single-era scan below.</summary>
private static readonly DateTime EraBase = new DateTime(2026, 9, 1, 8, 20, 0, DateTimeKind.Utc);
/// <summary>Scan with the given per-side medians, comfortably above all sample/batch minimums.</summary>
private static SessionSkewScan MakeScan(
double imeDeltaMinutes, double otherDeltaMinutes,
int imeSamples = 30, int otherSamples = 30,
int imeBatches = 5, int otherBatches = 5)
{
var scan = new SessionSkewScan
{
ImeDistinctBatchCount = imeBatches,
OtherDistinctBatchCount = otherBatches,
};
for (int i = 0; i < imeSamples; i++)
scan.ImeDeltaMinutes.Add(imeDeltaMinutes);
for (int i = 0; i < otherSamples; i++)
scan.OtherDeltaMinutes.Add(otherDeltaMinutes);
var scan = new SessionSkewScan();
AddSamples(scan, isIme: true, imeDeltaMinutes, imeSamples, imeBatches, EraBase);
AddSamples(scan, isIme: false, otherDeltaMinutes, otherSamples, otherBatches, EraBase);
return scan;
}
/// <summary>
/// Adds <paramref name="count"/> identical samples spread round-robin over
/// <paramref name="batches"/> ReceivedAt stamps one minute apart, starting at
/// <paramref name="eraStart"/> — one continuous ingest era.
/// </summary>
private static void AddSamples(
SessionSkewScan scan, bool isIme, double deltaMinutes, int count, int batches, DateTime eraStart)
{
for (int i = 0; i < count; i++)
scan.Add(isIme, deltaMinutes, eraStart.AddMinutes(i % Math.Max(batches, 1)));
}
private static List<double> ImeDeltas(SessionSkewScan scan)
=> scan.ImeSamples.Select(s => s.DeltaMinutes).ToList();
// ── Must fire ──────────────────────────────────────────────────────────
[Fact]
@@ -106,6 +119,7 @@ public class CmTraceSkewTripwireTests
{
Assert.Null(CmTraceSkewTripwire.Evaluate(null));
Assert.Null(CmTraceSkewTripwire.Evaluate(new SessionSkewScan()));
Assert.Null(CmTraceSkewTripwire.ResolveEraStartUtc(new SessionSkewScan()));
}
[Fact]
@@ -114,8 +128,8 @@ public class CmTraceSkewTripwireTests
// A handful of replayed old lines (huge off-grid deltas) inside an otherwise healthy
// IME stream must not drag the median onto the grid.
var scan = MakeScan(imeDeltaMinutes: 0.5, otherDeltaMinutes: 0.4);
scan.ImeDeltaMinutes.Add(-1327.0);
scan.ImeDeltaMinutes.Add(-738.0);
scan.Add(isIme: true, -1327.0, EraBase);
scan.Add(isIme: true, -738.0, EraBase);
Assert.Null(CmTraceSkewTripwire.Evaluate(scan));
}
@@ -126,21 +140,23 @@ public class CmTraceSkewTripwireTests
// re-tailed the IME log and uploaded 508 lines in one burst with a continuum of ages
// (46…733 min, quartiles 52 / 92 / 100) — median 91 min, i.e. 6×15 by pure chance.
// The other side was live. The per-sample grid-conformity guard must keep this silent.
// The burst arrives in ONE batch, inside the same era as the live traffic.
var scan = MakeScan(imeDeltaMinutes: 0.1, otherDeltaMinutes: 0.1, imeSamples: 100, imeBatches: 80);
var burstReceivedAt = EraBase.AddMinutes(80);
void AddRamp(double from, double to, int count)
{
for (int i = 0; i < count; i++)
scan.ImeDeltaMinutes.Add(from + i * (to - from) / count);
scan.Add(isIme: true, from + i * (to - from) / count, burstReceivedAt);
}
AddRamp(46.0, 88.0, 125);
AddRamp(88.0, 100.0, 250);
AddRamp(100.0, 733.0, 125);
var medianIme = scan.ImeDeltaMinutes.OrderBy(v => v).ElementAt(scan.ImeDeltaMinutes.Count / 2);
var medianIme = ImeDeltas(scan).OrderBy(v => v).ElementAt(scan.ImeSamples.Count / 2);
Assert.InRange(medianIme, 88.0, 92.0); // sanity: the median really does sit near 6×15
Assert.Null(CmTraceSkewTripwire.Evaluate(scan));
Assert.True(CmTraceSkewTripwire.ComputeGridConformantFraction(scan.ImeDeltaMinutes, 0.1) < 0.5);
Assert.True(CmTraceSkewTripwire.ComputeGridConformantFraction(ImeDeltas(scan), 0.1) < 0.5);
}
[Fact]
@@ -150,7 +166,7 @@ public class CmTraceSkewTripwireTests
// zone. Every line is individually on the grid, so the conformity guard must NOT mask it.
var scan = MakeScan(imeDeltaMinutes: -420.0, otherDeltaMinutes: 0.0, imeSamples: 65);
for (int i = 0; i < 35; i++)
scan.ImeDeltaMinutes.Add(120.0);
scan.Add(isIme: true, 120.0, EraBase);
var result = CmTraceSkewTripwire.Evaluate(scan);
Assert.NotNull(result);
@@ -166,6 +182,105 @@ public class CmTraceSkewTripwireTests
Assert.Equal(0, CmTraceSkewTripwire.ComputeGridConformantFraction(new List<double>(), 0.0));
}
// ── Ingest-era window ──────────────────────────────────────────────────
private static readonly DateTime PartOneLeg = new DateTime(2026, 8, 20, 8, 20, 0, DateTimeKind.Utc);
private static readonly DateTime PartTwoLeg = new DateTime(2026, 9, 1, 8, 20, 0, DateTimeKind.Utc);
[Fact]
public void Silent_OnPreProvisioningHandoverWhereOnlyTheOldLegWasSkewed()
{
// Field 2026-09-01, three sessions of one tenant (e797117b / c06d639d / d7c8032b):
// Part 1 ran on 2026-08-20 under agent 2.0.1409 — before per-line anchoring — and left
// 26 IME samples at exactly −60 min; Part 2 ran 12 days later on 2.0.1445 and is clean.
// The stale leg outnumbers the fresh one 26:9 and is perfectly grid-conformant, so
// nothing but the era window can separate them (event rows carry no agent version).
var scan = new SessionSkewScan();
AddSamples(scan, isIme: true, -60.0, count: 26, batches: 20, eraStart: PartOneLeg);
AddSamples(scan, isIme: false, 0.0, count: 116, batches: 60, eraStart: PartOneLeg);
AddSamples(scan, isIme: true, -0.16, count: 9, batches: 6, eraStart: PartTwoLeg);
AddSamples(scan, isIme: false, -0.26, count: 95, batches: 40, eraStart: PartTwoLeg);
Assert.Null(CmTraceSkewTripwire.Evaluate(scan));
Assert.Equal(PartTwoLeg, CmTraceSkewTripwire.ResolveEraStartUtc(scan));
// …and the old leg really is what used to fire: judged on its own it still does.
var oldLegOnly = new SessionSkewScan();
AddSamples(oldLegOnly, isIme: true, -60.0, count: 26, batches: 20, eraStart: PartOneLeg);
AddSamples(oldLegOnly, isIme: false, 0.0, count: 116, batches: 60, eraStart: PartOneLeg);
var legacy = CmTraceSkewTripwire.Evaluate(oldLegOnly);
Assert.NotNull(legacy);
Assert.Equal(-4, legacy!.GridSteps);
}
[Fact]
public void Fires_WhenTheCurrentEraIsSkewed_AndReportsWhatTheOlderLegContributed()
{
// The window must not become a blanket mute for resumed sessions: a skew in the leg the
// running agent just produced still fires, and the report says how much history it left
// out so an operator can tell a fresh regression from an inherited one.
var scan = new SessionSkewScan();
AddSamples(scan, isIme: true, 0.1, count: 30, batches: 10, eraStart: PartOneLeg);
AddSamples(scan, isIme: false, 0.1, count: 30, batches: 10, eraStart: PartOneLeg);
AddSamples(scan, isIme: true, -60.0, count: 25, batches: 8, eraStart: PartTwoLeg);
AddSamples(scan, isIme: false, 0.0, count: 40, batches: 12, eraStart: PartTwoLeg);
var result = CmTraceSkewTripwire.Evaluate(scan);
Assert.NotNull(result);
Assert.Equal(-4, result!.GridSteps);
Assert.Equal(-60.0, result.DiffMinutes, 3);
Assert.Equal(25, result.ImeSampleCount);
Assert.Equal(40, result.OtherSampleCount);
Assert.Equal(8, result.ImeBatchCount);
Assert.Equal(12, result.OtherBatchCount);
Assert.Equal(PartTwoLeg, result.EraStartUtc);
Assert.Equal(30, result.ImeSamplesOutsideEra);
Assert.Equal(30, result.OtherSamplesOutsideEra);
}
[Fact]
public void EraWindow_SplitsOnGapsWiderThanTheThreshold_NotOnTotalSpan()
{
// A long leg that keeps uploading is ONE era even when it spans far more than the gap
// threshold end to end — the boundary is the gap between consecutive batches.
var start = PartTwoLeg;
var continuous = new SessionSkewScan();
for (int i = 0; i < 8; i++)
continuous.Add(isIme: false, 0.0, start.AddMinutes(30 * i)); // 3.5 h span, 30 min gaps
Assert.Equal(start, CmTraceSkewTripwire.ResolveEraStartUtc(continuous));
// Exactly at the threshold is still one era (the split needs a STRICTLY wider gap).
var atThreshold = new SessionSkewScan();
atThreshold.Add(isIme: false, 0.0, start);
atThreshold.Add(isIme: false, 0.0, start.AddHours(CmTraceSkewTripwire.EraGapHours));
Assert.Equal(start, CmTraceSkewTripwire.ResolveEraStartUtc(atThreshold));
// One silence past the threshold ends the era, and the boundary is found across BOTH
// sides — an IME-only batch after the gap must still open the new era.
var split = new SessionSkewScan();
split.Add(isIme: false, 0.0, start);
split.Add(isIme: true, 0.0, start.AddHours(3));
Assert.Equal(start.AddHours(3), CmTraceSkewTripwire.ResolveEraStartUtc(split));
}
[Fact]
public void SampleCap_KeepsTheNewestSamples()
{
// The cap is a memory backstop for pathological sessions. Since the era window judges
// the NEWEST leg, the buffer must drop the oldest entries — keeping the head instead
// would blind the detector on exactly the sessions big enough to reach the cap.
var scan = new SessionSkewScan();
int adds = SessionSkewScan.MaxSamplesPerSide + 5_000;
for (int i = 0; i < adds; i++)
scan.Add(isIme: true, i, EraBase.AddSeconds(i));
Assert.True(scan.ImeSamples.Count >= SessionSkewScan.MaxSamplesPerSide);
Assert.True(scan.ImeSamples.Count < SessionSkewScan.MaxSamplesPerSide * 2);
Assert.Equal((double)(adds - 1), scan.ImeSamples[scan.ImeSamples.Count - 1].DeltaMinutes);
Assert.True(scan.ImeSamples[0].DeltaMinutes > 0); // the oldest adds were trimmed away
}
// ── Bias suppression predicate ─────────────────────────────────────────
[Fact]
@@ -22,6 +22,17 @@ namespace AutopilotMonitor.Functions.Services
/// grid multiple, so the grid-conformity test is applied per sample as well: the session
/// only counts when the bulk of the IME deltas individually sit on the grid.
///
/// Samples are additionally windowed to the session's most recent INGEST ERA. A session
/// partition is not one agent run: a pre-provisioning session's Part 1 is written weeks
/// before Part 2, by whatever agent build was current back then (field 2026-09-01, sessions
/// e797117b / c06d639d / d7c8032b: 26 IME samples at exactly −60 min from a 2026-08-20
/// technician leg under agent 2.0.1409 — pre per-line anchoring — dominated the 3…9 clean
/// samples of the same day's user leg and fired the tripwire against a build that had
/// already self-updated to 2.0.1445). Those hits are real skew but ancient history, not a
/// regression of the running build, and no scan-wide statistic can separate them because the
/// stored event rows carry no agent version. The era boundary does: ReceivedAt gaps larger
/// than <see cref="EraGapHours"/> split legs, and only the newest leg is judged.
///
/// Goal state: this never fires. Any hit is either a case the per-line anchoring does not
/// cover or a detector bug — both are actionable findings, never noise to tune away.
/// </summary>
@@ -41,6 +52,11 @@ namespace AutopilotMonitor.Functions.Services
// continuum hits that window only ~2·2/15 ≈ 27 % of the time by chance; a real skew,
// even one mixing two writer eras, hits it for essentially every line.
internal const double MinGridConformantFraction = 0.8;
// A silence longer than this ends an ingest era. A live agent uploads continuously
// (self-metrics and heartbeats every few minutes; it reports session_stalled after
// 60 min idle, and the maintenance sweep classifies silence at 2 h), so no gap this
// large occurs INSIDE one leg — while a pre-provisioning handover is hours to weeks.
internal const double EraGapHours = 2.0;
internal sealed class Result
{
@@ -54,24 +70,38 @@ namespace AutopilotMonitor.Functions.Services
public int ImeBatchCount { get; set; }
public int OtherBatchCount { get; set; }
public double GridConformantFraction { get; set; }
/// <summary>ReceivedAt of the oldest batch still inside the judged ingest era.</summary>
public DateTime EraStartUtc { get; set; }
/// <summary>IME samples dropped as belonging to an older ingest era.</summary>
public int ImeSamplesOutsideEra { get; set; }
/// <summary>Non-IME samples dropped as belonging to an older ingest era.</summary>
public int OtherSamplesOutsideEra { get; set; }
}
/// <summary>
/// Pure detector: non-null exactly when the session shows a grid-aligned divergence
/// between IME-derived and other event timestamps. All suppression inputs live in the
/// scan itself so the decision is fully unit-testable.
/// Pure detector: non-null exactly when the session's most recent ingest era shows a
/// grid-aligned divergence between IME-derived and other event timestamps. All
/// suppression inputs live in the scan itself so the decision is fully unit-testable.
/// </summary>
internal static Result? Evaluate(SessionSkewScan? scan)
{
if (scan == null)
return null;
if (scan.ImeDeltaMinutes.Count < MinSamplesPerSide || scan.OtherDeltaMinutes.Count < MinSamplesPerSide)
return null;
if (scan.ImeDistinctBatchCount < MinDistinctBatchesPerSide || scan.OtherDistinctBatchCount < MinDistinctBatchesPerSide)
var eraStart = ResolveEraStartUtc(scan);
if (eraStart == null)
return null;
var medianIme = PercentileMath.Median(scan.ImeDeltaMinutes);
var medianOther = PercentileMath.Median(scan.OtherDeltaMinutes);
var ime = SelectEra(scan.ImeSamples, eraStart.Value);
var other = SelectEra(scan.OtherSamples, eraStart.Value);
if (ime.Deltas.Count < MinSamplesPerSide || other.Deltas.Count < MinSamplesPerSide)
return null;
if (ime.BatchCount < MinDistinctBatchesPerSide || other.BatchCount < MinDistinctBatchesPerSide)
return null;
var medianIme = PercentileMath.Median(ime.Deltas);
var medianOther = PercentileMath.Median(other.Deltas);
var diff = medianIme - medianOther;
var steps = (int)Math.Round(diff / GridMinutes, MidpointRounding.AwayFromZero);
var residual = Math.Abs(diff - steps * (double)GridMinutes);
@@ -79,7 +109,7 @@ namespace AutopilotMonitor.Functions.Services
if (Math.Abs(steps) < 1 || residual >= ResidualToleranceMinutes)
return null;
var conformant = ComputeGridConformantFraction(scan.ImeDeltaMinutes, medianOther);
var conformant = ComputeGridConformantFraction(ime.Deltas, medianOther);
if (conformant < MinGridConformantFraction)
return null;
@@ -90,14 +120,68 @@ namespace AutopilotMonitor.Functions.Services
DiffMinutes = diff,
GridSteps = steps,
ResidualMinutes = residual,
ImeSampleCount = scan.ImeDeltaMinutes.Count,
OtherSampleCount = scan.OtherDeltaMinutes.Count,
ImeBatchCount = scan.ImeDistinctBatchCount,
OtherBatchCount = scan.OtherDistinctBatchCount,
ImeSampleCount = ime.Deltas.Count,
OtherSampleCount = other.Deltas.Count,
ImeBatchCount = ime.BatchCount,
OtherBatchCount = other.BatchCount,
GridConformantFraction = conformant,
EraStartUtc = eraStart.Value,
ImeSamplesOutsideEra = ime.Dropped,
OtherSamplesOutsideEra = other.Dropped,
};
}
/// <summary>
/// Start of the session's most recent ingest era: walk the distinct batch stamps of BOTH
/// sides backwards from the newest one and stop at the first gap wider than
/// <see cref="EraGapHours"/>. Null when the scan holds no samples at all.
/// </summary>
internal static DateTime? ResolveEraStartUtc(SessionSkewScan scan)
{
var stamps = new HashSet<DateTime>();
foreach (var sample in scan.ImeSamples)
stamps.Add(sample.ReceivedAtUtc);
foreach (var sample in scan.OtherSamples)
stamps.Add(sample.ReceivedAtUtc);
if (stamps.Count == 0)
return null;
var ordered = new List<DateTime>(stamps);
ordered.Sort();
var eraStart = ordered[ordered.Count - 1];
for (int i = ordered.Count - 1; i > 0; i--)
{
if ((ordered[i] - ordered[i - 1]).TotalHours > EraGapHours)
break;
eraStart = ordered[i - 1];
}
return eraStart;
}
/// <summary>
/// Splits one side's samples at the era boundary: the deltas inside the era, the number
/// of distinct upload batches backing them, and how many samples were dropped as older.
/// </summary>
private static (List<double> Deltas, int BatchCount, int Dropped) SelectEra(
IReadOnlyList<SkewSample> samples, DateTime eraStartUtc)
{
var deltas = new List<double>(samples.Count);
var batches = new HashSet<DateTime>();
int dropped = 0;
foreach (var sample in samples)
{
if (sample.ReceivedAtUtc < eraStartUtc)
{
dropped++;
continue;
}
deltas.Add(sample.DeltaMinutes);
batches.Add(sample.ReceivedAtUtc);
}
return (deltas, batches.Count, dropped);
}
/// <summary>
/// Fraction of samples whose delta minus <paramref name="reference"/> lies within
/// <see cref="ResidualToleranceMinutes"/> of any multiple of <see cref="GridMinutes"/>
@@ -672,6 +672,7 @@ namespace AutopilotMonitor.Functions.Services
$"({result.GridSteps}x15min grid, residual {result.ResidualMinutes:0.0} min, {result.GridConformantFraction:P0} of IME samples on grid; " +
$"median dIME {result.MedianImeDeltaMinutes:0.0} / dOther {result.MedianOtherDeltaMinutes:0.0} min; " +
$"{result.ImeSampleCount}/{result.OtherSampleCount} samples over {result.ImeBatchCount}/{result.OtherBatchCount} batches; " +
$"ingest era from {result.EraStartUtc:yyyy-MM-ddTHH:mm:ss}Z, {result.ImeSamplesOutsideEra}/{result.OtherSamplesOutsideEra} older samples excluded; " +
$"origins: {(origins.Length > 0 ? origins : "none")})";
await _opsEventService.RecordCmTraceTimeSkewRegressionAsync(tenantId, sessionId, agentVersion, message, new
@@ -688,6 +689,9 @@ namespace AutopilotMonitor.Functions.Services
otherSampleCount = result.OtherSampleCount,
imeBatchCount = result.ImeBatchCount,
otherBatchCount = result.OtherBatchCount,
eraStartUtc = result.EraStartUtc,
imeSamplesOutsideEra = result.ImeSamplesOutsideEra,
otherSamplesOutsideEra = result.OtherSamplesOutsideEra,
sourceOffsetOrigins = histogram,
});
}
@@ -3545,9 +3545,6 @@ namespace AutopilotMonitor.Functions.Services
int total = 0, reboots = 0;
var skew = new SessionSkewScan();
var imeBatches = new HashSet<DateTime>();
var otherBatches = new HashSet<DateTime>();
const int maxSkewSamplesPerSide = 20_000;
await foreach (var entity in query)
{
total++;
@@ -3563,14 +3560,11 @@ namespace AutopilotMonitor.Functions.Services
var occurredAt = ResolveEventTimestampOrNull(entity);
if (!receivedAt.HasValue || !occurredAt.HasValue)
continue;
// ReceivedAt travels with the sample: the detector windows on ingest eras
// and derives its distinct-batch counts from the same era-filtered set.
var isIme = string.Equals(entity.GetString("Source"), "ImeLogTracker", StringComparison.OrdinalIgnoreCase);
var side = isIme ? skew.ImeDeltaMinutes : skew.OtherDeltaMinutes;
if (side.Count < maxSkewSamplesPerSide)
side.Add((receivedAt.Value - occurredAt.Value).TotalMinutes);
(isIme ? imeBatches : otherBatches).Add(receivedAt.Value);
skew.Add(isIme, (receivedAt.Value - occurredAt.Value).TotalMinutes, receivedAt.Value);
}
skew.ImeDistinctBatchCount = imeBatches.Count;
skew.OtherDistinctBatchCount = otherBatches.Count;
return (total, reboots, skew);
}
catch (Exception ex)
@@ -1,7 +1,30 @@
using System;
using System.Collections.Generic;
namespace AutopilotMonitor.Shared.Models
{
/// <summary>
/// One timestamp-delta sample: <c>(ReceivedAt − OccurredUtc)</c> in minutes together with the
/// <c>ReceivedAt</c> it was measured against. ReceivedAt is stamped once per upload BATCH, so
/// it doubles as the batch identity (distinct-batch guard) and as the ingest-era coordinate
/// the tripwire windows on.
/// </summary>
public readonly struct SkewSample
{
/// <summary>Creates a sample from its delta and the batch's ReceivedAt (UTC).</summary>
public SkewSample(double deltaMinutes, DateTime receivedAtUtc)
{
DeltaMinutes = deltaMinutes;
ReceivedAtUtc = receivedAtUtc;
}
/// <summary>(ReceivedAt − OccurredUtc) in minutes.</summary>
public double DeltaMinutes { get; }
/// <summary>The upload batch's server-side receive time (UTC).</summary>
public DateTime ReceivedAtUtc { get; }
}
/// <summary>
/// Per-session timestamp-delta samples collected during the terminal counter reconcile's
/// single event-partition scan (zero extra storage reads). Samples are split by producing
@@ -9,23 +32,43 @@ namespace AutopilotMonitor.Shared.Models
/// Consumed by the CMTrace time-skew tripwire, which compares the two sides' median
/// (ReceivedAt − OccurredUtc) deltas — shared upload/spool latency is common-mode and
/// cancels in that difference; a timezone mis-conversion shifts only the IME side.
/// <para>
/// Every sample carries its ReceivedAt: a session partition can hold events from several
/// ingest eras days apart (pre-provisioning Part 1 → Part 2, written by different agent
/// builds), and the detector only judges the most recent one. Distinct-batch counts are
/// therefore NOT accumulated here — they are derived inside the detector from the same
/// era-filtered sample set the medians come from.
/// </para>
/// </summary>
public class SessionSkewScan
{
/// <summary>(ReceivedAt − OccurredUtc) in minutes for events with Source == "ImeLogTracker".</summary>
public List<double> ImeDeltaMinutes { get; } = new List<double>();
/// <summary>
/// Memory backstop for pathological sessions. The buffers keep the NEWEST samples per
/// side: the era window judges the most recent ingest era, so discarding the tail would
/// blind the detector on exactly the sessions large enough to hit this cap.
/// </summary>
public const int MaxSamplesPerSide = 20_000;
/// <summary>(ReceivedAt − OccurredUtc) in minutes for all other event sources.</summary>
public List<double> OtherDeltaMinutes { get; } = new List<double>();
// Trimming in chunks keeps the per-add cost amortized O(1) once the cap is reached
// (removing one element per add would be an O(cap) memmove on every further event).
private const int TrimChunk = 4096;
/// <summary>Samples for events with Source == "ImeLogTracker", in scan order.</summary>
public List<SkewSample> ImeSamples { get; } = new List<SkewSample>();
/// <summary>Samples for all other event sources, in scan order.</summary>
public List<SkewSample> OtherSamples { get; } = new List<SkewSample>();
/// <summary>
/// Distinct ReceivedAt values seen on the IME side. ReceivedAt is stamped once per
/// upload BATCH, not per event — a median over one or two batches is a
/// batch-composition artifact, not a session property.
/// Appends a sample to the IME or the other side, enforcing <see cref="MaxSamplesPerSide"/>
/// by dropping the oldest entries once the buffer grows past the cap.
/// </summary>
public int ImeDistinctBatchCount { get; set; }
/// <summary>Distinct ReceivedAt values seen on the non-IME side.</summary>
public int OtherDistinctBatchCount { get; set; }
public void Add(bool isIme, double deltaMinutes, DateTime receivedAtUtc)
{
var side = isIme ? ImeSamples : OtherSamples;
side.Add(new SkewSample(deltaMinutes, receivedAtUtc));
if (side.Count >= MaxSamplesPerSide + TrimChunk)
side.RemoveRange(0, side.Count - MaxSamplesPerSide);
}
}
}