fix: session duration inflated by inherited tz-skewed IME log timestamps (#123)

Session df1fcf47 (WG Part 1, real ~1h16m) reported 8h15m: one do_telemetry event inherited an unrelated, 8h-shifted IME log line timestamp via LastMatchedLogTimestamp, and the backend re-anchored StartedAt on it. Three layers: - Agent: OnDoTelemetryReceived now carries the source-line timestamp; the DO-collector poll path passes null and the adapter stamps the clock instead of inheriting whatever line matched last. - Backend: StartedAt alignment and duration-start selection reject earliest-event timestamps more than 2h before the current anchor (MaxStartedAtBackwardShift), on all four mutation sites incl. the SessionsIndex dual-write decision. - Parser: CMTrace time fields with a UTC-bias suffix ("+480") now parse and honor the writer-declared offset instead of failing to match. Co-authored-by: Claude Fable 5 <noreply@anthropic.com>
2026-10-24 04:32:32 +00:00 · 2026-08-04 11:54:41 +02:00
parent efc96258d6
commit 9ba67162aa
10 changed files with 375 additions and 51 deletions
@@ -92,6 +92,27 @@ namespace AutopilotMonitor.Agent.V2.Core.Tests.Monitoring.Ime
            Assert.InRange(entry.Timestamp, before.AddSeconds(-5), after.AddSeconds(5));
        }

+        // ── UTC-bias suffix on the time field → honored, not dropped ──────────
+
+        [Theory]
+        [InlineData("+480", 8 * 60)]   // PST writer (UTC-8): UTC = local + 480min
+        [InlineData("-060", -60)]      // UTC+1 writer: UTC = local - 60min
+        public void TryParseLine_applies_utc_bias_suffix(string bias, int expectedOffsetMinutes)
+        {
+            // Session df1fcf47: a writer whose timezone differs from the agent's stamped its
+            // lines with an explicit bias. Without bias handling the line either failed to
+            // match at all (old regex) or would be misread in agent-local time.
+            var line = CmTraceLine("msg", "11:46:19.0226610" + bias, "7-29-2026");
+
+            var ok = CmTraceLogParser.TryParseLine(line, out var entry);
+
+            Assert.True(ok);
+            Assert.Equal(
+                new DateTime(2026, 7, 29, 11, 46, 19, DateTimeKind.Utc).AddTicks(226610).AddMinutes(expectedOffsetMinutes),
+                entry.Timestamp);
+            Assert.Equal(DateTimeKind.Utc, entry.Timestamp.Kind);
+        }
+
        // ── Oversized line → parses, no throw, no truncation of the message ───

        [Fact]
@@ -83,13 +83,36 @@ namespace AutopilotMonitor.Agent.V2.Core.Tests.SignalAdapters
            using var adapter = new ImeLogTrackerAdapter(f.Tracker, f.Ingress, f.Clock);
            var app = new AppPackageState("app-do", 0);

-            f.Tracker.LastMatchedLogTimestamp = ClockNow.AddDays(-7);
-            adapter.TriggerDoTelemetryFromTest(app);
+            // Line-grounded path: the [DO TEL] line's own timestamp travels with the callback.
+            adapter.TriggerDoTelemetryFromTest(app, ClockNow.AddDays(-7));
            Assert.Empty(f.InfoEvents(SharedEventTypes.DoTelemetry));

-            f.Tracker.LastMatchedLogTimestamp = ClockNow.AddMinutes(-1);
-            adapter.TriggerDoTelemetryFromTest(app);
-            Assert.Single(f.InfoEvents(SharedEventTypes.DoTelemetry));
+            adapter.TriggerDoTelemetryFromTest(app, ClockNow.AddMinutes(-1));
+            var info = Assert.Single(f.InfoEvents(SharedEventTypes.DoTelemetry));
+            Assert.Equal(ClockNow.AddMinutes(-1), info.OccurredAtUtc);
+        }
+
+        [Fact]
+        public void Poll_driven_do_telemetry_stamps_clock_ignoring_unrelated_last_matched_line()
+        {
+            // Session df1fcf47: the DO collector observed the completion live, but the event
+            // inherited LastMatchedLogTimestamp from an unrelated, tz-skewed log line 8h in
+            // the past — which dragged Sessions.StartedAt back and inflated the duration.
+            // Poll-driven telemetry (null source timestamp) must stamp the clock, never the
+            // last matched line.
+            using var f = new ImeLogTrackerAdapterFixture(ClockNow);
+            using var adapter = new ImeLogTrackerAdapter(f.Tracker, f.Ingress, f.Clock);
+            var app = new AppPackageState("app-do-poll", 0);
+
+            f.Tracker.LastMatchedLogTimestamp = ClockNow.AddHours(-8);
+            adapter.TriggerDoTelemetryFromTest(app, sourceTimestampUtc: null);
+
+            var info = Assert.Single(f.InfoEvents(SharedEventTypes.DoTelemetry));
+            Assert.Equal(ClockNow, info.OccurredAtUtc);
+            Assert.Equal("true", info.Payload!["derivedTimestamp"]);
+            // Not a replay — live observation must never trip the replay suppression,
+            // even with a stale unrelated line on the tracker.
+            Assert.Empty(f.InfoEvents(SharedEventTypes.HistoricImeReplayDetected));
        }

        [Fact]
