mirror of
https://github.com/okieselbach/AutopilotMonitor.git
synced 2026-10-24 12:42:38 +00:00
feat(backend): P6 split - CMTrace skew tripwire + customer clock_skew rule
Operator side: CmTraceSkewTripwire piggybacks on the terminal counter
reconcile's partition scan (zero extra reads) and compares median
ReceivedAt-OccurredUtc deltas of IME-derived vs other events. A clean
15-min-grid divergence (residual < 2 min, >=20 samples / >=3 distinct
batches per side) emits the CmTraceTimeSkewRegression ops event
(Agent/Warning, goal: never fires). Bias-dominated sessions are
suppressed; measuredWriterOffsetMinutes is never consulted (sticky
after era flip-backs). Kill switch: CmTraceSkewTripwireDisabled.
Customer side: new clock_skew analyze condition source (skewMetric
clock_jump | sustained_offset, threshold seconds, gt/gte magnitude)
behind built-in ANALYZE-DEV-008 (300 s = the Kerberos/Entra 5-min skew
boundary). Batch-median frames with spool-spread filter, plateau step
detection (a gradual backlog-drain ramp cannot fake a jump) and
end-state persistence; IME-derived events are excluded so a CMTrace
anchoring regression can never fire the customer rule. Evidence carries
field=skewSummary/value for {{skewSummary}} interpolation.
Full mirroring chain: schema enum + skewMetric property, shared
manifest + combine.js regeneration, MCP authoring guide + semantic
lint, web SOURCES/form/card, ops-event dual-register, docs. P5 is
subsumed by the tripwire. PercentileMath is now the single nearest-rank
implementation (Apps/PlatformMetrics delegate).
Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
Oliver Kieselbach
+1
-1
Submodule .claude/commands updated: 3f98090747...2866e07882
@@ -86,10 +86,33 @@ with the process zone).
|
||||
* Mixed pairs remain possible across a restart (start resolved from backlog fallback,
|
||||
completion line-anchored) and are recognizable via provenance.
|
||||
|
||||
# Backend regression tripwire
|
||||
|
||||
The backend watches for the failure mode this design eliminates. On every terminal ingest
|
||||
batch, the counter reconcile's existing event-partition scan additionally collects
|
||||
`Δ = ReceivedAt − OccurredUtc` per event, split by `Source == "ImeLogTracker"` versus the
|
||||
rest (`CmTraceSkewTripwire`, hooked in `EventIngestProcessor`). If
|
||||
`median(Δ_IME) − median(Δ_other)` is a clean 15-minute multiple (residual < 2 min — the
|
||||
same constants as the anchoring grid guard above) with ≥ 20 samples over ≥ 3 distinct
|
||||
upload batches per side, a `CmTraceTimeSkewRegression` ops event fires (category Agent,
|
||||
warning), carrying all numbers plus the session's `sourceOffsetOrigin` histogram.
|
||||
Bias-dominated sessions are suppressed (writer-declared offsets cannot be an anchoring
|
||||
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.
|
||||
|
||||
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.
|
||||
|
||||
# Citations
|
||||
|
||||
* `CmTraceOffsetCalibrator.TryMeasureOffset` — the pure grid measurement, shared by
|
||||
anchoring and the observational per-file path.
|
||||
* `CmTraceSkewTripwire` + `CmTraceSkewTripwireTests` — the backend tripwire above.
|
||||
* `RuleEngine.ConditionEvaluators.EvaluateClockSkewCondition` + `RuleEngineClockSkewTests`
|
||||
— the customer-facing device-clock metrics (batch-median frames, spool-spread filter,
|
||||
plateau step detection, end-state persistence).
|
||||
* `ImeLogTracker.LogProcessing.cs` `ResolveEntryUtc` — the resolution order above.
|
||||
* `ImeLogTrackerLineAnchoringTests` — resolves-to-T across writer beliefs, interleaved
|
||||
eras in one chunk, freshness boundaries (first sight, empty first sight, poll gap,
|
||||
|
||||
@@ -2,6 +2,8 @@
|
||||
|
||||
## 2026-08-20
|
||||
|
||||
* **Update**: `agent/cmtrace-time-resolution.md` — new "Backend regression tripwire" section: the P6 split. Operator side: `CmTraceSkewTripwire` piggybacks on the terminal counter reconcile's partition scan (`Δ = ReceivedAt − OccurredUtc` per source, 15-min grid + 2-min residual, ≥20 samples / ≥3 distinct batches per side, bias-dominated sessions suppressed, `measuredWriterOffsetMinutes` never consulted) and emits the `CmTraceTimeSkewRegression` ops event — goal state: never fires; kill switch `CmTraceSkewTripwireDisabled`. Customer side: new `clock_skew` analyze condition source (`skewMetric: clock_jump | sustained_offset`, threshold in seconds, batch-median frames with spool-spread filter, plateau step detection, end-state persistence, IME-derived events excluded) behind the new built-in `ANALYZE-DEV-008` (300 s = the Kerberos/Entra 5-minute skew boundary) — the during-enrollment complement of ANALYZE-DEV-007's one-shot NTP check. P5 (agent-side observability event) is subsumed by the tripwire.
|
||||
|
||||
* **New**: `agent/cmtrace-time-resolution.md` — the era-aware answer to the CMTrace timezone skew: a bias-less line read in a provably fresh pass anchors its own UTC offset against the agent clock (15-minute grid, 2-minute residual), so interleaved writer eras in one file resolve line by line; everything not provably fresh keeps the uniform reader-zone fallback and says so via `sourceOffsetOrigin`. Freshness is a per-file `LastCheckedUtc` stamped on every poll look (empty first sights included) with a 30-second window, never restored across restarts. Replaces the reverted per-file offset application (04b1a7c6); the per-file measurement stays observational, now with the anchor timestamp logged as a tripwire for the still-unexplained e9753578 label/anchor crossing.
|
||||
|
||||
* **Update**: `agent/post-reboot-completion-recovery.md` — retitled to "ESP Completion Starvation — the sits-d Cloud-PC complex" and reframed on the evidence: the reboot was a red herring. The five stuck Cloud PCs carried `SkipUserStatusPage=True` (read 10+ times across three agent processes, `True` every time) and Windows honoured it — no user ESP page, hence no Shell-Core 62407 and no AccountSetup provisioning registry, verified over a six-hour uninterrupted post-reboot run with the watcher armed. Arm B of `ShouldTransitionToAwaitingHello` was open from the bootstrap onward, but every evaluation site hangs off a signal a skip-user flow cannot produce, and the `AdvisoryCompletion` backstop arms exclusively from an `EspExiting` signal — an open gate nobody knocks on. Fix: `HandleImeUserSessionCompletedV1` accepts the observed skip in place of the final-exit evidence (`skipUserExitEquivalent`; anchor + genuine IME completion + real-user desktop stay mandatory, WDP excluded). The knock deliberately lives on the IME edge, not desktop arrival — ordering-robust against Fix 10's AccountSetup bounce-back and semantically "user-phase work finished" rather than mid-install. `ClassicSkipUserEspUserSessionCompletionTests` replays the real 8110e262 ordering end to end. Docs and comments no longer claim an ESP-exit recovery that does not exist (stray duplicate summary in `ShellCoreTracker`, `ImeLogHost` restore-callback comment); the residual `SkipUser=false` exit-in-downtime gap is named honestly and surfaced by `ANALYZE-ESP-005` via `session_stalled` instead of being patched with an unorderable replay.
|
||||
|
||||
@@ -0,0 +1,69 @@
|
||||
{
|
||||
"$schema": "../schema/analyze-rule.schema.json",
|
||||
"ruleId": "ANALYZE-DEV-008",
|
||||
"title": "Device Clock Wrong During Enrollment — Jump or Sustained Offset",
|
||||
"description": "The device's clock changed by five or more minutes mid-session (clock jump) or ran five or more minutes off server time for the whole session (sustained offset), measured against the server receive frame of every uploaded event. Complements ANALYZE-DEV-007, which is a one-shot NTP check at agent startup and misses devices where NTP was unreachable or the clock changed later. IME-log-derived events are excluded from the measurement.",
|
||||
"severity": "warning",
|
||||
"category": "device",
|
||||
"version": "1.0.0",
|
||||
"author": "Autopilot Monitor",
|
||||
"enabled": true,
|
||||
"evaluateOn": [
|
||||
"enrollment_end",
|
||||
"whiteglove_sealed"
|
||||
],
|
||||
"trigger": "single",
|
||||
"baseConfidence": 60,
|
||||
"confidenceThreshold": 50,
|
||||
"conditions": [
|
||||
{
|
||||
"signal": "clock_jump",
|
||||
"source": "clock_skew",
|
||||
"skewMetric": "clock_jump",
|
||||
"operator": "gte",
|
||||
"value": "300",
|
||||
"required": false
|
||||
},
|
||||
{
|
||||
"signal": "sustained_offset",
|
||||
"source": "clock_skew",
|
||||
"skewMetric": "sustained_offset",
|
||||
"operator": "gte",
|
||||
"value": "300",
|
||||
"required": false
|
||||
}
|
||||
],
|
||||
"confidenceFactors": [],
|
||||
"explanation": "This device's {{skewSummary}} (five-minute threshold). A clock that is off by five or more minutes breaks time-sensitive enrollment mechanisms: Kerberos tolerates at most 5 minutes of skew (Hybrid Join), Entra ID token validation allows about 5 minutes on token lifetimes, and freshly issued SCEP/PKCS certificates can be rejected as \"not yet valid\" when the clock is more than ~10 minutes behind the CA.\n\nUnlike ANALYZE-DEV-007 (a one-shot NTP comparison at agent startup), this finding is measured across the whole session against the server's receive time — it also catches devices where the NTP check could not run (UDP 123 blocked) or where the clock changed after startup (time sync correcting a wrong RTC, Windows auto-timezone, manual change).",
|
||||
"remediation": [
|
||||
{
|
||||
"title": "Fix the device clock and time source",
|
||||
"steps": [
|
||||
"Check the BIOS/RTC clock and the CMOS battery if the wrong time recurs on the same hardware.",
|
||||
"At OOBE, run 'w32tm /resync /force' (Shift+F10 command prompt) to sync with time.windows.com.",
|
||||
"Ensure UDP port 123 to time.windows.com is open in the deployment network.",
|
||||
"Review 'Set time automatically' and 'Set time zone automatically' behavior on the deployment image — automatic timezone changes mid-enrollment show up as clock jumps."
|
||||
]
|
||||
}
|
||||
],
|
||||
"relatedDocs": [
|
||||
{
|
||||
"title": "Microsoft Learn: Windows Autopilot known issues — device provisioning can fail (clock skew)",
|
||||
"url": "https://learn.microsoft.com/en-us/autopilot/known-issues"
|
||||
},
|
||||
{
|
||||
"title": "Microsoft Learn: Windows Autopilot requirements — Networking (NTP)",
|
||||
"url": "https://learn.microsoft.com/en-us/autopilot/requirements?tabs=networking"
|
||||
}
|
||||
],
|
||||
"markSessionAsFailedDefault": false,
|
||||
"tags": [
|
||||
"device",
|
||||
"clock-skew",
|
||||
"clock-jump",
|
||||
"time-sync",
|
||||
"kerberos",
|
||||
"certificates",
|
||||
"known-issue"
|
||||
]
|
||||
}
|
||||
+68
@@ -1516,6 +1516,74 @@
|
||||
"known-issue"
|
||||
]
|
||||
},
|
||||
{
|
||||
"ruleId": "ANALYZE-DEV-008",
|
||||
"title": "Device Clock Wrong During Enrollment — Jump or Sustained Offset",
|
||||
"description": "The device's clock changed by five or more minutes mid-session (clock jump) or ran five or more minutes off server time for the whole session (sustained offset), measured against the server receive frame of every uploaded event. Complements ANALYZE-DEV-007, which is a one-shot NTP check at agent startup and misses devices where NTP was unreachable or the clock changed later. IME-log-derived events are excluded from the measurement.",
|
||||
"severity": "warning",
|
||||
"category": "device",
|
||||
"version": "1.0.0",
|
||||
"author": "Autopilot Monitor",
|
||||
"enabled": true,
|
||||
"evaluateOn": [
|
||||
"enrollment_end",
|
||||
"whiteglove_sealed"
|
||||
],
|
||||
"trigger": "single",
|
||||
"baseConfidence": 60,
|
||||
"confidenceThreshold": 50,
|
||||
"conditions": [
|
||||
{
|
||||
"signal": "clock_jump",
|
||||
"source": "clock_skew",
|
||||
"skewMetric": "clock_jump",
|
||||
"operator": "gte",
|
||||
"value": "300",
|
||||
"required": false
|
||||
},
|
||||
{
|
||||
"signal": "sustained_offset",
|
||||
"source": "clock_skew",
|
||||
"skewMetric": "sustained_offset",
|
||||
"operator": "gte",
|
||||
"value": "300",
|
||||
"required": false
|
||||
}
|
||||
],
|
||||
"confidenceFactors": [],
|
||||
"explanation": "This device's {{skewSummary}} (five-minute threshold). A clock that is off by five or more minutes breaks time-sensitive enrollment mechanisms: Kerberos tolerates at most 5 minutes of skew (Hybrid Join), Entra ID token validation allows about 5 minutes on token lifetimes, and freshly issued SCEP/PKCS certificates can be rejected as \"not yet valid\" when the clock is more than ~10 minutes behind the CA.\n\nUnlike ANALYZE-DEV-007 (a one-shot NTP comparison at agent startup), this finding is measured across the whole session against the server's receive time — it also catches devices where the NTP check could not run (UDP 123 blocked) or where the clock changed after startup (time sync correcting a wrong RTC, Windows auto-timezone, manual change).",
|
||||
"remediation": [
|
||||
{
|
||||
"title": "Fix the device clock and time source",
|
||||
"steps": [
|
||||
"Check the BIOS/RTC clock and the CMOS battery if the wrong time recurs on the same hardware.",
|
||||
"At OOBE, run 'w32tm /resync /force' (Shift+F10 command prompt) to sync with time.windows.com.",
|
||||
"Ensure UDP port 123 to time.windows.com is open in the deployment network.",
|
||||
"Review 'Set time automatically' and 'Set time zone automatically' behavior on the deployment image — automatic timezone changes mid-enrollment show up as clock jumps."
|
||||
]
|
||||
}
|
||||
],
|
||||
"relatedDocs": [
|
||||
{
|
||||
"title": "Microsoft Learn: Windows Autopilot known issues — device provisioning can fail (clock skew)",
|
||||
"url": "https://learn.microsoft.com/en-us/autopilot/known-issues"
|
||||
},
|
||||
{
|
||||
"title": "Microsoft Learn: Windows Autopilot requirements — Networking (NTP)",
|
||||
"url": "https://learn.microsoft.com/en-us/autopilot/requirements?tabs=networking"
|
||||
}
|
||||
],
|
||||
"markSessionAsFailedDefault": false,
|
||||
"tags": [
|
||||
"device",
|
||||
"clock-skew",
|
||||
"clock-jump",
|
||||
"time-sync",
|
||||
"kerberos",
|
||||
"certificates",
|
||||
"known-issue"
|
||||
]
|
||||
},
|
||||
{
|
||||
"ruleId": "ANALYZE-ENRL-001",
|
||||
"title": "Enrollment Failed",
|
||||
|
||||
@@ -169,7 +169,12 @@
|
||||
"source": {
|
||||
"type": "string",
|
||||
"description": "Source of the signal.",
|
||||
"enum": ["event_type", "event_data", "event_data_array", "phase_duration", "event_count", "app_install_duration", "event_correlation"]
|
||||
"enum": ["event_type", "event_data", "event_data_array", "phase_duration", "event_count", "app_install_duration", "event_correlation", "clock_skew"]
|
||||
},
|
||||
"skewMetric": {
|
||||
"type": "string",
|
||||
"description": "For clock_skew only: which device-clock metric to evaluate. clock_jump = persistent step in the device's clock frame mid-session; sustained_offset = whole session ran on a clock off by at least the threshold. value is the threshold in seconds; operator is limited to gt/gte on the magnitude. IME-log-derived events are excluded from the measurement.",
|
||||
"enum": ["clock_jump", "sustained_offset"]
|
||||
},
|
||||
"eventType": {
|
||||
"type": "string",
|
||||
|
||||
@@ -0,0 +1,152 @@
|
||||
using AutopilotMonitor.Functions.Services;
|
||||
using AutopilotMonitor.Shared.Models;
|
||||
|
||||
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
|
||||
/// < 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.
|
||||
/// </summary>
|
||||
public class CmTraceSkewTripwireTests
|
||||
{
|
||||
/// <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);
|
||||
return scan;
|
||||
}
|
||||
|
||||
// ── Must fire ──────────────────────────────────────────────────────────
|
||||
|
||||
[Fact]
|
||||
public void Fires_OnLargeNegativeGridDivergence()
|
||||
{
|
||||
// −540 min = the field-measured E.-Australia-style skew (−9 h), exactly 36 grid steps.
|
||||
// Both sides share 1.2 min upload latency — it cancels in the difference.
|
||||
var result = CmTraceSkewTripwire.Evaluate(MakeScan(imeDeltaMinutes: -538.8, otherDeltaMinutes: 1.2));
|
||||
|
||||
Assert.NotNull(result);
|
||||
Assert.Equal(-36, result!.GridSteps);
|
||||
Assert.Equal(-540.0, result.DiffMinutes, 3);
|
||||
Assert.True(result.ResidualMinutes < CmTraceSkewTripwire.ResidualToleranceMinutes);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void Fires_OnSingleGridStepWithBoundaryResidual()
|
||||
{
|
||||
// +15 min divergence with residual 1.9 — just inside the 2-min tolerance.
|
||||
var result = CmTraceSkewTripwire.Evaluate(MakeScan(imeDeltaMinutes: 16.9, otherDeltaMinutes: 0.0));
|
||||
|
||||
Assert.NotNull(result);
|
||||
Assert.Equal(1, result!.GridSteps);
|
||||
Assert.Equal(1.9, result.ResidualMinutes, 3);
|
||||
}
|
||||
|
||||
// ── Must stay silent ───────────────────────────────────────────────────
|
||||
|
||||
[Fact]
|
||||
public void Silent_WhenResidualOffGrid()
|
||||
{
|
||||
// 17.5 min divergence: nearest grid step is 15, residual 2.5 ≥ tolerance — the replay
|
||||
// signature (old log lines re-read late land anywhere, not on the zone grid).
|
||||
Assert.Null(CmTraceSkewTripwire.Evaluate(MakeScan(imeDeltaMinutes: 17.5, otherDeltaMinutes: 0.0)));
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void Silent_WhenDivergenceBelowOneGridStep()
|
||||
{
|
||||
// 9 min rounds to 1 step but with residual 6 — and 7 min rounds to 0 steps. Both silent.
|
||||
Assert.Null(CmTraceSkewTripwire.Evaluate(MakeScan(imeDeltaMinutes: 9.0, otherDeltaMinutes: 0.0)));
|
||||
Assert.Null(CmTraceSkewTripwire.Evaluate(MakeScan(imeDeltaMinutes: 7.0, otherDeltaMinutes: 0.0)));
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void Silent_WhenImeSampleCountBelowMinimum()
|
||||
{
|
||||
var scan = MakeScan(imeDeltaMinutes: -540.0, otherDeltaMinutes: 0.0,
|
||||
imeSamples: CmTraceSkewTripwire.MinSamplesPerSide - 1);
|
||||
Assert.Null(CmTraceSkewTripwire.Evaluate(scan));
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void Silent_WhenDistinctBatchesBelowMinimum()
|
||||
{
|
||||
// A perfect grid divergence backed by only 2 upload batches on one side is a
|
||||
// batch-composition artifact (ReceivedAt is stamped per batch, not per event).
|
||||
var scan = MakeScan(imeDeltaMinutes: -540.0, otherDeltaMinutes: 0.0,
|
||||
otherBatches: CmTraceSkewTripwire.MinDistinctBatchesPerSide - 1);
|
||||
Assert.Null(CmTraceSkewTripwire.Evaluate(scan));
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void Silent_OnCommonModeSpoolLatency()
|
||||
{
|
||||
// Offline spool: BOTH sides uploaded 45 min late. The difference of medians is ~0 —
|
||||
// shared latency must cancel, never read as skew.
|
||||
Assert.Null(CmTraceSkewTripwire.Evaluate(MakeScan(imeDeltaMinutes: 45.3, otherDeltaMinutes: 45.1)));
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void Silent_OnNullOrEmptyScan()
|
||||
{
|
||||
Assert.Null(CmTraceSkewTripwire.Evaluate(null));
|
||||
Assert.Null(CmTraceSkewTripwire.Evaluate(new SessionSkewScan()));
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void MedianIsRobustAgainstMinorityOutliers()
|
||||
{
|
||||
// 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);
|
||||
Assert.Null(CmTraceSkewTripwire.Evaluate(scan));
|
||||
}
|
||||
|
||||
// ── Bias suppression predicate ─────────────────────────────────────────
|
||||
|
||||
[Fact]
|
||||
public void BiasDominated_WhenBiasIsHalfOrMore()
|
||||
{
|
||||
Assert.True(CmTraceSkewTripwire.IsBiasDominated(new Dictionary<string, int>
|
||||
{
|
||||
["bias"] = 50,
|
||||
["line-anchored"] = 50,
|
||||
}));
|
||||
Assert.True(CmTraceSkewTripwire.IsBiasDominated(new Dictionary<string, int>
|
||||
{
|
||||
["Bias"] = 3, // case-insensitive
|
||||
}));
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void NotBiasDominated_WhenAnchoredMajorityOrNoData()
|
||||
{
|
||||
Assert.False(CmTraceSkewTripwire.IsBiasDominated(new Dictionary<string, int>
|
||||
{
|
||||
["bias"] = 10,
|
||||
["line-anchored"] = 80,
|
||||
["reader-zone-fallback"] = 11,
|
||||
}));
|
||||
// Empty/missing histogram fails open toward reporting — a storage error must not
|
||||
// silently swallow a genuine regression hit.
|
||||
Assert.False(CmTraceSkewTripwire.IsBiasDominated(new Dictionary<string, int>()));
|
||||
Assert.False(CmTraceSkewTripwire.IsBiasDominated(null));
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,319 @@
|
||||
using AutopilotMonitor.Functions.Services;
|
||||
using AutopilotMonitor.Shared.DataAccess;
|
||||
using AutopilotMonitor.Shared.Models;
|
||||
using Microsoft.Extensions.Logging.Abstractions;
|
||||
using Moq;
|
||||
|
||||
namespace AutopilotMonitor.Functions.Tests;
|
||||
|
||||
/// <summary>
|
||||
/// Pins the clock_skew condition source (EvaluateClockSkewCondition) behind ANALYZE-DEV-008.
|
||||
/// The evaluator measures d = Timestamp − ReceivedAt per event, reduces to per-upload-batch
|
||||
/// medians (wide-spread batches = spool backlogs are dropped), then detects either a persistent
|
||||
/// mid-session step between two flat plateaus (clock_jump) or a stable whole-session offset
|
||||
/// (sustained_offset). IME-log-derived events are excluded so a CMTrace anchoring regression
|
||||
/// (the e9753578 field incident) can never fire this customer-facing rule — that failure mode
|
||||
/// belongs to the operator-side CmTraceSkewTripwire.
|
||||
/// </summary>
|
||||
public class RuleEngineClockSkewTests
|
||||
{
|
||||
private const string TenantId = "a1b2c3d4-e5f6-7890-abcd-ef1234567890";
|
||||
private const string SessionId = "b2c3d4e5-f6a7-8901-bcde-f12345678901";
|
||||
|
||||
private static readonly DateTime T0 = new(2026, 8, 20, 10, 0, 0, DateTimeKind.Utc);
|
||||
|
||||
// ── clock_jump: must fire ──────────────────────────────────────────────
|
||||
|
||||
[Fact]
|
||||
public async Task ClockJump_ForwardStepBetweenStableFrames_Fires()
|
||||
{
|
||||
// 4 batches on frame 0, then 4 batches on frame +1800 s (+30 min) that hold to
|
||||
// session end — the canonical "clock corrected forward mid-session".
|
||||
var events = Batches(offsetsSeconds: new double[] { 0, 0, 0, 0, 1800, 1800, 1800, 1800 });
|
||||
|
||||
var outcome = await RunAsync(MakeRule("clock_jump", "300"), events);
|
||||
|
||||
var result = Assert.Single(outcome.Results);
|
||||
var evidence = Assert.IsAssignableFrom<IDictionary<string, object>>(result.MatchedConditions["clock_jump"]);
|
||||
Assert.Equal("forward", evidence["direction"]);
|
||||
Assert.Equal(1800, Convert.ToDouble(evidence["jumpSeconds"]), 0);
|
||||
Assert.Equal("skewSummary", evidence["field"]);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task ClockJump_BackwardStep_Fires()
|
||||
{
|
||||
var events = Batches(new double[] { 0, 0, 0, -1200, -1200, -1200 });
|
||||
|
||||
var outcome = await RunAsync(MakeRule("clock_jump", "300"), events);
|
||||
|
||||
var result = Assert.Single(outcome.Results);
|
||||
var evidence = Assert.IsAssignableFrom<IDictionary<string, object>>(result.MatchedConditions["clock_jump"]);
|
||||
Assert.Equal("backward", evidence["direction"]);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task ClockJump_WrongClockCorrectedMidSession_Fires()
|
||||
{
|
||||
// Device started 30 min behind, time sync fixed it: the CORRECTED frame is the
|
||||
// persistent change, so this must fire (direction forward).
|
||||
var events = Batches(new double[] { -1800, -1800, -1800, 0, 0, 0 });
|
||||
|
||||
var outcome = await RunAsync(MakeRule("clock_jump", "300"), events);
|
||||
|
||||
var result = Assert.Single(outcome.Results);
|
||||
var evidence = Assert.IsAssignableFrom<IDictionary<string, object>>(result.MatchedConditions["clock_jump"]);
|
||||
Assert.Equal("forward", evidence["direction"]);
|
||||
}
|
||||
|
||||
// ── clock_jump: must stay silent ───────────────────────────────────────
|
||||
|
||||
[Fact]
|
||||
public async Task ClockJump_ImeDerivedSkew_ExcludedFromMeasurement_Silent()
|
||||
{
|
||||
// e9753578 replica: IME-derived events carry grid-multiple wrong timestamps (−7 h /
|
||||
// +2 h writer eras) while the agent's own events are healthy. The IME events must be
|
||||
// excluded up front — a CMTrace anchoring regression may NEVER fire the customer rule.
|
||||
var events = Batches(new double[] { 0, 0, 0, 0, 0, 0 });
|
||||
var seq = 1000;
|
||||
foreach (var imeOffset in new double[] { -25200, -25200, 7200, 7200, -25200, 7200 })
|
||||
events.Add(MakeEvent(seq++, T0.AddMinutes(seq % 7), imeOffset, source: "ImeLogTracker"));
|
||||
|
||||
var outcome = await RunAsync(MakeRule("clock_jump", "300"), events);
|
||||
|
||||
Assert.Empty(outcome.Results);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task ClockJump_WideSpreadSpoolBatch_DroppedAndSilent()
|
||||
{
|
||||
// One upload batch whose events span 30 min of emission time (offline spool flushed
|
||||
// in a single request) — the internal spread disqualifies the batch entirely.
|
||||
var events = Batches(new double[] { 0, 0, 0, 0, 0, 0 });
|
||||
var flushReceivedAt = T0.AddMinutes(90);
|
||||
for (int i = 0; i < 20; i++)
|
||||
events.Add(MakeEvent(2000 + i, flushReceivedAt, offsetSeconds: -1800 + i * 90));
|
||||
|
||||
var outcome = await RunAsync(MakeRule("clock_jump", "300"), events);
|
||||
|
||||
Assert.Empty(outcome.Results);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task ClockJump_GradualBacklogDrainRamp_Silent()
|
||||
{
|
||||
// A chunked spool drain produces a RAMP of batch medians walking back to the live
|
||||
// frame. No flat pre/post plateau pair exists, so the step detector must reject every
|
||||
// candidate — this is a delivery artifact, not a clock change.
|
||||
var events = Batches(new double[] { -1800, -1500, -1200, -900, -600, -300, 0, 0 });
|
||||
|
||||
var outcome = await RunAsync(MakeRule("clock_jump", "300"), events);
|
||||
|
||||
Assert.Empty(outcome.Results);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task ClockJump_TransientExcursionReturnsToFrame_Silent()
|
||||
{
|
||||
// Frame dips for two batches and returns — whatever it was, it is not a SUSTAINED
|
||||
// clock change (end-state persistence fails).
|
||||
var events = Batches(new double[] { 0, 0, 0, -1800, -1800, 0, 0, 0 });
|
||||
|
||||
var outcome = await RunAsync(MakeRule("clock_jump", "300"), events);
|
||||
|
||||
Assert.Empty(outcome.Results);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task ClockJump_BelowThreshold_Silent()
|
||||
{
|
||||
var events = Batches(new double[] { 0, 0, 0, 240, 240, 240 });
|
||||
|
||||
var outcome = await RunAsync(MakeRule("clock_jump", "300"), events);
|
||||
|
||||
Assert.Empty(outcome.Results);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task ClockJump_TooFewBatches_Silent()
|
||||
{
|
||||
// 5 batches < 2×ConfirmBatches — not enough for two comparison windows.
|
||||
var events = Batches(new double[] { 0, 0, 1800, 1800, 1800 });
|
||||
|
||||
var outcome = await RunAsync(MakeRule("clock_jump", "300"), events);
|
||||
|
||||
Assert.Empty(outcome.Results);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task ClockJump_LegacyEventsWithoutReceivedAt_Silent()
|
||||
{
|
||||
var events = Batches(new double[] { 0, 0, 0, 1800, 1800, 1800 });
|
||||
foreach (var e in events)
|
||||
e.ReceivedAt = null;
|
||||
|
||||
var outcome = await RunAsync(MakeRule("clock_jump", "300"), events);
|
||||
|
||||
Assert.Empty(outcome.Results);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task ClockJump_ClampedEventsExcluded_Silent()
|
||||
{
|
||||
// The jumped frame exists only on clamped events (their OccurredUtc is already
|
||||
// server-substituted in production) — excluding them leaves a single stable frame.
|
||||
var events = Batches(new double[] { 0, 0, 0, 0, 0, 0 });
|
||||
var seq = 3000;
|
||||
foreach (var offset in new double[] { 1800, 1800, 1800 })
|
||||
{
|
||||
var evt = MakeEvent(seq, T0.AddMinutes(120 + seq % 5), offset);
|
||||
evt.TimestampClamped = true;
|
||||
events.Add(evt);
|
||||
seq++;
|
||||
}
|
||||
|
||||
var outcome = await RunAsync(MakeRule("clock_jump", "300"), events);
|
||||
|
||||
Assert.Empty(outcome.Results);
|
||||
}
|
||||
|
||||
// ── sustained_offset ───────────────────────────────────────────────────
|
||||
|
||||
[Fact]
|
||||
public async Task SustainedOffset_ClockBehindWholeSession_Fires()
|
||||
{
|
||||
var events = Batches(new double[] { -1800, -1795, -1805, -1800, -1798 });
|
||||
|
||||
var outcome = await RunAsync(MakeRule("sustained_offset", "300"), events);
|
||||
|
||||
var result = Assert.Single(outcome.Results);
|
||||
var evidence = Assert.IsAssignableFrom<IDictionary<string, object>>(result.MatchedConditions["sustained_offset"]);
|
||||
Assert.Equal("behind", evidence["direction"]);
|
||||
Assert.Equal(-30.0, Convert.ToDouble(evidence["offsetMinutes"]), 1);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task SustainedOffset_HealthyUploadLatencyBaseline_Silent()
|
||||
{
|
||||
// Healthy devices sit a few seconds NEGATIVE (upload latency) — far below threshold.
|
||||
var events = Batches(new double[] { -3, -5, -2, -4, -6 });
|
||||
|
||||
var outcome = await RunAsync(MakeRule("sustained_offset", "300"), events);
|
||||
|
||||
Assert.Empty(outcome.Results);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task SustainedOffset_UnstableBatchOffsets_Silent()
|
||||
{
|
||||
// Medians all over the place: no honest single-offset statement exists, even though
|
||||
// the overall median magnitude clears the threshold.
|
||||
var events = Batches(new double[] { -2400, 900, -1800, 300, -3600, -600 });
|
||||
|
||||
var outcome = await RunAsync(MakeRule("sustained_offset", "300"), events);
|
||||
|
||||
Assert.Empty(outcome.Results);
|
||||
}
|
||||
|
||||
// ── ANALYZE-DEV-008 shape: two optional conditions ─────────────────────
|
||||
|
||||
[Fact]
|
||||
public async Task Dev008Shape_JumpOnlySession_FiresWithSingleMatchedCondition()
|
||||
{
|
||||
// Mirrors the shipped rule: both conditions required:false, baseConfidence 60,
|
||||
// threshold 50 — one matching metric must be enough to fire.
|
||||
var rule = MakeRule("clock_jump", "300");
|
||||
rule.Conditions[0].Required = false;
|
||||
rule.Conditions.Add(new RuleCondition
|
||||
{
|
||||
Signal = "sustained_offset",
|
||||
Source = "clock_skew",
|
||||
SkewMetric = "sustained_offset",
|
||||
Operator = "gte",
|
||||
Value = "300",
|
||||
Required = false,
|
||||
});
|
||||
|
||||
var events = Batches(new double[] { 0, 0, 0, 0, 1800, 1800, 1800, 1800 });
|
||||
|
||||
var outcome = await RunAsync(rule, events);
|
||||
|
||||
var result = Assert.Single(outcome.Results);
|
||||
Assert.True(result.MatchedConditions.ContainsKey("clock_jump"));
|
||||
}
|
||||
|
||||
// ===== Helpers =====
|
||||
|
||||
private static AnalyzeRule MakeRule(string skewMetric, string thresholdSeconds) => new()
|
||||
{
|
||||
RuleId = "ANALYZE-TST-CLK",
|
||||
Title = "clock_skew test rule",
|
||||
Severity = "warning",
|
||||
Category = "device",
|
||||
Enabled = true,
|
||||
IsBuiltIn = false,
|
||||
BaseConfidence = 60,
|
||||
ConfidenceThreshold = 50,
|
||||
Conditions = new List<RuleCondition>
|
||||
{
|
||||
new()
|
||||
{
|
||||
Signal = skewMetric,
|
||||
Source = "clock_skew",
|
||||
SkewMetric = skewMetric,
|
||||
Operator = "gte",
|
||||
Value = thresholdSeconds,
|
||||
Required = true,
|
||||
}
|
||||
},
|
||||
Explanation = "test {{skewSummary}}",
|
||||
};
|
||||
|
||||
/// <summary>
|
||||
/// One upload batch per entry: 5 events sharing the batch's ReceivedAt, each with
|
||||
/// Timestamp = ReceivedAt + offsetSeconds (±2 s in-batch jitter, well under the spread cap).
|
||||
/// </summary>
|
||||
private static List<EnrollmentEvent> Batches(double[] offsetsSeconds)
|
||||
{
|
||||
var events = new List<EnrollmentEvent>();
|
||||
long seq = 1;
|
||||
for (int b = 0; b < offsetsSeconds.Length; b++)
|
||||
{
|
||||
var receivedAt = T0.AddMinutes(b * 5);
|
||||
for (int i = 0; i < 5; i++)
|
||||
events.Add(MakeEvent(seq++, receivedAt, offsetsSeconds[b] + (i - 2)));
|
||||
}
|
||||
return events;
|
||||
}
|
||||
|
||||
private static EnrollmentEvent MakeEvent(long sequence, DateTime receivedAt, double offsetSeconds,
|
||||
string source = "DecisionEngine") => new()
|
||||
{
|
||||
EventId = Guid.NewGuid().ToString(),
|
||||
TenantId = TenantId,
|
||||
SessionId = SessionId,
|
||||
EventType = "status_update",
|
||||
Source = source,
|
||||
Timestamp = receivedAt.AddSeconds(offsetSeconds),
|
||||
ReceivedAt = receivedAt,
|
||||
Sequence = sequence,
|
||||
Data = new Dictionary<string, object>(),
|
||||
};
|
||||
|
||||
private static async Task<AnalysisOutcome> RunAsync(AnalyzeRule rule, List<EnrollmentEvent> events)
|
||||
{
|
||||
var ruleRepo = new Mock<IRuleRepository>();
|
||||
ruleRepo.Setup(r => r.GetAnalyzeRulesAsync("global")).ReturnsAsync(new List<AnalyzeRule> { rule });
|
||||
ruleRepo.Setup(r => r.GetAnalyzeRulesAsync(TenantId)).ReturnsAsync(new List<AnalyzeRule>());
|
||||
ruleRepo.Setup(r => r.GetRuleStatesAsync(It.IsAny<string>())).ReturnsAsync(new Dictionary<string, RuleState>());
|
||||
ruleRepo.Setup(r => r.GetRuleResultsAsync(It.IsAny<string>(), It.IsAny<string>())).ReturnsAsync(new List<RuleResult>());
|
||||
|
||||
var sessionRepo = new Mock<ISessionRepository>();
|
||||
sessionRepo.Setup(s => s.GetSessionEventsStrictAsync(TenantId, SessionId, It.IsAny<int>())).ReturnsAsync(events);
|
||||
|
||||
var ruleService = new AnalyzeRuleService(ruleRepo.Object, NullLogger<AnalyzeRuleService>.Instance);
|
||||
var engine = new RuleEngine(ruleService, ruleRepo.Object, sessionRepo.Object, NullLogger<RuleEngineClockSkewTests>.Instance);
|
||||
|
||||
return await engine.AnalyzeSessionAsync(TenantId, SessionId);
|
||||
}
|
||||
}
|
||||
@@ -270,6 +270,59 @@ public class RuleEngineDryRunTests
|
||||
Assert.Contains(DryRunAnalyzeRuleFunction.ValidateDraftRule(rule), e => e.Contains("joinField"));
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void Validate_ClockSkewMissingOrUnknownMetric_Error()
|
||||
{
|
||||
var rule = MakeRule();
|
||||
rule.Conditions[0].Source = "clock_skew";
|
||||
rule.Conditions[0].SkewMetric = null;
|
||||
rule.Conditions[0].Operator = "gte";
|
||||
rule.Conditions[0].Value = "300";
|
||||
Assert.Contains(DryRunAnalyzeRuleFunction.ValidateDraftRule(rule), e => e.Contains("skewMetric"));
|
||||
|
||||
rule.Conditions[0].SkewMetric = "drift"; // not a known metric
|
||||
Assert.Contains(DryRunAnalyzeRuleFunction.ValidateDraftRule(rule), e => e.Contains("skewMetric"));
|
||||
}
|
||||
|
||||
[Theory]
|
||||
[InlineData("lt")]
|
||||
[InlineData("equals")]
|
||||
public void Validate_ClockSkewUnsupportedOperator_Error(string op)
|
||||
{
|
||||
var rule = MakeRule();
|
||||
rule.Conditions[0].Source = "clock_skew";
|
||||
rule.Conditions[0].SkewMetric = "clock_jump";
|
||||
rule.Conditions[0].Operator = op;
|
||||
rule.Conditions[0].Value = "300";
|
||||
Assert.Contains(DryRunAnalyzeRuleFunction.ValidateDraftRule(rule), e => e.Contains("only gt/gte"));
|
||||
}
|
||||
|
||||
[Theory]
|
||||
[InlineData("")]
|
||||
[InlineData("abc")]
|
||||
[InlineData("-300")]
|
||||
[InlineData("0")]
|
||||
public void Validate_ClockSkewNonPositiveThreshold_Error(string value)
|
||||
{
|
||||
var rule = MakeRule();
|
||||
rule.Conditions[0].Source = "clock_skew";
|
||||
rule.Conditions[0].SkewMetric = "sustained_offset";
|
||||
rule.Conditions[0].Operator = "gte";
|
||||
rule.Conditions[0].Value = value;
|
||||
Assert.Contains(DryRunAnalyzeRuleFunction.ValidateDraftRule(rule), e => e.Contains("positive numeric value"));
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void Validate_ClockSkewValidCondition_NoErrors()
|
||||
{
|
||||
var rule = MakeRule();
|
||||
rule.Conditions[0].Source = "clock_skew";
|
||||
rule.Conditions[0].SkewMetric = "clock_jump";
|
||||
rule.Conditions[0].Operator = "gte";
|
||||
rule.Conditions[0].Value = "300";
|
||||
Assert.Empty(DryRunAnalyzeRuleFunction.ValidateDraftRule(rule));
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void Validate_InvalidRegex_Error()
|
||||
{
|
||||
|
||||
+4
-1
@@ -131,9 +131,12 @@ namespace AutopilotMonitor.Functions.DataAccess.TableStorage
|
||||
earliestEventTimestamp, latestEventTimestamp, currentPhase,
|
||||
platformScriptIncrement, remediationScriptIncrement, rebootIncrement);
|
||||
|
||||
public Task ReconcileSessionCountersAsync(string tenantId, string sessionId)
|
||||
public Task<SessionSkewScan?> ReconcileSessionCountersAsync(string tenantId, string sessionId)
|
||||
=> _storage.ReconcileSessionCountersAsync(tenantId, sessionId);
|
||||
|
||||
public Task<Dictionary<string, int>> GetImeOffsetOriginHistogramAsync(string tenantId, string sessionId, int maxRows = 500)
|
||||
=> _storage.GetImeOffsetOriginHistogramAsync(tenantId, sessionId, maxRows);
|
||||
|
||||
public Task UpdateSessionDiagnosticsBlobAsync(
|
||||
string tenantId, string sessionId, string blobName, string? destination = null)
|
||||
=> _storage.UpdateSessionDiagnosticsBlobAsync(tenantId, sessionId, blobName, destination);
|
||||
|
||||
@@ -606,16 +606,14 @@ namespace AutopilotMonitor.Functions.Functions.Apps
|
||||
return date.AddDays(-diff);
|
||||
}
|
||||
|
||||
// Internal so AppVersionRegressionRadar shares the same nearest-rank convention
|
||||
// (a third percentile implementation would invite subtle median drift).
|
||||
// Internal so AppVersionRegressionRadar shares the same nearest-rank convention.
|
||||
// Thin delegation onto the single shared implementation (PercentileMath) — parallel
|
||||
// implementations would invite subtle median drift.
|
||||
internal static int Percentile(List<int> values, double percentile)
|
||||
{
|
||||
if (values.Count == 0) return 0;
|
||||
var sorted = values.OrderBy(v => v).ToList();
|
||||
var rank = (int)Math.Ceiling(percentile * sorted.Count) - 1;
|
||||
if (rank < 0) rank = 0;
|
||||
if (rank >= sorted.Count) rank = sorted.Count - 1;
|
||||
return sorted[rank];
|
||||
var sorted = values.Select(v => (double)v).OrderBy(v => v).ToList();
|
||||
return (int)Helpers.PercentileMath.Percentile(sorted, percentile);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
+18
-1
@@ -128,7 +128,12 @@ namespace AutopilotMonitor.Functions.Functions.Rules
|
||||
internal static readonly HashSet<string> KnownSources = new(StringComparer.Ordinal)
|
||||
{
|
||||
"event_type", "event_data", "event_data_array", "event_count",
|
||||
"phase_duration", "app_install_duration", "event_correlation",
|
||||
"phase_duration", "app_install_duration", "event_correlation", "clock_skew",
|
||||
};
|
||||
|
||||
internal static readonly HashSet<string> KnownSkewMetrics = new(StringComparer.Ordinal)
|
||||
{
|
||||
"clock_jump", "sustained_offset",
|
||||
};
|
||||
|
||||
internal static readonly HashSet<string> KnownOperators = new(StringComparer.OrdinalIgnoreCase)
|
||||
@@ -196,6 +201,18 @@ namespace AutopilotMonitor.Functions.Functions.Rules
|
||||
errors.Add($"{label}: event_correlation requires joinField");
|
||||
}
|
||||
|
||||
if (string.Equals(c.Source, "clock_skew", StringComparison.Ordinal))
|
||||
{
|
||||
if (string.IsNullOrWhiteSpace(c.SkewMetric) || !KnownSkewMetrics.Contains(c.SkewMetric))
|
||||
errors.Add($"{label}: clock_skew requires skewMetric (one of: {string.Join(", ", KnownSkewMetrics)})");
|
||||
if (!string.Equals(c.Operator, "gt", StringComparison.OrdinalIgnoreCase)
|
||||
&& !string.Equals(c.Operator, "gte", StringComparison.OrdinalIgnoreCase))
|
||||
errors.Add($"{label}: clock_skew supports only gt/gte (magnitude vs threshold)");
|
||||
if (!double.TryParse(c.Value, System.Globalization.NumberStyles.Float,
|
||||
System.Globalization.CultureInfo.InvariantCulture, out var skewThreshold) || skewThreshold <= 0)
|
||||
errors.Add($"{label}: clock_skew requires a positive numeric value (threshold in seconds)");
|
||||
}
|
||||
|
||||
if ((string.Equals(c.Operator, "regex", StringComparison.OrdinalIgnoreCase)
|
||||
|| string.Equals(c.Operator, "not_regex", StringComparison.OrdinalIgnoreCase))
|
||||
&& !string.IsNullOrEmpty(c.Value))
|
||||
|
||||
@@ -0,0 +1,30 @@
|
||||
using System;
|
||||
using System.Collections.Generic;
|
||||
using System.Linq;
|
||||
|
||||
namespace AutopilotMonitor.Functions.Helpers
|
||||
{
|
||||
/// <summary>
|
||||
/// The single nearest-rank percentile implementation (rank = ceil(p·n) − 1, clamped).
|
||||
/// AppsAnalyticsHelper.Percentile and PlatformMetricsService.Percentile delegate here;
|
||||
/// keep it that way — parallel implementations invite subtle median drift between
|
||||
/// surfaces that report the "same" statistic.
|
||||
/// </summary>
|
||||
internal static class PercentileMath
|
||||
{
|
||||
/// <summary>Nearest-rank percentile over an ascending-sorted list.</summary>
|
||||
internal static double Percentile(IReadOnlyList<double> sortedValues, double percentile)
|
||||
{
|
||||
if (sortedValues.Count == 0) return 0;
|
||||
var rank = (int)Math.Ceiling(percentile * sortedValues.Count) - 1;
|
||||
return sortedValues[Math.Max(0, Math.Min(rank, sortedValues.Count - 1))];
|
||||
}
|
||||
|
||||
/// <summary>Median (p50) over unsorted values.</summary>
|
||||
internal static double Median(IEnumerable<double> values)
|
||||
{
|
||||
var sorted = values.OrderBy(v => v).ToList();
|
||||
return Percentile(sorted, 0.50);
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,107 @@
|
||||
using System;
|
||||
using System.Collections.Generic;
|
||||
using AutopilotMonitor.Functions.Helpers;
|
||||
using AutopilotMonitor.Shared.Models;
|
||||
|
||||
namespace AutopilotMonitor.Functions.Services
|
||||
{
|
||||
/// <summary>
|
||||
/// Regression tripwire for the agent's CMTrace time resolution (per-line self-anchoring,
|
||||
/// docs/agent/cmtrace-time-resolution.md). Runs once per session on the terminal ingest
|
||||
/// batch, over samples the counter reconcile collected in its existing partition scan.
|
||||
///
|
||||
/// Detection: diff = median(Δ_IME) − median(Δ_other), where Δ = ReceivedAt − OccurredUtc.
|
||||
/// Shared upload/spool latency is common-mode and cancels in the difference; a timezone
|
||||
/// mis-conversion shifts only the IME side and lands on the 15-minute offset grid.
|
||||
/// Replayed old log lines produce large OFF-grid deltas — the grid-residual test is what
|
||||
/// separates the two (field measurement 2026-08-20: without it, 48 instead of 26 true hits).
|
||||
///
|
||||
/// 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>
|
||||
internal static class CmTraceSkewTripwire
|
||||
{
|
||||
// Same grid the agent anchors on and the same residual tolerance as its own grid
|
||||
// guard (docs/agent/cmtrace-time-resolution.md) — keep these in lockstep with the
|
||||
// agent so a future anchoring change stays traceable to this detector.
|
||||
internal const int GridMinutes = 15;
|
||||
internal const double ResidualToleranceMinutes = 2.0;
|
||||
internal const int MinSamplesPerSide = 20;
|
||||
// ReceivedAt is stamped per upload batch, not per event; a side backed by fewer
|
||||
// distinct batches yields a batch-composition artifact, not a session property.
|
||||
internal const int MinDistinctBatchesPerSide = 3;
|
||||
|
||||
internal sealed class Result
|
||||
{
|
||||
public double MedianImeDeltaMinutes { get; set; }
|
||||
public double MedianOtherDeltaMinutes { get; set; }
|
||||
public double DiffMinutes { get; set; }
|
||||
public int GridSteps { get; set; }
|
||||
public double ResidualMinutes { get; set; }
|
||||
public int ImeSampleCount { get; set; }
|
||||
public int OtherSampleCount { get; set; }
|
||||
public int ImeBatchCount { get; set; }
|
||||
public int OtherBatchCount { 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.
|
||||
/// </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)
|
||||
return null;
|
||||
|
||||
var medianIme = PercentileMath.Median(scan.ImeDeltaMinutes);
|
||||
var medianOther = PercentileMath.Median(scan.OtherDeltaMinutes);
|
||||
var diff = medianIme - medianOther;
|
||||
var steps = (int)Math.Round(diff / GridMinutes, MidpointRounding.AwayFromZero);
|
||||
var residual = Math.Abs(diff - steps * (double)GridMinutes);
|
||||
|
||||
if (Math.Abs(steps) < 1 || residual >= ResidualToleranceMinutes)
|
||||
return null;
|
||||
|
||||
return new Result
|
||||
{
|
||||
MedianImeDeltaMinutes = medianIme,
|
||||
MedianOtherDeltaMinutes = medianOther,
|
||||
DiffMinutes = diff,
|
||||
GridSteps = steps,
|
||||
ResidualMinutes = residual,
|
||||
ImeSampleCount = scan.ImeDeltaMinutes.Count,
|
||||
OtherSampleCount = scan.OtherDeltaMinutes.Count,
|
||||
ImeBatchCount = scan.ImeDistinctBatchCount,
|
||||
OtherBatchCount = scan.OtherDistinctBatchCount,
|
||||
};
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// True when the session's IME events are dominated by writer-declared ("bias")
|
||||
/// offsets — a bias offset comes verbatim from the log line's own zone marker and
|
||||
/// cannot be an anchoring regression, so a grid hit there is not ours to alarm on.
|
||||
/// measuredWriterOffsetMinutes is deliberately NOT consulted: the per-file
|
||||
/// observation sticks to a stale value after an era flip-back (monotonicity guard),
|
||||
/// making it non-authoritative for exactly the sessions this tripwire inspects.
|
||||
/// </summary>
|
||||
internal static bool IsBiasDominated(IReadOnlyDictionary<string, int>? originHistogram)
|
||||
{
|
||||
if (originHistogram == null || originHistogram.Count == 0)
|
||||
return false;
|
||||
|
||||
int total = 0, bias = 0;
|
||||
foreach (var kv in originHistogram)
|
||||
{
|
||||
total += kv.Value;
|
||||
if (string.Equals(kv.Key, "bias", StringComparison.OrdinalIgnoreCase))
|
||||
bias += kv.Value;
|
||||
}
|
||||
return total > 0 && bias * 2 >= total;
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -6,6 +6,7 @@ using AutopilotMonitor.Shared.DataAccess;
|
||||
using AutopilotMonitor.Shared.Models;
|
||||
using Microsoft.ApplicationInsights;
|
||||
using Microsoft.Azure.Functions.Worker;
|
||||
using Microsoft.Extensions.Configuration;
|
||||
using Microsoft.Extensions.Logging;
|
||||
|
||||
namespace AutopilotMonitor.Functions.Services
|
||||
@@ -44,6 +45,10 @@ namespace AutopilotMonitor.Functions.Services
|
||||
private readonly AutopilotMonitor.Functions.Services.Analyze.InterimTriggerRegistry _interimTriggerRegistry;
|
||||
private readonly IVulnerabilityCorrelateProducer _vulnProducer;
|
||||
private readonly Ime.IImeMsiArchiveProducer _imeMsiArchiveProducer;
|
||||
private readonly IConfiguration _configuration;
|
||||
|
||||
/// <summary>App setting kill switch: set to "true" to skip the CMTrace skew tripwire entirely. Fail-open — the tripwire only notifies, it never mutates data.</summary>
|
||||
internal const string CmTraceSkewTripwireKillSwitchSetting = "CmTraceSkewTripwireDisabled";
|
||||
|
||||
public EventIngestProcessor(
|
||||
ILogger<EventIngestProcessor> logger,
|
||||
@@ -60,7 +65,8 @@ namespace AutopilotMonitor.Functions.Services
|
||||
AutopilotMonitor.Functions.Services.Analyze.IAnalyzeOnEnrollmentEndProducer analyzeProducer,
|
||||
AutopilotMonitor.Functions.Services.Analyze.InterimTriggerRegistry interimTriggerRegistry,
|
||||
IVulnerabilityCorrelateProducer vulnProducer,
|
||||
Ime.IImeMsiArchiveProducer imeMsiArchiveProducer)
|
||||
Ime.IImeMsiArchiveProducer imeMsiArchiveProducer,
|
||||
IConfiguration configuration)
|
||||
{
|
||||
_logger = logger;
|
||||
_sessionRepo = sessionRepo;
|
||||
@@ -77,6 +83,7 @@ namespace AutopilotMonitor.Functions.Services
|
||||
_interimTriggerRegistry = interimTriggerRegistry;
|
||||
_vulnProducer = vulnProducer;
|
||||
_imeMsiArchiveProducer = imeMsiArchiveProducer;
|
||||
_configuration = configuration;
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
@@ -247,7 +254,17 @@ namespace AutopilotMonitor.Functions.Services
|
||||
// replays where UpdateSessionStatusAsync no-ops.
|
||||
// Idempotent (no-ops when already correct) and fail-soft.
|
||||
if (isTerminalBatch)
|
||||
await _sessionRepo.ReconcileSessionCountersAsync(request.TenantId, request.SessionId);
|
||||
{
|
||||
var skewScan = await _sessionRepo.ReconcileSessionCountersAsync(request.TenantId, request.SessionId);
|
||||
|
||||
// CMTrace time-skew tripwire (goal state: never fires). Gated on the actual
|
||||
// status transition so terminal-batch REPLAYS (statusTransitioned == false)
|
||||
// cannot re-emit — the reconcile above runs on replays by design, the
|
||||
// tripwire must not. Fail-soft: ingest stays a 200 no matter what.
|
||||
if (skewScan != null && (statusTransitioned || whiteGloveStatusTransitioned))
|
||||
await TryFireCmTraceSkewTripwireAsync(request.TenantId, request.SessionId,
|
||||
updatedSession?.AgentVersion, skewScan);
|
||||
}
|
||||
|
||||
// Auto-analyze fan-out: enqueue a queue message instead of running fire-and-forget
|
||||
// Task.Run inside the function. The previous in-function approach could be killed
|
||||
@@ -545,6 +562,67 @@ namespace AutopilotMonitor.Functions.Services
|
||||
System.Globalization.CultureInfo.InvariantCulture, out value);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// CMTrace time-skew regression tripwire. Evaluates the per-source timestamp-delta
|
||||
/// samples the counter reconcile collected and emits a CmTraceTimeSkewRegression ops
|
||||
/// event when the IME-derived events diverge from the other sources by a clean
|
||||
/// 15-minute-grid multiple (see <see cref="CmTraceSkewTripwire"/>). One extra bounded
|
||||
/// storage read happens ONLY on the suspicion path (≈ never): the sourceOffsetOrigin
|
||||
/// histogram that rules out writer-declared ("bias") offsets, which come verbatim
|
||||
/// from the log line and cannot be an anchoring regression.
|
||||
/// </summary>
|
||||
private async Task TryFireCmTraceSkewTripwireAsync(
|
||||
string tenantId, string sessionId, string? agentVersion, SessionSkewScan skewScan)
|
||||
{
|
||||
try
|
||||
{
|
||||
if (string.Equals(_configuration[CmTraceSkewTripwireKillSwitchSetting], "true", StringComparison.OrdinalIgnoreCase))
|
||||
return;
|
||||
|
||||
var result = CmTraceSkewTripwire.Evaluate(skewScan);
|
||||
if (result == null)
|
||||
return;
|
||||
|
||||
var histogram = await _sessionRepo.GetImeOffsetOriginHistogramAsync(tenantId, sessionId);
|
||||
if (CmTraceSkewTripwire.IsBiasDominated(histogram))
|
||||
{
|
||||
_logger.LogInformation(
|
||||
"CMTrace skew tripwire suppressed for session {SessionId}: grid divergence {DiffMinutes:+0.0;-0.0} min, but IME offsets are bias-dominated (writer-declared)",
|
||||
sessionId, result.DiffMinutes);
|
||||
return;
|
||||
}
|
||||
|
||||
var origins = string.Join(", ",
|
||||
histogram.OrderByDescending(kv => kv.Value).Select(kv => $"{kv.Key} {kv.Value}"));
|
||||
var message =
|
||||
$"IME-derived events skewed {result.DiffMinutes:+0.0;-0.0} min vs other sources " +
|
||||
$"({result.GridSteps}x15min grid, residual {result.ResidualMinutes:0.0} min; " +
|
||||
$"median dIME {result.MedianImeDeltaMinutes:0.0} / dOther {result.MedianOtherDeltaMinutes:0.0} min; " +
|
||||
$"{result.ImeSampleCount}/{result.OtherSampleCount} samples over {result.ImeBatchCount}/{result.OtherBatchCount} batches; " +
|
||||
$"origins: {(origins.Length > 0 ? origins : "none")})";
|
||||
|
||||
await _opsEventService.RecordCmTraceTimeSkewRegressionAsync(tenantId, sessionId, agentVersion, message, new
|
||||
{
|
||||
sessionId,
|
||||
agentVersion,
|
||||
diffMinutes = Math.Round(result.DiffMinutes, 2),
|
||||
gridSteps = result.GridSteps,
|
||||
residualMinutes = Math.Round(result.ResidualMinutes, 2),
|
||||
medianImeDeltaMinutes = Math.Round(result.MedianImeDeltaMinutes, 2),
|
||||
medianOtherDeltaMinutes = Math.Round(result.MedianOtherDeltaMinutes, 2),
|
||||
imeSampleCount = result.ImeSampleCount,
|
||||
otherSampleCount = result.OtherSampleCount,
|
||||
imeBatchCount = result.ImeBatchCount,
|
||||
otherBatchCount = result.OtherBatchCount,
|
||||
sourceOffsetOrigins = histogram,
|
||||
});
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
_logger.LogWarning(ex, "CMTrace skew tripwire failed for session {SessionId}", sessionId);
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Stamps authoritative server-side fields onto all events before storage.
|
||||
/// TenantId and SessionId always come from the validated request metadata,
|
||||
|
||||
@@ -622,6 +622,22 @@ namespace AutopilotMonitor.Functions.Services
|
||||
$"Agent binary integrity mismatch on session {sessionId ?? "?"} (agent {agentVersion ?? "?"}): {message}",
|
||||
tenantId, "System.EmergencyChannel", new { sessionId, agentVersion });
|
||||
|
||||
/// <summary>
|
||||
/// The CMTrace time-skew tripwire fired: a terminal session's IME-derived event
|
||||
/// timestamps diverge from its other events by a clean 15-minute-grid multiple —
|
||||
/// the signature of a timezone mis-conversion the per-line self-anchoring
|
||||
/// (docs/agent/cmtrace-time-resolution.md) failed to catch. Goal state: this event
|
||||
/// NEVER fires; any occurrence is a real anchoring regression or a detector bug,
|
||||
/// both actionable. Emitted by <see cref="EventIngestProcessor"/> once per session
|
||||
/// (gated on the terminal status transition). The message carries the full numbers
|
||||
/// so the event is verifiable without a portal round-trip.
|
||||
/// Dual-register per memory feedback_ops_event_types_dual_register.
|
||||
/// </summary>
|
||||
public Task RecordCmTraceTimeSkewRegressionAsync(string tenantId, string sessionId, string? agentVersion, string message, object details)
|
||||
=> WriteAsync(OpsEventCategory.Agent, "CmTraceTimeSkewRegression", OpsEventSeverity.Warning,
|
||||
$"CMTrace time-skew tripwire on session {sessionId} (agent {agentVersion ?? "?"}): {message}",
|
||||
tenantId, "System.Ingest", details);
|
||||
|
||||
public Task RecordExcessiveSessionEventsAsync(string tenantId, string sessionId, int eventCount, int threshold)
|
||||
=> WriteAsync(OpsEventCategory.Agent, "ExcessiveSessionEvents", OpsEventSeverity.Warning,
|
||||
$"Session {sessionId} has {eventCount} events (threshold {threshold}) — likely agent loop bug",
|
||||
|
||||
@@ -312,12 +312,9 @@ namespace AutopilotMonitor.Functions.Services
|
||||
};
|
||||
}
|
||||
|
||||
// Thin delegation onto the single shared nearest-rank implementation (PercentileMath).
|
||||
private static double Percentile(List<double> sortedValues, double percentile)
|
||||
{
|
||||
if (sortedValues.Count == 0) return 0;
|
||||
var index = (int)Math.Ceiling(percentile * sortedValues.Count) - 1;
|
||||
return sortedValues[Math.Max(0, Math.Min(index, sortedValues.Count - 1))];
|
||||
}
|
||||
=> Helpers.PercentileMath.Percentile(sortedValues, percentile);
|
||||
|
||||
// Payload getters live in AgentMetricsAggregation (shared with the efficiency
|
||||
// service); these thin wrappers keep the call sites above unchanged.
|
||||
|
||||
@@ -98,6 +98,9 @@ namespace AutopilotMonitor.Functions.Services
|
||||
case "event_correlation":
|
||||
return EvaluateEventCorrelationCondition(condition, events);
|
||||
|
||||
case "clock_skew":
|
||||
return EvaluateClockSkewCondition(condition, events);
|
||||
|
||||
default:
|
||||
return (false, "unknown source");
|
||||
}
|
||||
@@ -808,6 +811,181 @@ namespace AutopilotMonitor.Functions.Services
|
||||
return (false, "no app install duration matched");
|
||||
}
|
||||
|
||||
// ===== clock_skew constants =====
|
||||
// Named constants, deliberately NOT condition fields: promoting a constant to a field
|
||||
// later is non-breaking, the reverse is not.
|
||||
/// <summary>Frame windows (in upload batches) that must agree before/after a step for it to count as a clock jump.</summary>
|
||||
internal const int ClockSkewConfirmBatches = 3;
|
||||
/// <summary>Floor for the per-batch internal-spread cap, seconds.</summary>
|
||||
internal const double ClockSkewSpoolSpreadFloorSeconds = 60;
|
||||
/// <summary>Per-batch internal-spread cap as a fraction of the condition threshold.</summary>
|
||||
internal const double ClockSkewSpoolSpreadCapFraction = 0.5;
|
||||
/// <summary>Fraction of surviving batches that must sit near the overall median for a sustained-offset verdict.</summary>
|
||||
internal const double ClockSkewStableBatchFraction = 0.75;
|
||||
|
||||
/// <summary>
|
||||
/// Detects DEVICE clock problems from the per-event offset d = Timestamp − ReceivedAt
|
||||
/// (device frame minus server frame, in seconds; healthy baseline ≈ −upload-latency,
|
||||
/// i.e. a few seconds negative).
|
||||
/// skewMetric "clock_jump": a persistent step in the device's clock frame mid-session
|
||||
/// (time sync correcting a wrong RTC, Windows auto-timezone, manual change).
|
||||
/// skewMetric "sustained_offset": the whole session ran on a clock off by ≥ threshold
|
||||
/// — the during-enrollment complement of ANALYZE-DEV-007's one-shot NTP check.
|
||||
///
|
||||
/// Ordering key is Sequence, never Timestamp (see EvaluateAppInstallDurationCondition).
|
||||
/// This evaluator then inspects Timestamp AGAINST that Sequence order ON PURPOSE — a
|
||||
/// device-clock problem is precisely a Timestamp anomaly relative to the clock-immune
|
||||
/// Sequence/ReceivedAt frame. Timestamp is a MEASUREMENT here, never a sort key.
|
||||
///
|
||||
/// CMTrace-derived events (Source == "ImeLogTracker") are excluded up front: their
|
||||
/// Timestamp carries the log WRITER's timezone belief, so an anchoring regression must
|
||||
/// fire the operator-side CmTraceSkewTripwire, never this customer-facing rule.
|
||||
/// </summary>
|
||||
private (bool matched, object evidence) EvaluateClockSkewCondition(RuleCondition condition, List<EnrollmentEvent> events)
|
||||
{
|
||||
var metric = condition.SkewMetric ?? string.Empty;
|
||||
if (!double.TryParse(condition.Value, System.Globalization.NumberStyles.Float,
|
||||
System.Globalization.CultureInfo.InvariantCulture, out var thresholdSeconds) || thresholdSeconds <= 0)
|
||||
return (false, "clock_skew requires a positive numeric value (threshold in seconds)");
|
||||
|
||||
var usable = events
|
||||
.Where(e => e.ReceivedAt.HasValue
|
||||
&& !e.TimestampClamped
|
||||
&& !string.Equals(e.Source, "ImeLogTracker", StringComparison.OrdinalIgnoreCase))
|
||||
.OrderBy(e => e.Sequence)
|
||||
.ToList();
|
||||
if (usable.Count == 0)
|
||||
return (false, "no events with a server receive frame");
|
||||
|
||||
// Group into upload batches (ReceivedAt is stamped once per ingest request). A clock
|
||||
// offset shifts a whole batch uniformly (internal spread ≈ upload debounce, seconds);
|
||||
// a spool-backlog batch spans the offline period (spread = minutes–hours). Dropping
|
||||
// wide-spread batches is what makes replayed backlogs inert.
|
||||
var spreadCap = Math.Max(ClockSkewSpoolSpreadFloorSeconds, thresholdSeconds * ClockSkewSpoolSpreadCapFraction);
|
||||
var batches = new List<(double MedianOffsetSeconds, EnrollmentEvent FirstEvent)>();
|
||||
int usedSamples = 0;
|
||||
foreach (var batch in usable.GroupBy(e => e.ReceivedAt!.Value))
|
||||
{
|
||||
var offsets = batch.Select(e => (e.Timestamp - e.ReceivedAt!.Value).TotalSeconds).OrderBy(v => v).ToList();
|
||||
if (offsets[offsets.Count - 1] - offsets[0] > spreadCap)
|
||||
continue;
|
||||
usedSamples += offsets.Count;
|
||||
batches.Add((Helpers.PercentileMath.Percentile(offsets, 0.50), batch.First()));
|
||||
}
|
||||
|
||||
switch (metric)
|
||||
{
|
||||
case "sustained_offset":
|
||||
{
|
||||
if (batches.Count < 2)
|
||||
return (false, "insufficient upload batches for an offset measurement");
|
||||
|
||||
var batchMedians = batches.Select(b => b.MedianOffsetSeconds).ToList();
|
||||
var overall = Helpers.PercentileMath.Median(batchMedians);
|
||||
|
||||
// Constant offset, not drift chaos: most batches must individually sit near
|
||||
// the session median, otherwise no single "the clock was off by X" statement
|
||||
// is honest.
|
||||
var stable = batchMedians.Count(m => Math.Abs(m - overall) <= spreadCap);
|
||||
if (stable < Math.Max(2, (int)Math.Ceiling(batchMedians.Count * ClockSkewStableBatchFraction)))
|
||||
return (false, "batch offsets too unstable for a sustained-offset verdict");
|
||||
|
||||
if (!CompareClockSkewThreshold(Math.Abs(overall), condition.Operator, thresholdSeconds))
|
||||
return (false, $"sustained offset {overall:0} s below threshold");
|
||||
|
||||
var first = batches[0].FirstEvent;
|
||||
var offsetMinutes = Math.Round(overall / 60.0, 1);
|
||||
var direction = overall > 0 ? "ahead" : "behind";
|
||||
return (true, new Dictionary<string, object>
|
||||
{
|
||||
// field/value pair drives {{skewSummary}} template interpolation
|
||||
// (lib/interpolateRuleTemplate.ts resolves field→value first).
|
||||
["field"] = "skewSummary",
|
||||
["value"] = $"clock ran {Math.Abs(offsetMinutes)} min {direction} of server time throughout the session",
|
||||
["offsetSeconds"] = Math.Round(overall, 0),
|
||||
["offsetMinutes"] = offsetMinutes,
|
||||
["direction"] = direction,
|
||||
["eventId"] = first.EventId,
|
||||
["sequence"] = first.Sequence,
|
||||
["timestamp"] = first.Timestamp,
|
||||
["sampleCount"] = usedSamples,
|
||||
["batchCount"] = batches.Count,
|
||||
});
|
||||
}
|
||||
|
||||
case "clock_jump":
|
||||
{
|
||||
const int w = ClockSkewConfirmBatches;
|
||||
if (batches.Count < 2 * w)
|
||||
return (false, "insufficient upload batches for a frame comparison");
|
||||
|
||||
var medians = batches.Select(b => b.MedianOffsetSeconds).ToList();
|
||||
(int Index, double Pre, double Post, double Jump)? best = null;
|
||||
for (int j = w - 1; j <= batches.Count - 1 - w; j++)
|
||||
{
|
||||
var preWindow = medians.Skip(j - w + 1).Take(w).ToList();
|
||||
var postWindow = medians.Skip(j + 1).Take(w).ToList();
|
||||
|
||||
// Both comparison windows must be flat plateaus: a genuine clock change
|
||||
// is a step between two stable frames, while a gradually draining spool
|
||||
// backlog produces a RAMP of batch medians — sloping windows would let
|
||||
// any pre/post pair straddling the ramp fake a step.
|
||||
if (preWindow.Max() - preWindow.Min() > spreadCap || postWindow.Max() - postWindow.Min() > spreadCap)
|
||||
continue;
|
||||
|
||||
var pre = Helpers.PercentileMath.Median(preWindow);
|
||||
var post = Helpers.PercentileMath.Median(postWindow);
|
||||
var jump = post - pre;
|
||||
if (CompareClockSkewThreshold(Math.Abs(jump), condition.Operator, thresholdSeconds)
|
||||
&& (best == null || Math.Abs(jump) > Math.Abs(best.Value.Jump)))
|
||||
best = (j, pre, post, jump);
|
||||
}
|
||||
if (best == null)
|
||||
return (false, "no persistent clock step");
|
||||
|
||||
// End-state persistence: a transient excursion that returns to the original
|
||||
// frame (spool backlog draining, historic replay) must stay silent; a genuine
|
||||
// clock change — including "wrong clock corrected by time sync mid-session" —
|
||||
// persists to session end, because the corrected frame IS the change.
|
||||
var tail = Helpers.PercentileMath.Median(medians.Skip(medians.Count - w).Take(w));
|
||||
var tailShift = tail - best.Value.Pre;
|
||||
if (Math.Abs(tailShift) < thresholdSeconds || Math.Sign(tailShift) != Math.Sign(best.Value.Jump))
|
||||
return (false, "transient timestamp excursion, not a sustained clock change");
|
||||
|
||||
var boundary = batches[best.Value.Index + 1].FirstEvent;
|
||||
var jumpMinutes = Math.Round(best.Value.Jump / 60.0, 1);
|
||||
var direction = best.Value.Jump > 0 ? "forward" : "backward";
|
||||
return (true, new Dictionary<string, object>
|
||||
{
|
||||
// field/value pair drives {{skewSummary}} template interpolation
|
||||
// (lib/interpolateRuleTemplate.ts resolves field→value first).
|
||||
["field"] = "skewSummary",
|
||||
["value"] = $"clock jumped {Math.Abs(jumpMinutes)} min {direction} mid-session",
|
||||
["jumpSeconds"] = Math.Round(best.Value.Jump, 0),
|
||||
["jumpMinutes"] = jumpMinutes,
|
||||
["direction"] = direction,
|
||||
["beforeOffsetMinutes"] = Math.Round(best.Value.Pre / 60.0, 1),
|
||||
["afterOffsetMinutes"] = Math.Round(best.Value.Post / 60.0, 1),
|
||||
["eventId"] = boundary.EventId,
|
||||
["sequence"] = boundary.Sequence,
|
||||
["timestamp"] = boundary.Timestamp,
|
||||
["eventType"] = boundary.EventType,
|
||||
["sampleCount"] = usedSamples,
|
||||
["batchCount"] = batches.Count,
|
||||
});
|
||||
}
|
||||
|
||||
default:
|
||||
return (false, $"unknown skewMetric '{metric}' (expected clock_jump or sustained_offset)");
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>gt/gte magnitude comparison for clock_skew; unknown operators behave as gte (validation rejects them upstream).</summary>
|
||||
private static bool CompareClockSkewThreshold(double magnitude, string? op, double thresholdSeconds)
|
||||
=> string.Equals(op, "gt", StringComparison.OrdinalIgnoreCase)
|
||||
? magnitude > thresholdSeconds
|
||||
: magnitude >= thresholdSeconds;
|
||||
|
||||
/// <summary>
|
||||
/// Generic event correlation: finds pairs of events (A before B) sharing
|
||||
/// the same value in a join field, with optional filters on each event
|
||||
|
||||
@@ -2418,17 +2418,20 @@ namespace AutopilotMonitor.Functions.Services
|
||||
/// are left untouched rather than being zeroed. Idempotent: no-ops when the row is
|
||||
/// already correct. PlatformScriptCount/RemediationScriptCount stay increment-only (their
|
||||
/// drift is bounded and they feed no enforcement heuristics; EventCount does).
|
||||
/// Returns the skew-scan samples collected during the authoritative count's partition
|
||||
/// scan (CmTraceSkewTripwire input), or null when the scan itself failed — the counter
|
||||
/// reconcile outcome does not affect the returned scan.
|
||||
/// </summary>
|
||||
public async Task ReconcileSessionCountersAsync(string tenantId, string sessionId)
|
||||
public async Task<SessionSkewScan?> ReconcileSessionCountersAsync(string tenantId, string sessionId)
|
||||
{
|
||||
SecurityValidator.EnsureValidGuid(tenantId, nameof(tenantId));
|
||||
SecurityValidator.EnsureValidGuid(sessionId, nameof(sessionId));
|
||||
|
||||
var authoritative = await CountSessionEventRowsAsync(tenantId, sessionId);
|
||||
if (!authoritative.HasValue)
|
||||
return; // storage error — keep the incremental live values, do not zero them
|
||||
return null; // storage error — keep the incremental live values, do not zero them
|
||||
|
||||
var (totalEvents, rebootEvents) = authoritative.Value;
|
||||
var (totalEvents, rebootEvents, skewScan) = authoritative.Value;
|
||||
|
||||
const int maxRetries = 5;
|
||||
int retryCount = 0;
|
||||
@@ -2440,7 +2443,7 @@ namespace AutopilotMonitor.Functions.Services
|
||||
var tableClient = _tableServiceClient.GetTableClient(Constants.TableNames.Sessions);
|
||||
var entityResponse = await tableClient.GetEntityIfExistsAsync<TableEntity>(tenantId, sessionId);
|
||||
if (!entityResponse.HasValue)
|
||||
return;
|
||||
return skewScan;
|
||||
var entity = entityResponse.Value!;
|
||||
|
||||
// No-op when already correct — avoids a needless write + index sync on the
|
||||
@@ -2459,7 +2462,7 @@ namespace AutopilotMonitor.Functions.Services
|
||||
drifted = true;
|
||||
}
|
||||
if (!drifted)
|
||||
return;
|
||||
return skewScan;
|
||||
|
||||
var currentStartedAt = entity.GetDateTimeOffset("StartedAt")?.UtcDateTime ?? DateTime.MaxValue;
|
||||
|
||||
@@ -2468,7 +2471,7 @@ namespace AutopilotMonitor.Functions.Services
|
||||
// Dual-write: keep SessionsIndex in sync (merge — no StartedAt shift here).
|
||||
await SyncSessionIndexAsync(tenantId, sessionId, entity, update, currentStartedAt, null);
|
||||
|
||||
return;
|
||||
return skewScan;
|
||||
}
|
||||
catch (Azure.RequestFailedException ex) when (ex.Status == 412)
|
||||
{
|
||||
@@ -2476,7 +2479,7 @@ namespace AutopilotMonitor.Functions.Services
|
||||
if (retryCount >= maxRetries)
|
||||
{
|
||||
_logger.LogWarning($"Failed to reconcile session counters for session {sessionId} after {maxRetries} retries due to ETag conflicts");
|
||||
return;
|
||||
return skewScan;
|
||||
}
|
||||
var baseDelay = 50 * (int)Math.Pow(2, retryCount - 1);
|
||||
await Task.Delay(baseDelay + Random.Shared.Next(0, baseDelay));
|
||||
@@ -2484,9 +2487,11 @@ namespace AutopilotMonitor.Functions.Services
|
||||
catch (Exception ex)
|
||||
{
|
||||
_logger.LogError(ex, $"Failed to reconcile session counters for session {sessionId}");
|
||||
return;
|
||||
return skewScan;
|
||||
}
|
||||
}
|
||||
|
||||
return skewScan;
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
@@ -3287,9 +3292,10 @@ namespace AutopilotMonitor.Functions.Services
|
||||
/// and a unique RowKey per event, so the number of stored rows equals the true distinct
|
||||
/// count — making this an idempotent, retry-immune authoritative count (used to reconcile
|
||||
/// EventCount + RebootCount at the terminal transition). Single partition scan with a
|
||||
/// RowKey+EventType projection so both counts cost one query.
|
||||
/// slim projection so both counts cost one query. The CmTraceSkewTripwire's per-source
|
||||
/// timestamp-delta samples piggyback on the same enumeration (zero extra reads).
|
||||
/// </summary>
|
||||
private async Task<(int Total, int Reboots)?> CountSessionEventRowsAsync(string tenantId, string sessionId)
|
||||
private async Task<(int Total, int Reboots, SessionSkewScan Skew)?> CountSessionEventRowsAsync(string tenantId, string sessionId)
|
||||
{
|
||||
try
|
||||
{
|
||||
@@ -3297,17 +3303,38 @@ namespace AutopilotMonitor.Functions.Services
|
||||
var partitionKey = $"{tenantId}_{sessionId}";
|
||||
var query = tableClient.QueryAsync<TableEntity>(
|
||||
filter: $"PartitionKey eq '{partitionKey}'",
|
||||
select: new[] { "RowKey", "EventType" }
|
||||
select: new[] { "RowKey", "EventType", "Source", "ReceivedAt", BusinessTimestamp.OccurredUtcColumn, "TimestampClamped" }
|
||||
);
|
||||
|
||||
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++;
|
||||
if (entity.GetString("EventType") == Constants.EventTypes.SystemRebootDetected)
|
||||
reboots++;
|
||||
|
||||
// Skew sample. Clamped events carry a server-substituted OccurredUtc
|
||||
// (Δ ≈ 0) and would dilute the medians; legacy rows without ReceivedAt
|
||||
// have no server frame at all — both are skipped, never zero-filled.
|
||||
if (entity.GetBoolean("TimestampClamped") == true)
|
||||
continue;
|
||||
var receivedAt = entity.GetDateTimeOffset("ReceivedAt")?.UtcDateTime;
|
||||
var occurredAt = ResolveEventTimestampOrNull(entity);
|
||||
if (!receivedAt.HasValue || !occurredAt.HasValue)
|
||||
continue;
|
||||
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);
|
||||
}
|
||||
return (total, reboots);
|
||||
skew.ImeDistinctBatchCount = imeBatches.Count;
|
||||
skew.OtherDistinctBatchCount = otherBatches.Count;
|
||||
return (total, reboots, skew);
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
@@ -3316,6 +3343,49 @@ namespace AutopilotMonitor.Functions.Services
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Bounded histogram of the <c>sourceOffsetOrigin</c> provenance values ("bias",
|
||||
/// "line-anchored", "reader-zone-fallback", retired "calibrated") over a session's
|
||||
/// IME-derived events. Only read on the CmTraceSkewTripwire suspicion path (≈ never),
|
||||
/// so the row cap bounds the worst case rather than tuning the common one. Fail-soft:
|
||||
/// storage errors return an empty map, which the caller treats as "not bias-dominated"
|
||||
/// — the tripwire fails open toward reporting.
|
||||
/// </summary>
|
||||
public async Task<Dictionary<string, int>> GetImeOffsetOriginHistogramAsync(string tenantId, string sessionId, int maxRows = 500)
|
||||
{
|
||||
SecurityValidator.EnsureValidGuid(tenantId, nameof(tenantId));
|
||||
SecurityValidator.EnsureValidGuid(sessionId, nameof(sessionId));
|
||||
|
||||
var histogram = new Dictionary<string, int>(StringComparer.OrdinalIgnoreCase);
|
||||
try
|
||||
{
|
||||
var tableClient = _tableServiceClient.GetTableClient(Constants.TableNames.Events);
|
||||
var partitionKey = $"{tenantId}_{sessionId}";
|
||||
var query = tableClient.QueryAsync<TableEntity>(
|
||||
filter: $"PartitionKey eq '{partitionKey}' and Source eq 'ImeLogTracker'",
|
||||
select: new[] { "RowKey", "DataJson" });
|
||||
|
||||
int rows = 0;
|
||||
await foreach (var entity in query)
|
||||
{
|
||||
if (++rows > maxRows)
|
||||
break;
|
||||
var data = DeserializeEventData(entity.GetString("DataJson"));
|
||||
if (data != null && data.TryGetValue("sourceOffsetOrigin", out var origin))
|
||||
{
|
||||
var key = origin?.ToString();
|
||||
if (!string.IsNullOrEmpty(key))
|
||||
histogram[key!] = histogram.TryGetValue(key!, out var count) ? count + 1 : 1;
|
||||
}
|
||||
}
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
_logger.LogDebug(ex, $"Could not read IME offset-origin histogram for session {sessionId}");
|
||||
}
|
||||
return histogram;
|
||||
}
|
||||
|
||||
#region IME Version History
|
||||
|
||||
/// <summary>
|
||||
|
||||
@@ -361,6 +361,41 @@ describe('analyze semantic lint', () => {
|
||||
expect(errors(validateRuleDraft(corr).findings).some((m) => m.includes('joinField'))).toBe(true);
|
||||
});
|
||||
|
||||
it('clock_skew: missing/unknown skewMetric, bad operator, bad threshold are errors; {{skewSummary}} resolves', () => {
|
||||
const missingMetric = validAnalyze();
|
||||
(missingMetric.conditions as Array<Record<string, unknown>>)[0] = {
|
||||
signal: 'clock_jump', source: 'clock_skew', operator: 'gte', value: '300', required: true,
|
||||
};
|
||||
expect(errors(validateRuleDraft(missingMetric).findings).some((m) => m.includes('skewMetric'))).toBe(true);
|
||||
|
||||
const badOperator = validAnalyze();
|
||||
(badOperator.conditions as Array<Record<string, unknown>>)[0] = {
|
||||
signal: 'clock_jump', source: 'clock_skew', skewMetric: 'clock_jump', operator: 'lt', value: '300', required: true,
|
||||
};
|
||||
expect(errors(validateRuleDraft(badOperator).findings).some((m) => m.includes('gt/gte'))).toBe(true);
|
||||
|
||||
const badThreshold = validAnalyze();
|
||||
(badThreshold.conditions as Array<Record<string, unknown>>)[0] = {
|
||||
signal: 'clock_jump', source: 'clock_skew', skewMetric: 'clock_jump', operator: 'gte', value: '-5', required: true,
|
||||
};
|
||||
expect(errors(validateRuleDraft(badThreshold).findings).some((m) => m.includes('positive numeric'))).toBe(true);
|
||||
|
||||
const noisy = validAnalyze();
|
||||
(noisy.conditions as Array<Record<string, unknown>>)[0] = {
|
||||
signal: 'clock_jump', source: 'clock_skew', skewMetric: 'clock_jump', operator: 'gte', value: '30', required: true,
|
||||
};
|
||||
expect(warnings(validateRuleDraft(noisy).findings).some((m) => m.includes('upload-latency noise'))).toBe(true);
|
||||
|
||||
const good = validAnalyze();
|
||||
(good.conditions as Array<Record<string, unknown>>)[0] = {
|
||||
signal: 'clock_jump', source: 'clock_skew', skewMetric: 'clock_jump', operator: 'gte', value: '300', required: true,
|
||||
};
|
||||
good.explanation = 'This device {{skewSummary}}.';
|
||||
const r = validateRuleDraft(good);
|
||||
expect(errors(r.findings)).toEqual([]);
|
||||
expect(warnings(r.findings).filter((m) => m.includes('{{'))).toEqual([]);
|
||||
});
|
||||
|
||||
it('invalid regex is an error; unanchored not_regex warns', () => {
|
||||
const bad = validAnalyze();
|
||||
(bad.conditions as Array<Record<string, unknown>>)[0] = {
|
||||
|
||||
@@ -165,6 +165,11 @@ export const RULE_AUTHORING_GUIDE = {
|
||||
'Join two event types over a shared field: eventType + correlateEventType + joinField, ' +
|
||||
'optional timeWindowSeconds, optional eventAFilterField/-Operator/-Value to pre-filter the ' +
|
||||
'first event, optional dataField+operator+value tested on the correlated pair.',
|
||||
clock_skew:
|
||||
'Device-clock check against the server receive frame: skewMetric=clock_jump (persistent ' +
|
||||
'mid-session step) or sustained_offset (whole session off by >= value). value = threshold ' +
|
||||
'seconds, operator gt/gte on the magnitude. No eventType/dataField; IME-log-derived events ' +
|
||||
'are excluded from the measurement.',
|
||||
},
|
||||
caseSensitivity:
|
||||
'source values are matched CASE-SENSITIVELY by the engine — write them exactly as listed ' +
|
||||
@@ -270,7 +275,7 @@ export const RULE_AUTHORING_GUIDE = {
|
||||
gatherOutputSeverities: ['Info', 'Warning', 'Error', 'Critical'],
|
||||
gatherCollectorTypes: ['registry', 'eventlog', 'wmi', 'file', 'json', 'xml', 'command_allowlisted', 'logparser'],
|
||||
gatherTriggers: ['startup', 'phase_change', 'phase_exit', 'interval', 'on_event'],
|
||||
analyzeConditionSources: ['event_type', 'event_data', 'event_data_array', 'phase_duration', 'event_count', 'app_install_duration', 'event_correlation'],
|
||||
analyzeConditionSources: ['event_type', 'event_data', 'event_data_array', 'phase_duration', 'event_count', 'app_install_duration', 'event_correlation', 'clock_skew'],
|
||||
analyzeEvaluateOnTriggers: ['enrollment_end', 'whiteglove_sealed', 'on_event:<eventType>'],
|
||||
},
|
||||
|
||||
|
||||
@@ -439,7 +439,16 @@ export const ANALYZE_RULE_SCHEMA: Record<string, unknown> = {
|
||||
"phase_duration",
|
||||
"event_count",
|
||||
"app_install_duration",
|
||||
"event_correlation"
|
||||
"event_correlation",
|
||||
"clock_skew"
|
||||
]
|
||||
},
|
||||
"skewMetric": {
|
||||
"type": "string",
|
||||
"description": "For clock_skew only: which device-clock metric to evaluate. clock_jump = persistent step in the device's clock frame mid-session; sustained_offset = whole session ran on a clock off by at least the threshold. value is the threshold in seconds; operator is limited to gt/gte on the magnitude. IME-log-derived events are excluded from the measurement.",
|
||||
"enum": [
|
||||
"clock_jump",
|
||||
"sustained_offset"
|
||||
]
|
||||
},
|
||||
"eventType": {
|
||||
|
||||
@@ -238,6 +238,7 @@ interface DraftCondition {
|
||||
required?: boolean;
|
||||
correlateEventType?: string;
|
||||
joinField?: string;
|
||||
skewMetric?: string;
|
||||
}
|
||||
|
||||
function checkEventTypeKnown(eventType: string | undefined, where: string): ValidationFinding[] {
|
||||
@@ -293,6 +294,24 @@ function lintAnalyzeRule(rule: Record<string, unknown>): ValidationFinding[] {
|
||||
findings.push(...checkEventTypeKnown(c.correlateEventType, `${label}.correlateEventType`));
|
||||
}
|
||||
|
||||
if (c.source === 'clock_skew') {
|
||||
if (c.skewMetric !== 'clock_jump' && c.skewMetric !== 'sustained_offset') {
|
||||
findings.push({ level: 'error', message: `${label}: clock_skew requires skewMetric ("clock_jump" or "sustained_offset").` });
|
||||
}
|
||||
if (c.operator !== 'gt' && c.operator !== 'gte') {
|
||||
findings.push({ level: 'error', message: `${label}: clock_skew supports only gt/gte — the threshold is compared against the skew MAGNITUDE.` });
|
||||
}
|
||||
const threshold = Number(c.value);
|
||||
if (!c.value || !Number.isFinite(threshold) || threshold <= 0) {
|
||||
findings.push({ level: 'error', message: `${label}: clock_skew requires a positive numeric value (threshold in seconds).` });
|
||||
} else if (threshold < 60) {
|
||||
findings.push({ level: 'warning', message: `${label}: clock_skew threshold below 60 s sits inside normal upload-latency noise — expect false positives.` });
|
||||
}
|
||||
if (c.eventType || c.dataField) {
|
||||
findings.push({ level: 'warning', message: `${label}: clock_skew ignores eventType/dataField — it measures Timestamp against ReceivedAt across the whole session.` });
|
||||
}
|
||||
}
|
||||
|
||||
findings.push(...checkEventTypeKnown(c.eventType, label));
|
||||
|
||||
if ((c.operator === 'regex' || c.operator === 'not_regex') && c.value) {
|
||||
@@ -364,6 +383,9 @@ function lintAnalyzeRule(rule: Record<string, unknown>): ValidationFinding[] {
|
||||
// event_data_array evidence carries the matched element under field=itemField,
|
||||
// so {{itemField}} interpolates at runtime (see EvaluateEventDataArrayCondition).
|
||||
if (c.itemField) resolvable.add(c.itemField);
|
||||
// clock_skew evidence carries a preformatted summary under field=skewSummary,
|
||||
// so {{skewSummary}} interpolates at runtime (see EvaluateClockSkewCondition).
|
||||
if (c.source === 'clock_skew') resolvable.add('skewSummary');
|
||||
}
|
||||
for (const token of extractTokens(rule)) {
|
||||
if (!resolvable.has(token)) {
|
||||
|
||||
@@ -109,8 +109,17 @@ namespace AutopilotMonitor.Shared.DataAccess
|
||||
/// Reconciles the stored EventCount and RebootCount with the authoritative row counts
|
||||
/// from the Events table (rows dedupe on deterministic RowKeys; the read-modify-write
|
||||
/// increments do not). Call as the last counter write on terminal ingest batches.
|
||||
/// Returns the per-source timestamp-delta samples collected during the same partition
|
||||
/// scan (input to the CMTrace time-skew tripwire), or null when the scan failed.
|
||||
/// </summary>
|
||||
Task ReconcileSessionCountersAsync(string tenantId, string sessionId);
|
||||
Task<SessionSkewScan?> ReconcileSessionCountersAsync(string tenantId, string sessionId);
|
||||
|
||||
/// <summary>
|
||||
/// Bounded histogram of the sourceOffsetOrigin provenance values over a session's
|
||||
/// IME-derived events ("bias" / "line-anchored" / "reader-zone-fallback"). Read only
|
||||
/// on the CMTrace skew tripwire's suspicion path; fail-soft (empty map on error).
|
||||
/// </summary>
|
||||
Task<Dictionary<string, int>> GetImeOffsetOriginHistogramAsync(string tenantId, string sessionId, int maxRows = 500);
|
||||
Task UpdateSessionDiagnosticsBlobAsync(
|
||||
string tenantId, string sessionId, string blobName, string? destination = null);
|
||||
Task SetSessionPreProvisionedAsync(string tenantId, string sessionId, bool isPreProvisioned,
|
||||
|
||||
@@ -222,11 +222,19 @@ namespace AutopilotMonitor.Shared.Models
|
||||
|
||||
/// <summary>
|
||||
/// Source of the signal: "event_type", "event_data", "event_data_array", "phase_duration",
|
||||
/// "event_count", "app_install_duration", "event_correlation".
|
||||
/// "event_count", "app_install_duration", "event_correlation", "clock_skew".
|
||||
/// Must stay in sync with DryRunAnalyzeRuleFunction.KnownSources and the RuleEngine switch.
|
||||
/// </summary>
|
||||
public string Source { get; set; } = default!;
|
||||
|
||||
/// <summary>
|
||||
/// For "clock_skew" only: which device-clock metric to evaluate — "clock_jump"
|
||||
/// (persistent step in the device's clock frame mid-session) or "sustained_offset"
|
||||
/// (whole session ran on a clock off by ≥ Value). Value is the threshold in seconds;
|
||||
/// Operator is limited to gt/gte on the magnitude.
|
||||
/// </summary>
|
||||
public string? SkewMetric { get; set; }
|
||||
|
||||
/// <summary>
|
||||
/// Event type to match on.
|
||||
/// For "event_type"/"event_data": the event type to match.
|
||||
|
||||
@@ -0,0 +1,31 @@
|
||||
using System.Collections.Generic;
|
||||
|
||||
namespace AutopilotMonitor.Shared.Models
|
||||
{
|
||||
/// <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
|
||||
/// source: CMTrace-derived events (Source == "ImeLogTracker") versus everything else.
|
||||
/// 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.
|
||||
/// </summary>
|
||||
public class SessionSkewScan
|
||||
{
|
||||
/// <summary>(ReceivedAt − OccurredUtc) in minutes for events with Source == "ImeLogTracker".</summary>
|
||||
public List<double> ImeDeltaMinutes { get; } = new List<double>();
|
||||
|
||||
/// <summary>(ReceivedAt − OccurredUtc) in minutes for all other event sources.</summary>
|
||||
public List<double> OtherDeltaMinutes { get; } = new List<double>();
|
||||
|
||||
/// <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.
|
||||
/// </summary>
|
||||
public int ImeDistinctBatchCount { get; set; }
|
||||
|
||||
/// <summary>Distinct ReceivedAt values seen on the non-IME side.</summary>
|
||||
public int OtherDistinctBatchCount { get; set; }
|
||||
}
|
||||
}
|
||||
@@ -151,6 +151,12 @@ const OPS_EVENT_TYPES: Record<string, string[]> = {
|
||||
// build). Emitted by ReportAgentErrorFunction (RecordAgentBinaryIntegrityMismatchAsync).
|
||||
// Dual-register per memory feedback_ops_event_types_dual_register.
|
||||
"AgentBinaryIntegrityMismatch",
|
||||
// CMTrace time-skew regression tripwire: a terminal session's IME-derived event
|
||||
// timestamps diverged from its other events by a clean 15-minute-grid multiple — the
|
||||
// per-line self-anchoring missed a case. Goal state: never fires. Emitted by
|
||||
// EventIngestProcessor (RecordCmTraceTimeSkewRegressionAsync). Dual-register per memory
|
||||
// feedback_ops_event_types_dual_register.
|
||||
"CmTraceTimeSkewRegression",
|
||||
],
|
||||
Platform: [
|
||||
// Azure Monitor alerts relayed through POST ops/alert-webhook (Common Alert Schema →
|
||||
|
||||
@@ -382,6 +382,7 @@ export default function AnalyzeRuleCard({
|
||||
<span className="text-gray-600 font-medium">Source:</span> {condition.source}
|
||||
{condition.eventType && (<span> | <span className="text-gray-600 font-medium">Event Type:</span> {condition.eventType}</span>)}
|
||||
{condition.dataField && (<span> | <span className="text-gray-600 font-medium">Field:</span> <code className="bg-gray-200 px-1 rounded text-xs">{condition.dataField}</code></span>)}
|
||||
{condition.skewMetric && (<span> | <span className="text-gray-600 font-medium">Skew Metric:</span> <code className="bg-gray-200 px-1 rounded text-xs">{condition.skewMetric}</code></span>)}
|
||||
</p>
|
||||
<p>
|
||||
<span className="text-gray-600 font-medium">Operator:</span> <code className="bg-gray-200 px-1 rounded text-xs">{condition.operator}</code>{" "}
|
||||
|
||||
+20
-2
@@ -2,7 +2,7 @@
|
||||
|
||||
import {
|
||||
RuleForm, RuleCondition, RulePrecondition,
|
||||
CATEGORIES, SEVERITIES, TRIGGERS, OPERATORS, SOURCES, PRECONDITION_OPERATORS,
|
||||
CATEGORIES, SEVERITIES, TRIGGERS, OPERATORS, SOURCES, PRECONDITION_OPERATORS, SKEW_METRICS,
|
||||
EMPTY_CONDITION, EMPTY_FACTOR, EMPTY_PRECONDITION,
|
||||
validateOnEventTypes,
|
||||
} from "../types";
|
||||
@@ -184,6 +184,7 @@ export default function AnalyzeRuleFormFields({ form, setForm, showRuleId, exist
|
||||
{form.conditions.map((cond, idx) => {
|
||||
const isCorrelation = cond.source === "event_correlation";
|
||||
const isArray = cond.source === "event_data_array";
|
||||
const isClockSkew = cond.source === "clock_skew";
|
||||
const updateCond = (patch: Partial<RuleCondition>) => {
|
||||
const c = [...form.conditions];
|
||||
c[idx] = { ...c[idx], ...patch };
|
||||
@@ -196,6 +197,7 @@ export default function AnalyzeRuleFormFields({ form, setForm, showRuleId, exist
|
||||
Condition {idx + 1}
|
||||
{isCorrelation && <span className="ml-2 px-1.5 py-0.5 bg-indigo-100 text-indigo-700 rounded text-xs">event_correlation</span>}
|
||||
{isArray && <span className="ml-2 px-1.5 py-0.5 bg-teal-100 text-teal-700 rounded text-xs">event_data_array</span>}
|
||||
{isClockSkew && <span className="ml-2 px-1.5 py-0.5 bg-amber-100 text-amber-700 rounded text-xs">clock_skew</span>}
|
||||
</span>
|
||||
{form.conditions.length > 1 && (
|
||||
<button type="button" onClick={() => setForm({ ...form, conditions: form.conditions.filter((_, i) => i !== idx) })} className="text-xs text-red-500 hover:text-red-700">Remove</button>
|
||||
@@ -205,7 +207,7 @@ export default function AnalyzeRuleFormFields({ form, setForm, showRuleId, exist
|
||||
{/* Row 1: Signal, Source, Event Type A */}
|
||||
<div className="grid grid-cols-1 sm:grid-cols-3 gap-2">
|
||||
<input type="text" value={cond.signal} onChange={(e) => updateCond({ signal: e.target.value })} placeholder="Signal name" autoComplete="off" className="px-3 py-1.5 border border-gray-300 rounded text-sm text-gray-900 placeholder-gray-400 focus:outline-none focus:ring-1 focus:ring-green-500" />
|
||||
<select value={cond.source} onChange={(e) => updateCond({ source: e.target.value })} className="px-3 py-1.5 border border-gray-300 rounded text-sm text-gray-900 bg-white focus:outline-none focus:ring-1 focus:ring-green-500">
|
||||
<select value={cond.source} onChange={(e) => updateCond({ source: e.target.value, ...(e.target.value === "clock_skew" && !cond.skewMetric ? { skewMetric: "clock_jump" } : {}) })} className="px-3 py-1.5 border border-gray-300 rounded text-sm text-gray-900 bg-white focus:outline-none focus:ring-1 focus:ring-green-500">
|
||||
{SOURCES.map((s) => (<option key={s} value={s}>{s}</option>))}
|
||||
</select>
|
||||
<input type="text" value={cond.eventType} onChange={(e) => updateCond({ eventType: e.target.value })} placeholder={isCorrelation ? "Event A type (e.g. app_install_completed)" : "Event type"} autoComplete="off" className="px-3 py-1.5 border border-gray-300 rounded text-sm text-gray-900 placeholder-gray-400 focus:outline-none focus:ring-1 focus:ring-green-500" />
|
||||
@@ -259,6 +261,22 @@ export default function AnalyzeRuleFormFields({ form, setForm, showRuleId, exist
|
||||
</div>
|
||||
)}
|
||||
|
||||
{/* clock_skew extra fields */}
|
||||
{isClockSkew && (
|
||||
<div className="pt-2 border-t border-amber-200 space-y-2">
|
||||
<p className="text-xs font-medium text-amber-700">Clock skew settings</p>
|
||||
<div>
|
||||
<label className="block text-xs text-gray-500 mb-1">Skew metric <span className="text-red-500">*</span></label>
|
||||
<select value={cond.skewMetric ?? "clock_jump"} onChange={(e) => updateCond({ skewMetric: e.target.value })} className="w-full px-3 py-1.5 border border-amber-300 rounded text-sm text-gray-900 bg-white focus:outline-none focus:ring-1 focus:ring-amber-500">
|
||||
{SKEW_METRICS.map((m) => (<option key={m} value={m}>{m}</option>))}
|
||||
</select>
|
||||
</div>
|
||||
<p className="text-xs text-gray-500">
|
||||
<strong>Value</strong> is the threshold in seconds; operator is limited to <code className="bg-gray-100 px-1 rounded">gt</code>/<code className="bg-gray-100 px-1 rounded">gte</code> on the skew magnitude. Event type and data field are ignored — the metric measures each event's timestamp against the server receive time across the whole session (IME-log-derived events excluded). <code className="bg-gray-100 px-1 rounded">clock_jump</code> = persistent mid-session step; <code className="bg-gray-100 px-1 rounded">sustained_offset</code> = whole session off by at least the threshold.
|
||||
</p>
|
||||
</div>
|
||||
)}
|
||||
|
||||
{/* event_data_array extra field */}
|
||||
{isArray && (
|
||||
<div className="pt-2 border-t border-teal-200 space-y-2">
|
||||
|
||||
@@ -37,6 +37,9 @@ export interface RuleCondition {
|
||||
eventAFilterField?: string;
|
||||
eventAFilterOperator?: string;
|
||||
eventAFilterValue?: string;
|
||||
// clock_skew: which device-clock metric to evaluate ("clock_jump" | "sustained_offset");
|
||||
// value is the threshold in seconds, operator is limited to gt/gte on the magnitude
|
||||
skewMetric?: string;
|
||||
}
|
||||
|
||||
export interface RulePrecondition {
|
||||
@@ -126,7 +129,8 @@ export const CATEGORIES = ["network", "identity", "apps", "device", "esp", "enro
|
||||
export const SEVERITIES = ["info", "warning", "high", "critical"] as const;
|
||||
export const TRIGGERS = ["single", "correlation"] as const;
|
||||
export const OPERATORS = ["equals", "not_equals", "contains", "not_contains", "regex", "not_regex", "gt", "lt", "gte", "lte", "exists", "not_exists", "count_gte", "count_per_group_gte", "in", "not_in"] as const;
|
||||
export const SOURCES = ["event_type", "event_data", "event_data_array", "phase_duration", "app_install_duration", "event_count", "event_correlation"] as const;
|
||||
export const SOURCES = ["event_type", "event_data", "event_data_array", "phase_duration", "app_install_duration", "event_count", "event_correlation", "clock_skew"] as const;
|
||||
export const SKEW_METRICS = ["clock_jump", "sustained_offset"] as const;
|
||||
export const PRECONDITION_OPERATORS = ["equals", "not_equals", "contains", "not_contains", "regex", "not_regex", "gt", "lt", "gte", "lte", "exists", "not_exists", "in", "not_in"] as const;
|
||||
|
||||
export const SEVERITY_COLORS: Record<string, { bg: string; text: string; border: string; dot: string }> = {
|
||||
@@ -168,6 +172,9 @@ export const EMPTY_CONDITION: RuleCondition = {
|
||||
eventAFilterField: "",
|
||||
eventAFilterOperator: "equals",
|
||||
eventAFilterValue: "",
|
||||
// skewMetric deliberately absent: it is an enum field ("clock_jump" | "sustained_offset")
|
||||
// and an empty string would violate the schema — the source select defaults it when
|
||||
// switching a condition to clock_skew.
|
||||
};
|
||||
|
||||
export const EMPTY_FACTOR: ConfidenceFactor = {
|
||||
|
||||
@@ -445,6 +445,7 @@ export const SHARED_MANIFEST = {
|
||||
],
|
||||
"analyzeRuleSources": [
|
||||
"app_install_duration",
|
||||
"clock_skew",
|
||||
"event_correlation",
|
||||
"event_count",
|
||||
"event_data",
|
||||
|
||||
@@ -441,6 +441,7 @@
|
||||
],
|
||||
"analyzeRuleSources": [
|
||||
"app_install_duration",
|
||||
"clock_skew",
|
||||
"event_correlation",
|
||||
"event_count",
|
||||
"event_data",
|
||||
|
||||
Reference in New Issue
Block a user