@@ -647,13 +647,13 @@ namespace AutopilotMonitor.Agent.V2.Core.Tests.SignalAdapters
            int priorDoCalls = 0;
            int priorScriptCalls = 0;
            f.Tracker.OnImeAgentVersion = _ => priorVersionCalls++;
-            f.Tracker.OnDoTelemetryReceived = _ => priorDoCalls++;
+            f.Tracker.OnDoTelemetryReceived = (_, _) => priorDoCalls++;
            f.Tracker.OnScriptCompleted = _ => priorScriptCalls++;

            using var adapter = new ImeLogTrackerAdapter(f.Tracker, f.Ingress, f.Clock);

            f.Tracker.OnImeAgentVersion("1.0.0.0");
-            f.Tracker.OnDoTelemetryReceived(new AppPackageState("app-1", 0));
+            f.Tracker.OnDoTelemetryReceived(new AppPackageState("app-1", 0), null);
            f.Tracker.OnScriptCompleted(new ScriptExecutionState { PolicyId = "p", ScriptType = "platform" });

            Assert.Equal(1, priorVersionCalls);
@@ -667,7 +667,7 @@ namespace AutopilotMonitor.Agent.V2.Core.Tests.SignalAdapters
        {
            using var f = new ImeLogTrackerAdapterFixture();
            Action<string> priorVersion = _ => { };
-            Action<AppPackageState> priorDo = _ => { };
+            Action<AppPackageState, DateTime?> priorDo = (_, _) => { };
            Action<ScriptExecutionState> priorScript = _ => { };
            f.Tracker.OnImeAgentVersion = priorVersion;
            f.Tracker.OnDoTelemetryReceived = priorDo;
@@ -22,9 +22,12 @@ namespace AutopilotMonitor.Agent.V2.Core.Monitoring.Enrollment.Ime
    /// </summary>
    public static class CmTraceLogParser
    {
-        // Pre-compiled regex for the CMTrace format
+        // Pre-compiled regex for the CMTrace format. The time field may carry a UTC-bias
+        // suffix in minutes ("06:08:04.8834397+480", GetTimeZoneInformation convention:
+        // UTC = local + bias) — without the optional bias group such lines would not match
+        // at all and their content would be invisible to every pattern.
        private static readonly Regex CmTraceRegex = new Regex(
-            @"<!\[LOG\[(?<message>.*)\]LOG\]!><time=""(?<time>[\d:.]+)""\s+date=""(?<date>[\d-]+)""\s+component=""(?<component>[^""]*)""\s+context=""[^""]*""\s+type=""(?<type>\d+)""\s+thread=""(?<thread>\d+)""\s+file=""[^""]*"">",
+            @"<!\[LOG\[(?<message>.*)\]LOG\]!><time=""(?<time>[\d:.]+)(?<bias>[+-]\d{1,4})?""\s+date=""(?<date>[\d-]+)""\s+component=""(?<component>[^""]*)""\s+context=""[^""]*""\s+type=""(?<type>\d+)""\s+thread=""(?<thread>\d+)""\s+file=""[^""]*"">",
            RegexOptions.Compiled | RegexOptions.Singleline
        );

@@ -45,6 +48,7 @@ namespace AutopilotMonitor.Agent.V2.Core.Monitoring.Enrollment.Ime

            var message = match.Groups["message"].Value;
            var timeStr = match.Groups["time"].Value;
+            var biasStr = match.Groups["bias"].Value;
            var dateStr = match.Groups["date"].Value;
            var component = match.Groups["component"].Value;
            var typeStr = match.Groups["type"].Value;
@@ -52,7 +56,7 @@ namespace AutopilotMonitor.Agent.V2.Core.Monitoring.Enrollment.Ime

            // Parse timestamp: date is "M-d-yyyy", time is "HH:mm:ss.ticks"
            DateTime timestamp;
-            if (!TryParseTimestamp(dateStr, timeStr, out timestamp))
+            if (!TryParseTimestamp(dateStr, timeStr, biasStr, out timestamp))
            {
                timestamp = DateTime.UtcNow;
            }
@@ -72,7 +76,7 @@ namespace AutopilotMonitor.Agent.V2.Core.Monitoring.Enrollment.Ime
            return true;
        }

-        private static bool TryParseTimestamp(string dateStr, string timeStr, out DateTime result)
+        private static bool TryParseTimestamp(string dateStr, string timeStr, string biasStr, out DateTime result)
        {
            result = DateTime.MinValue;

@@ -116,7 +120,23 @@ namespace AutopilotMonitor.Agent.V2.Core.Monitoring.Enrollment.Ime
                "M-d-yyyy HH:mm:ss"
            };

-            // CMTrace timestamps are always in LOCAL time. Parse as local, then convert to UTC.
+            // Bias-suffixed time ("+480"): the writer declared its own UTC offset in minutes
+            // (GetTimeZoneInformation convention: UTC = written time + bias). Honor it instead
+            // of assuming the agent's local timezone — writer and agent can disagree (mixed
+            // components, mid-enrollment timezone change; session df1fcf47).
+            if (!string.IsNullOrEmpty(biasStr)
+                && int.TryParse(biasStr, NumberStyles.AllowLeadingSign, CultureInfo.InvariantCulture, out var biasMinutes))
+            {
+                if (DateTime.TryParseExact(combined, formats, CultureInfo.InvariantCulture,
+                    DateTimeStyles.None, out result))
+                {
+                    result = DateTime.SpecifyKind(result.AddMinutes(biasMinutes), DateTimeKind.Utc);
+                    return true;
+                }
+                return false;
+            }
+
+            // No bias: CMTrace timestamps are in LOCAL time. Parse as local, then convert to UTC.
            if (DateTime.TryParseExact(combined, formats, CultureInfo.InvariantCulture,
                DateTimeStyles.AssumeLocal, out result))
            {
@@ -112,7 +112,7 @@ namespace AutopilotMonitor.Agent.V2.Core.Monitoring.Enrollment.Ime
                    $"size={fileSize}, peers={bytesFromPeers} ({peerCachingPct}%), " +
                    $"http={bytesFromHttp}, cache={bytesFromCacheServer}, mode={downloadMode}, duration={downloadDuration}");

-                OnDoTelemetryReceived?.Invoke(pkg);
+                OnDoTelemetryReceived?.Invoke(pkg, LastMatchedLogTimestamp);
                _stateDirty = true;
            }
            catch (Exception ex)
@@ -194,7 +194,11 @@ namespace AutopilotMonitor.Agent.V2.Core.Monitoring.Enrollment.Ime
        public Action OnAllAppsCompleted { get; set; }
        public Action OnUserSessionCompleted { get; set; }
        public Action<string> OnImeSessionChange { get; set; }
-        public Action<AppPackageState> OnDoTelemetryReceived { get; set; }
+        // Second arg: the [DO TEL] source log line's timestamp when the DO data came from a
+        // log line, null when the DO collector observed the completion live via polling. The
+        // consumer must stamp poll-driven telemetry with the clock — inheriting an unrelated
+        // line's LastMatchedLogTimestamp put a tz-skewed time on the event (session df1fcf47).
+        public Action<AppPackageState, DateTime?> OnDoTelemetryReceived { get; set; }
        public Action<ScriptExecutionState> OnScriptCompleted { get; set; }

        /// <summary>
@@ -51,7 +51,9 @@ namespace AutopilotMonitor.Agent.V2.Core.Orchestration
                getPackageStates: () => imeHost.PackageStates,
                onDoTelemetryReceived: pkg =>
                {
-                    try { imeHost.Tracker.OnDoTelemetryReceived?.Invoke(pkg); }
+                    // null source timestamp: poll-driven completion, no log line backs it —
+                    // the adapter stamps the clock instead of LastMatchedLogTimestamp.
+                    try { imeHost.Tracker.OnDoTelemetryReceived?.Invoke(pkg, null); }
                    catch (Exception ex) { logger.Warning($"DeliveryOptimizationHost: OnDoTelemetryReceived invocation threw: {ex.Message}"); }
                },
                logDirectory: Environment.ExpandEnvironmentVariables(Constants.LogDirectory),
@@ -72,7 +72,7 @@ namespace AutopilotMonitor.Agent.V2.Core.SignalAdapters
        private readonly Action<AppPackageState, AppInstallationState, AppInstallationState>? _prevOnAppStateChanged;
        private readonly Action<string>? _prevOnPatternMatched;
        private readonly Action<string>? _prevOnImeAgentVersion;
-        private readonly Action<AppPackageState>? _prevOnDoTelemetryReceived;
+        private readonly Action<AppPackageState, DateTime?>? _prevOnDoTelemetryReceived;
        private readonly Action<ScriptExecutionState>? _prevOnScriptCompleted;
        private readonly Action<ScriptStartedInfo>? _prevOnScriptStarted;

@@ -82,7 +82,7 @@ namespace AutopilotMonitor.Agent.V2.Core.SignalAdapters
        private readonly Action<AppPackageState, AppInstallationState, AppInstallationState> _ourOnAppStateChanged;
        private readonly Action<string> _ourOnPatternMatched;
        private readonly Action<string> _ourOnImeAgentVersion;
-        private readonly Action<AppPackageState> _ourOnDoTelemetryReceived;
+        private readonly Action<AppPackageState, DateTime?> _ourOnDoTelemetryReceived;
        private readonly Action<ScriptExecutionState> _ourOnScriptCompleted;
        private readonly Action<ScriptStartedInfo> _ourOnScriptStarted;

@@ -242,10 +242,10 @@ namespace AutopilotMonitor.Agent.V2.Core.SignalAdapters
            EmitImeAgentVersion(version);
        }

-        private void OnDoTelemetryReceived(AppPackageState app)
+        private void OnDoTelemetryReceived(AppPackageState app, DateTime? sourceTimestampUtc)
        {
-            _prevOnDoTelemetryReceived?.Invoke(app);
-            EmitDoTelemetry(app);
+            _prevOnDoTelemetryReceived?.Invoke(app, sourceTimestampUtc);
+            EmitDoTelemetry(app, sourceTimestampUtc);
        }

        private void OnScriptCompleted(ScriptExecutionState script)
@@ -266,7 +266,8 @@ namespace AutopilotMonitor.Agent.V2.Core.SignalAdapters
            EmitAppState(app, oldState, newState);
        internal void TriggerPatternMatchedFromTest(string patternId) => MaybeEmitWhiteGloveSealingPattern(patternId);
        internal void TriggerImeAgentVersionFromTest(string version) => EmitImeAgentVersion(version);
-        internal void TriggerDoTelemetryFromTest(AppPackageState app) => EmitDoTelemetry(app);
+        internal void TriggerDoTelemetryFromTest(AppPackageState app, DateTime? sourceTimestampUtc = null) =>
+            EmitDoTelemetry(app, sourceTimestampUtc);
        internal void TriggerScriptCompletedFromTest(ScriptExecutionState script) => EmitScriptCompleted(script);
        internal void TriggerScriptStartedFromTest(ScriptStartedInfo info) => EmitScriptStarted(info);

@@ -876,7 +877,7 @@ namespace AutopilotMonitor.Agent.V2.Core.SignalAdapters
            _logger?.Info($"ImeAdapter: IME agent version detected: {version}");
        }

-        private void EmitDoTelemetry(AppPackageState app)
+        private void EmitDoTelemetry(AppPackageState app, DateTime? sourceTimestampUtc)
        {
            if (app == null || string.IsNullOrEmpty(app.Id)) return;

@@ -912,11 +913,31 @@ namespace AutopilotMonitor.Agent.V2.Core.SignalAdapters
            var patternId = _tracker.LastMatchedPatternId;
            if (!string.IsNullOrEmpty(patternId)) data["patternId"] = patternId!;

-            var now = ResolveOccurredAt(out var derivedFromClock, out var rawSourceTs);
+            // Two trigger paths with different time grounding (session df1fcf47: a single
+            // tz-skewed unrelated log line inherited via LastMatchedLogTimestamp dragged the
+            // session StartedAt back 7h and inflated the reported duration):
+            //  - [DO TEL] log line (sourceTimestampUtc set): the line's own timestamp is the
+            //    event time — resolve/clamp it and apply historic-replay suppression.
+            //  - DO collector poll (sourceTimestampUtc null): the completion was observed live
+            //    via Get-DeliveryOptimizationStatus; no log line is involved, so the clock is
+            //    the only honest source. LastMatchedLogTimestamp must NOT be consulted here.
+            DateTime now;
+            bool derivedFromClock;
+            DateTime? rawSourceTs;
+            if (sourceTimestampUtc.HasValue)
+            {
+                now = ResolveOccurredAt(sourceTimestampUtc, out derivedFromClock, out rawSourceTs);

-            // Historic replay — a week-old DO summary is not this enrollment's download.
-            if (SuppressIfHistoricReplay(derivedFromClock, rawSourceTs, SharedEventTypes.DoTelemetry, app.Id))
-                return;
+                // Historic replay — a week-old DO summary is not this enrollment's download.
+                if (SuppressIfHistoricReplay(derivedFromClock, rawSourceTs, SharedEventTypes.DoTelemetry, app.Id))
+                    return;
+            }
+            else
+            {
+                now = _clock.UtcNow;
+                derivedFromClock = true;
+                rawSourceTs = null;
+            }

            TagDerivedTimestamp(data, derivedFromClock, rawSourceTs);

@@ -0,0 +1,177 @@
+using System;
+using System.Collections.Generic;
+using System.Threading;
+using System.Threading.Tasks;
+using Azure;
+using Azure.Data.Tables;
+using AutopilotMonitor.Functions.Services;
+using AutopilotMonitor.Shared;
+using AutopilotMonitor.Shared.Models;
+using Microsoft.Extensions.Logging.Abstractions;
+using Moq;
+using Xunit;
+
+namespace AutopilotMonitor.Functions.Tests;
+
+/// <summary>
+/// Plausibility guard for the session anchor (session df1fcf47): ONE do_telemetry event whose
+/// OccurredUtc inherited a tz-shifted IME log line 8h in the past dragged StartedAt back ~7h
+/// and inflated the WG Part 1 DurationSeconds from ~1h16m to 8h15m. The earliest-event
+/// timestamp — whether from the ingest batch or the Events-table probe — may only re-anchor
+/// StartedAt / the duration start when it sits within
+/// <see cref="TableStorageService.MaxStartedAtBackwardShift"/> of the current anchor.
+/// </summary>
+public class SessionStartedAtBackwardShiftGuardTests
+{
+    private const string TenantId  = "11111111-1111-1111-1111-111111111111";
+    private const string SessionId = "22222222-2222-2222-2222-222222222222";
+
+    private static readonly DateTime StartedAt = new(2026, 7, 29, 18, 45, 15, DateTimeKind.Utc);
+
+    // ==================================================== StartedAt alignment ====
+
+    [Fact]
+    public async Task IncrementEventCount_rejects_backward_shift_beyond_window()
+    {
+        var harness = new Harness(SessionRow());
+
+        // The df1fcf47 shape: one event 8h before the session's real start.
+        var skewed = StartedAt.AddHours(-7);
+        var result = await harness.Sut.IncrementSessionEventCountAsync(
+            TenantId, SessionId, increment: 1,
+            earliestEventTimestamp: skewed, latestEventTimestamp: StartedAt.AddMinutes(10));
+
+        Assert.NotNull(result);
+        Assert.NotNull(harness.Written);
+        Assert.False(harness.Written!.ContainsKey("StartedAt"));
+    }
+
+    [Fact]
+    public async Task IncrementEventCount_accepts_backward_shift_within_window()
+    {
+        var harness = new Harness(SessionRow());
+
+        // Legitimate pre-registration backlog: minutes, not hours.
+        var earlier = StartedAt.AddMinutes(-30);
+        var result = await harness.Sut.IncrementSessionEventCountAsync(
+            TenantId, SessionId, increment: 1,
+            earliestEventTimestamp: earlier, latestEventTimestamp: StartedAt.AddMinutes(10));
+
+        Assert.NotNull(result);
+        Assert.NotNull(harness.Written);
+        Assert.Equal(new DateTimeOffset(earlier), harness.Written!.GetDateTimeOffset("StartedAt"));
+    }
+
+    // ============================================= Pending (WG Part 1) duration ====
+
+    [Fact]
+    public async Task Pending_duration_ignores_skewed_stored_earliest_event()
+    {
+        // Exact df1fcf47 replica: the poisoned event is already IN the Events table
+        // (probe returns it) AND rides in the batch as earliestEventTimestamp.
+        var poisoned = StartedAt.AddHours(-7);
+        var lastEvent = new DateTime(2026, 7, 29, 20, 1, 9, DateTimeKind.Utc);
+        var harness = new Harness(SessionRow(), earliestStoredEvent: poisoned);
+
+        var ok = await harness.Sut.UpdateSessionStatusAsync(
+            TenantId, SessionId, SessionStatus.Pending,
+            earliestEventTimestamp: poisoned, latestEventTimestamp: lastEvent);
+
+        Assert.True(ok);
+        Assert.NotNull(harness.Written);
+        // Duration anchors on StartedAt (18:45:15 → 20:01:09 = 4554s), not on the skewed
+        // event (which would give 29690s less the 7h offset — the original inflation).
+        Assert.Equal((int)(lastEvent - StartedAt).TotalSeconds, harness.Written!.GetInt32("DurationSeconds"));
+        Assert.False(harness.Written.ContainsKey("StartedAt"));
+    }
+
+    [Fact]
+    public async Task Pending_duration_uses_plausible_stored_earliest_event()
+    {
+        var earliest = StartedAt.AddMinutes(-5);
+        var lastEvent = StartedAt.AddMinutes(70);
+        var harness = new Harness(SessionRow(), earliestStoredEvent: earliest);
+
+        var ok = await harness.Sut.UpdateSessionStatusAsync(
+            TenantId, SessionId, SessionStatus.Pending,
+            latestEventTimestamp: lastEvent);
+
+        Assert.True(ok);
+        Assert.NotNull(harness.Written);
+        Assert.Equal((int)(lastEvent - earliest).TotalSeconds, harness.Written!.GetInt32("DurationSeconds"));
+    }
+
+    // ============================================================ Harness ====
+
+    private static TableEntity SessionRow()
+    {
+        var entity = new TableEntity(TenantId, SessionId)
+        {
+            ["StartedAt"] = new DateTimeOffset(StartedAt),
+            ["Status"]    = "InProgress",
+            ["EventCount"] = 10,
+        };
+        entity.ETag = new ETag("0xEXISTING");
+        return entity;
+    }
+
+    /// <summary>
+    /// SDK-mock harness (same shape as <see cref="StoreSessionReregistrationPreserveTests"/>):
+    /// Sessions table configured for read + merge-capture; Events table optionally serves the
+    /// earliest-event probe; SessionsIndex stays unconfigured — its sync is swallow-all in the
+    /// SUT and irrelevant here.
+    /// </summary>
+    private sealed class Harness
+    {
+        public TableStorageService Sut { get; }
+        public TableEntity? Written { get; private set; }
+
+        public Harness(TableEntity existing, DateTime? earliestStoredEvent = null)
+        {
+            var sessions = new Mock<TableClient>();
+
+            sessions.Setup(t => t.GetEntityAsync<TableEntity>(
+                    TenantId, SessionId, It.IsAny<IEnumerable<string>>(), It.IsAny<CancellationToken>()))
+                .ReturnsAsync(Response.FromValue(existing, new Mock<Response>().Object));
+
+            var ifExists = new Mock<NullableResponse<TableEntity>>();
+            ifExists.SetupGet(r => r.HasValue).Returns(true);
+            ifExists.SetupGet(r => r.Value).Returns(existing);
+            sessions.Setup(t => t.GetEntityIfExistsAsync<TableEntity>(
+                    TenantId, SessionId, It.IsAny<IEnumerable<string>>(), It.IsAny<CancellationToken>()))
+                .ReturnsAsync(ifExists.Object);
+
+            sessions.Setup(t => t.UpdateEntityAsync(
+                    It.IsAny<TableEntity>(), It.IsAny<ETag>(), It.IsAny<TableUpdateMode>(), It.IsAny<CancellationToken>()))
+                .Returns<TableEntity, ETag, TableUpdateMode, CancellationToken>((e, _, _, _) =>
+                {
+                    Written = e;
+                    return Task.FromResult(new Mock<Response>().Object);
+                });
+
+            var events = new Mock<TableClient>();
+            var eventRows = earliestStoredEvent.HasValue
+                ? new[]
+                {
+                    new TableEntity($"{TenantId}_{SessionId}", $"{earliestStoredEvent.Value:yyyyMMddHHmmssfff}_0000000001")
+                    {
+                        ["OccurredUtc"] = new DateTimeOffset(earliestStoredEvent.Value),
+                    },
+                }
+                : Array.Empty<TableEntity>();
+            events.Setup(t => t.QueryAsync<TableEntity>(
+                    It.IsAny<string>(), It.IsAny<int?>(), It.IsAny<IEnumerable<string>>(), It.IsAny<CancellationToken>()))
+                .Returns(AsyncPageable<TableEntity>.FromPages(new[]
+                {
+                    Page<TableEntity>.FromValues(eventRows, null, new Mock<Response>().Object),
+                }));
+
+            var serviceClient = new Mock<TableServiceClient>();
+            serviceClient.Setup(s => s.GetTableClient(Constants.TableNames.Sessions)).Returns(sessions.Object);
+            serviceClient.Setup(s => s.GetTableClient(Constants.TableNames.Events)).Returns(events.Object);
+            // SessionsIndex intentionally unconfigured → null client → SUT swallows.
+
+            Sut = new TableStorageService(serviceClient.Object, NullLogger<TableStorageService>.Instance);
+        }
+    }
+}
@@ -271,6 +271,45 @@ namespace AutopilotMonitor.Functions.Services
            }
        }

+        /// <summary>
+        /// Upper bound for how far the earliest-event timestamp may sit BEFORE the session's
+        /// current StartedAt and still be trusted as the new session anchor. Legitimate cases
+        /// (events ingested before registration, agent bootstrapping mid-ESP and replaying
+        /// accumulated IME log lines) are minutes to at most ~an hour. Anything beyond is a
+        /// skewed source timestamp — session df1fcf47: ONE do_telemetry event whose OccurredUtc
+        /// inherited a tz-shifted IME log line 8h in the past re-anchored StartedAt and inflated
+        /// DurationSeconds from ~1h16m to 8h15m. The agent accepts source timestamps up to 24h
+        /// old by design (replay timeline fidelity), so the session anchor needs its own,
+        /// much tighter plausibility window.
+        /// </summary>
+        internal static readonly TimeSpan MaxStartedAtBackwardShift = TimeSpan.FromHours(2);
+
+        /// <summary>
+        /// Returns <paramref name="earliestEventTimestamp"/> when it is a plausible session
+        /// anchor (missing anchor, forward-in-time, or backward within
+        /// <see cref="MaxStartedAtBackwardShift"/>), null when the backward shift is implausible.
+        /// Mutation sites MUST apply this before BOTH the StartedAt alignment and the
+        /// SyncSessionIndexAsync call — the index derives its shift-vs-merge decision from the
+        /// same comparison, so feeding it the unsanitized value would move the index RowKey
+        /// while the session row keeps its anchor. The rejected event itself stays in the
+        /// timeline; only the anchor refuses to follow it.
+        /// </summary>
+        private DateTime? SanitizeEarliestEventTimestamp(
+            DateTime? earliestEventTimestamp, DateTime currentStartedAt, string sessionId)
+        {
+            if (!earliestEventTimestamp.HasValue || currentStartedAt == DateTime.MaxValue)
+                return earliestEventTimestamp;
+
+            var backwardShift = currentStartedAt - earliestEventTimestamp.Value;
+            if (backwardShift <= MaxStartedAtBackwardShift)
+                return earliestEventTimestamp;
+
+            _logger.LogWarning(
+                "Session {SessionId}: earliest event timestamp {EarliestEventTimestamp:o} is {ShiftHours:F1}h before StartedAt {StartedAt:o} — backward shift rejected as skewed source timestamp",
+                sessionId, earliestEventTimestamp.Value, backwardShift.TotalHours, currentStartedAt);
+            return null;
+        }
+
        /// <summary>
        /// Centralizes the SessionsIndex dual-write decision shared by the session-mutation paths
        /// (UpdateSessionStatusAsync normal + ETag force-write, IncrementSessionEventCountAsync): when
@@ -483,8 +522,11 @@ namespace AutopilotMonitor.Functions.Services
                }

                // If events were ingested before session registration succeeded, align StartedAt
-                // with the earliest event we already have for this session.
-                var earliestEventTimestamp = await GetEarliestSessionEventTimestampAsync(registration.TenantId, registration.SessionId);
+                // with the earliest event we already have for this session (plausibility-guarded:
+                // a skewed source timestamp must not pre-poison the anchor at registration).
+                var earliestEventTimestamp = SanitizeEarliestEventTimestamp(
+                    await GetEarliestSessionEventTimestampAsync(registration.TenantId, registration.SessionId),
+                    startedAt, registration.SessionId);
                if (earliestEventTimestamp.HasValue && earliestEventTimestamp.Value < startedAt)
                {
                    startedAt = earliestEventTimestamp.Value;
@@ -1655,10 +1697,12 @@ namespace AutopilotMonitor.Functions.Services
                    }

                    // Align StartedAt with the earliest event timestamp provided by the caller
+                    // (plausibility-guarded — one skewed event must not re-anchor the session).
                    var currentStartedAt = session.GetDateTimeOffset("StartedAt")?.UtcDateTime ?? DateTime.MaxValue;
-                    if (earliestEventTimestamp.HasValue && earliestEventTimestamp.Value < currentStartedAt)
+                    var safeEarliestEventTimestamp = SanitizeEarliestEventTimestamp(earliestEventTimestamp, currentStartedAt, sessionId);
+                    if (safeEarliestEventTimestamp.HasValue && safeEarliestEventTimestamp.Value < currentStartedAt)
                    {
-                        update["StartedAt"] = EnsureUtc(earliestEventTimestamp.Value);
+                        update["StartedAt"] = EnsureUtc(safeEarliestEventTimestamp.Value);
                    }

                    // Set completion time if succeeded or failed
@@ -1688,11 +1732,14 @@ namespace AutopilotMonitor.Functions.Services
                            // Standard session (or WhiteGlove without stored Part 1 data — fallback):
                            // Read earliest event from Events table — authoritative source, immune to
                            // concurrent StartedAt update races. This is a single-row lookup (maxPerPage: 1)
-                            // and only happens once per session lifecycle (at completion).
-                            var earliestStoredEvent = await GetEarliestSessionEventTimestampAsync(tenantId, sessionId);
+                            // and only happens once per session lifecycle (at completion). Same
+                            // plausibility guard as the anchor: one skewed stored event must not
+                            // stretch the duration by hours.
+                            var earliestStoredEvent = SanitizeEarliestEventTimestamp(
+                                await GetEarliestSessionEventTimestampAsync(tenantId, sessionId), currentStartedAt, sessionId);
                            var durationStart = earliestStoredEvent ?? currentStartedAt;
-                            if (earliestEventTimestamp.HasValue && earliestEventTimestamp.Value < durationStart)
-                                durationStart = earliestEventTimestamp.Value;
+                            if (safeEarliestEventTimestamp.HasValue && safeEarliestEventTimestamp.Value < durationStart)
+                                durationStart = safeEarliestEventTimestamp.Value;

                            if (durationStart < effectiveCompletedAt)
                                update["DurationSeconds"] = (int)(effectiveCompletedAt - durationStart).TotalSeconds;
@@ -1712,10 +1759,13 @@ namespace AutopilotMonitor.Functions.Services
                    {
                        if (latestEventTimestamp.HasValue)
                        {
-                            var earliestStoredEvent = await GetEarliestSessionEventTimestampAsync(tenantId, sessionId);
+                            // Same plausibility guard as the completion path (session df1fcf47:
+                            // this exact block computed the inflated 8h15m WG Part 1 duration).
+                            var earliestStoredEvent = SanitizeEarliestEventTimestamp(
+                                await GetEarliestSessionEventTimestampAsync(tenantId, sessionId), currentStartedAt, sessionId);
                            var durationStart = earliestStoredEvent ?? currentStartedAt;
-                            if (earliestEventTimestamp.HasValue && earliestEventTimestamp.Value < durationStart)
-                                durationStart = earliestEventTimestamp.Value;
+                            if (safeEarliestEventTimestamp.HasValue && safeEarliestEventTimestamp.Value < durationStart)
+                                durationStart = safeEarliestEventTimestamp.Value;

                            if (durationStart < latestEventTimestamp.Value)
                                update["DurationSeconds"] = (int)(latestEventTimestamp.Value - durationStart).TotalSeconds;
@@ -1826,7 +1876,7 @@ namespace AutopilotMonitor.Functions.Services
                    await tableClient.UpdateEntityAsync(update, session.ETag, TableUpdateMode.Merge);

                    // Dual-write: keep SessionsIndex in sync (StartedAt-shift → full upsert, else merge).
-                    await SyncSessionIndexAsync(tenantId, sessionId, session, update, currentStartedAt, earliestEventTimestamp);
+                    await SyncSessionIndexAsync(tenantId, sessionId, session, update, currentStartedAt, safeEarliestEventTimestamp);

                    _logger.LogInformation($"Updated session {sessionId} status to {status}");
                    return true;
@@ -1919,8 +1969,10 @@ namespace AutopilotMonitor.Functions.Services
                            }

                            var freshStartedAt = freshSession.GetDateTimeOffset("StartedAt")?.UtcDateTime ?? DateTime.MaxValue;
-                            if (earliestEventTimestamp.HasValue && earliestEventTimestamp.Value < freshStartedAt)
-                                forceUpdate["StartedAt"] = EnsureUtc(earliestEventTimestamp.Value);
+                            // Re-sanitize against the FRESH anchor (mirror of the normal path's guard).
+                            var safeForceEarliestEventTimestamp = SanitizeEarliestEventTimestamp(earliestEventTimestamp, freshStartedAt, sessionId);
+                            if (safeForceEarliestEventTimestamp.HasValue && safeForceEarliestEventTimestamp.Value < freshStartedAt)
+                                forceUpdate["StartedAt"] = EnsureUtc(safeForceEarliestEventTimestamp.Value);

                            if (status == SessionStatus.Succeeded || status == SessionStatus.Failed)
                            {
@@ -1943,10 +1995,11 @@ namespace AutopilotMonitor.Functions.Services
                                }
                                else
                                {
-                                    var earliestStoredEvent = await GetEarliestSessionEventTimestampAsync(tenantId, sessionId);
+                                    var earliestStoredEvent = SanitizeEarliestEventTimestamp(
+                                        await GetEarliestSessionEventTimestampAsync(tenantId, sessionId), freshStartedAt, sessionId);
                                    var durationStart = earliestStoredEvent ?? freshStartedAt;
-                                    if (earliestEventTimestamp.HasValue && earliestEventTimestamp.Value < durationStart)
-                                        durationStart = earliestEventTimestamp.Value;
+                                    if (safeForceEarliestEventTimestamp.HasValue && safeForceEarliestEventTimestamp.Value < durationStart)
+                                        durationStart = safeForceEarliestEventTimestamp.Value;

                                    if (durationStart < effectiveCompletedAt)
                                        forceUpdate["DurationSeconds"] = (int)(effectiveCompletedAt - durationStart).TotalSeconds;
@@ -1956,10 +2009,11 @@ namespace AutopilotMonitor.Functions.Services
                            {
                                if (latestEventTimestamp.HasValue)
                                {
-                                    var earliestStoredEvent = await GetEarliestSessionEventTimestampAsync(tenantId, sessionId);
+                                    var earliestStoredEvent = SanitizeEarliestEventTimestamp(
+                                        await GetEarliestSessionEventTimestampAsync(tenantId, sessionId), freshStartedAt, sessionId);
                                    var durationStart = earliestStoredEvent ?? freshStartedAt;
-                                    if (earliestEventTimestamp.HasValue && earliestEventTimestamp.Value < durationStart)
-                                        durationStart = earliestEventTimestamp.Value;
+                                    if (safeForceEarliestEventTimestamp.HasValue && safeForceEarliestEventTimestamp.Value < durationStart)
+                                        durationStart = safeForceEarliestEventTimestamp.Value;

                                    if (durationStart < latestEventTimestamp.Value)
                                        forceUpdate["DurationSeconds"] = (int)(latestEventTimestamp.Value - durationStart).TotalSeconds;
@@ -2026,7 +2080,7 @@ namespace AutopilotMonitor.Functions.Services
                            await forceTableClient.UpdateEntityAsync(forceUpdate, ETag.All, TableUpdateMode.Merge);

                            // Dual-write: keep SessionsIndex in sync (StartedAt-shift → full upsert, else merge).
-                            await SyncSessionIndexAsync(tenantId, sessionId, freshSession, forceUpdate, freshStartedAt, earliestEventTimestamp);
+                            await SyncSessionIndexAsync(tenantId, sessionId, freshSession, forceUpdate, freshStartedAt, safeForceEarliestEventTimestamp);

                            _logger.LogInformation($"Force-updated session {sessionId} status to {status} (unconditional merge after ETag exhaustion)");
                            return true;
@@ -2119,10 +2173,12 @@ namespace AutopilotMonitor.Functions.Services
                    }

                    // Align StartedAt with the earliest event timestamp provided by the caller
+                    // (plausibility-guarded — one skewed event must not re-anchor the session).
                    var currentStartedAt = entity.GetDateTimeOffset("StartedAt")?.UtcDateTime ?? DateTime.MaxValue;
-                    if (earliestEventTimestamp.HasValue && earliestEventTimestamp.Value < currentStartedAt)
+                    var safeEarliestEventTimestamp = SanitizeEarliestEventTimestamp(earliestEventTimestamp, currentStartedAt, sessionId);
+                    if (safeEarliestEventTimestamp.HasValue && safeEarliestEventTimestamp.Value < currentStartedAt)
                    {
-                        update["StartedAt"] = EnsureUtc(earliestEventTimestamp.Value);
+                        update["StartedAt"] = EnsureUtc(safeEarliestEventTimestamp.Value);
                    }

                    // Track the most recent event timestamp for excessive data sender detection
@@ -2159,7 +2215,7 @@ namespace AutopilotMonitor.Functions.Services
                    await tableClient.UpdateEntityAsync(update, entity.ETag, TableUpdateMode.Merge);

                    // Dual-write: keep SessionsIndex in sync (StartedAt-shift → full upsert, else merge).
-                    await SyncSessionIndexAsync(tenantId, sessionId, entity, update, currentStartedAt, earliestEventTimestamp);
+                    await SyncSessionIndexAsync(tenantId, sessionId, entity, update, currentStartedAt, safeEarliestEventTimestamp);

                    // Apply the merged fields onto the RMW read and map it through the shared
                    // mapper — the post-merge snapshot the caller would otherwise re-read.