Read the Sessions row projected on the agent hot path

Backend 1.5.1256 raised IngestTelemetry p50 from 66 to ~100 ms and RegisterSession from 90 to ~110 ms. Storage metrics per API located it: server latency of point reads unchanged, end-to-end latency of QueryEntity alone up from 3.7 to 6.4 ms, every other API flat. The excess times the point reads per upload is the whole regression, and the request histogram gained a second mode 46 ms above the first. The only point-read response that had changed was the Sessions row (~400 bytes of tracker state and twins), which every upload read twice in full. - SessionRowProjections: the guard row for the agent routes (Status, SerialNumber, owner columns) and the counter merge's slice (counters, anchor, index key, tracker state and twins, the fields non-terminal batches consume). The guard takes a select and adds its lock columns; the inventory reader gets a projected overload, the deletion flows keep their full reads. - The counter merge returns SessionIngestSnapshot, a narrow type instead of a SessionSummary built from a projected row. The processor reads the whole row only on terminal, diagnostics and transition batches or when the merge returned nothing, and derives the snapshot there. The SignalR delta and the control signals carry the same fields as before. - The signal-only upload reads the same slice; registration takes its lifecycle entry from the guard's row instead of a second read. - SessionRowProjectionTests: a select-strict harness pins each list. The merge's write, index mirror and snapshot, the slice mapper, the guard row's readers and the owner policy come out identical whole or projected; a dropped column fails there (verified by mutation). Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
2026-10-24 04:32:32 +00:00 · 2026-10-07 15:33:37 +02:00
parent aaf9b74426
commit 50bb09eb92
18 changed files with 872 additions and 102 deletions
@@ -372,6 +372,9 @@ public class CascadeVerificationServiceTests
        public Task<TableEntity?> GetSessionRowAsync(string tenantId, string sessionId, CancellationToken ct = default)
            => Task.FromResult<TableEntity?>(null);

+        public Task<TableEntity?> GetSessionRowAsync(string tenantId, string sessionId, IEnumerable<string> select, CancellationToken ct = default)
+            => Task.FromResult<TableEntity?>(null);
+
        public Task<TableEntity?> GetSessionsIndexRowAsync(string tenantId, string indexRowKey, CancellationToken ct = default)
            => Task.FromResult<TableEntity?>(null);

@@ -1,4 +1,4 @@
-using AutopilotMonitor.Functions.Functions.Ingest;
+using AutopilotMonitor.Functions.Helpers;
 using AutopilotMonitor.Shared.Models;
 using Azure.Data.Tables;
 using Xunit;
@@ -26,14 +26,14 @@ public class IngestStatusPrefetchTests
    [InlineData("succeeded", SessionStatus.Succeeded)] // mapper parses case-insensitively — mirror it
    public void StringStatus_TheActualWireFormat_IsParsed(string stored, SessionStatus expected)
    {
-        Assert.Equal(expected, IngestTelemetryFunction.TryReadSessionStatus(Row(stored)));
+        Assert.Equal(expected, SessionRowProjections.TryReadStatus(Row(stored)));
    }

    [Fact]
    public void IntStatus_LegacyDefensiveFallback_IsCast()
    {
        var defined = (int)SessionStatus.Stalled;
-        Assert.Equal(SessionStatus.Stalled, IngestTelemetryFunction.TryReadSessionStatus(Row(defined)));
+        Assert.Equal(SessionStatus.Stalled, SessionRowProjections.TryReadStatus(Row(defined)));
    }

    [Theory]
@@ -41,13 +41,13 @@ public class IngestStatusPrefetchTests
    [InlineData("")]
    public void UnparseableStatus_ReturnsNull_SoCallerFallsBackToOwnRead(string stored)
    {
-        Assert.Null(IngestTelemetryFunction.TryReadSessionStatus(Row(stored)));
+        Assert.Null(SessionRowProjections.TryReadStatus(Row(stored)));
    }

    [Fact]
    public void MissingStatusOrRow_ReturnsNull()
    {
-        Assert.Null(IngestTelemetryFunction.TryReadSessionStatus(Row(status: null)));
-        Assert.Null(IngestTelemetryFunction.TryReadSessionStatus(null));
+        Assert.Null(SessionRowProjections.TryReadStatus(Row(status: null)));
+        Assert.Null(SessionRowProjections.TryReadStatus(null));
    }
 }
@@ -2,6 +2,7 @@ using System.Threading;
 using System.Threading.Tasks;
 using Azure.Data.Tables;
 using AutopilotMonitor.Functions.Functions.Ingest;
+using AutopilotMonitor.Functions.Helpers;
 using AutopilotMonitor.Functions.Services.Deletion;
 using AutopilotMonitor.Shared.Models;
 using AutopilotMonitor.Shared.Models.Deletion;
@@ -37,7 +38,7 @@ namespace AutopilotMonitor.Functions.Tests;
 ///   <c>OrdinalIgnoreCase</c> equality Run performs (Run L169-180).</item>
 ///   <item>session deletion locked → 410: the real <see cref="SessionDeletionGuard"/> throwing
 ///   <see cref="SessionDeletionLockedException"/> (Run L205-213), plus the unique
-///   <see cref="IngestTelemetryFunction.TryReadSessionStatus"/> that consumes the guard's row.</item>
+///   <see cref="SessionRowProjections.TryReadStatus"/> that consumes the guard's row.</item>
 ///   <item>kill-switch / device-blocked → 200-block: NOT duplicated here. The verdict Run() routes
 ///   on (<c>IsBlocked</c>/<c>IsKill</c>/<c>UnblockAt</c>) is produced by
 ///   <see cref="Functions.Services.KillSwitchEvaluator"/> and is already exhaustively covered at the
@@ -138,7 +139,7 @@ public class IngestTelemetryAuthPathTests
    public async Task DeletionUnlocked_guard_returns_row_that_feeds_the_status_prefetch()
    {
        // Run L205-217: on an unlocked session the guard hands back the loaded row, and Run reads
-        // Status off it via TryReadSessionStatus to seed the stall-heal prefetch (no second read).
+        // Status off it via SessionRowProjections.TryReadStatus to seed the stall-heal prefetch (no second read).
        var guard = NewGuard(out var reader);
        var row = new TableEntity(TenantId, SessionId)
        {
@@ -151,10 +152,10 @@ public class IngestTelemetryAuthPathTests
        var returned = await guard.EnsureWritableAndGetRowAsync(TenantId, SessionId, callerContext: "V2.IngestTelemetry");

        Assert.Same(row, returned);
-        Assert.Equal(SessionStatus.Succeeded, IngestTelemetryFunction.TryReadSessionStatus(returned));
+        Assert.Equal(SessionStatus.Succeeded, SessionRowProjections.TryReadStatus(returned));
    }

-    // ============================================================ TryReadSessionStatus (unique helper)
+    // ============================================================ TryReadStatus (unique helper)

    [Theory]
    [InlineData("InProgress", SessionStatus.InProgress)]
@@ -164,27 +165,27 @@ public class IngestTelemetryAuthPathTests
    public void TryReadSessionStatus_parses_string_status_case_insensitively(string stored, SessionStatus expected)
    {
        var row = new TableEntity(TenantId, SessionId) { ["Status"] = stored };
-        Assert.Equal(expected, IngestTelemetryFunction.TryReadSessionStatus(row));
+        Assert.Equal(expected, SessionRowProjections.TryReadStatus(row));
    }

    [Fact]
    public void TryReadSessionStatus_returns_null_for_null_row()
    {
-        Assert.Null(IngestTelemetryFunction.TryReadSessionStatus(null));
+        Assert.Null(SessionRowProjections.TryReadStatus(null));
    }

    [Fact]
    public void TryReadSessionStatus_returns_null_when_status_column_absent()
    {
        var row = new TableEntity(TenantId, SessionId); // no Status column
-        Assert.Null(IngestTelemetryFunction.TryReadSessionStatus(row));
+        Assert.Null(SessionRowProjections.TryReadStatus(row));
    }

    [Fact]
    public void TryReadSessionStatus_returns_null_for_unparseable_status()
    {
        var row = new TableEntity(TenantId, SessionId) { ["Status"] = "NotARealStatus" };
-        Assert.Null(IngestTelemetryFunction.TryReadSessionStatus(row));
+        Assert.Null(SessionRowProjections.TryReadStatus(row));
    }

    // ============================================================ Harness (mirrors SessionDeletionGuardTests)
@@ -0,0 +1,535 @@
+using System;
+using System.Collections.Generic;
+using System.Linq;
+using System.Threading;
+using System.Threading.Tasks;
+using Azure;
+using Azure.Data.Tables;
+using AutopilotMonitor.Functions.Helpers;
+using AutopilotMonitor.Functions.Security;
+using AutopilotMonitor.Functions.Services;
+using AutopilotMonitor.Functions.Services.Deletion;
+using AutopilotMonitor.Shared;
+using AutopilotMonitor.Shared.DataAccess;
+using AutopilotMonitor.Shared.Models;
+using AutopilotMonitor.Shared.Models.Deletion;
+using Microsoft.Extensions.Logging.Abstractions;
+using Moq;
+using Xunit;
+
+namespace AutopilotMonitor.Functions.Tests;
+
+/// <summary>
+/// The agent hot path reads the Sessions row projected (<see cref="SessionRowProjections"/>). A
+/// select-strict harness — it hands back only the requested columns, unlike the SDK mocks of the
+/// other session tests — pins every list against its consumers: the counter merge's write, its
+/// index mirror and its snapshot, the ingest-slice mapper, the signal-only read, and the guard
+/// row's readers (status prefetch, kill-switch serial, owner policy) come out identical whether
+/// the row arrives whole or projected. A column a consumer reads but the list lacks fails here,
+/// not in production as a silent null. Fixtures carry every column a real row carries, so the
+/// projection has something to drop.
+/// </summary>
+public class SessionRowProjectionTests
+{
+    private const string TenantId = "11111111-1111-1111-1111-111111111111";
+    private const string SessionId = "22222222-2222-2222-2222-222222222222";
+
+    private static readonly TimeSpan Ahead = TimeSpan.FromHours(9);
+    private static readonly DateTime ServerStart = new(2026, 1, 1, 8, 0, 0, DateTimeKind.Utc);
+    private static readonly DateTime DeviceStart = ServerStart + Ahead;
+
+    private const string Thumb1 = "AA11BB22CC33DD44EE55FF6677889900AABBCCDD";
+    private const string Thumb2 = "0011223344556677889900AABBCCDDEEFF001122";
+    private const string Dev1 = "0f4e1c2d-1111-4aaa-8bbb-000000000001";
+
+    public enum RowKind { RunningWhiteGlovePart2, Completed }
+
+    public static IEnumerable<object[]> RowKinds => new[] { new object[] { RowKind.RunningWhiteGlovePart2 }, new object[] { RowKind.Completed } };
+
+    // ============================================================== counter merge ====
+
+    [Theory]
+    [MemberData(nameof(RowKinds))]
+    public async Task Counter_merge_write_index_mirror_and_snapshot_are_identical_whole_or_projected(RowKind kind)
+    {
+        var whole = new Harness(FullRow(kind), honourSelect: false);
+        var projected = new Harness(FullRow(kind), honourSelect: true);
+
+        var wholeSnapshot = await Increment(whole.Sut);
+        var projectedSnapshot = await Increment(projected.Sut);
+
+        // The production read passes the list, and the strict harness really dropped columns.
+        Assert.NotNull(projected.LastSelect);
+        Assert.Equal(SessionRowProjections.IngestSnapshot, projected.LastSelect!);
+        Assert.True(projected.LastReadColumnCount < whole.LastReadColumnCount,
+            $"projected read returned {projected.LastReadColumnCount} columns, whole read {whole.LastReadColumnCount}");
+
+        Assert.NotNull(whole.Written);
+        Assert.NotNull(projected.Written);
+        Assert.Equal(whole.WrittenETag, projected.WrittenETag); // the CAS ETag survives the projection
+        AssertSameEntity(whole.Written!, projected.Written!);
+        Assert.NotNull(whole.IndexWritten);
+        Assert.NotNull(projected.IndexWritten);
+        AssertSameEntity(whole.IndexWritten!, projected.IndexWritten!);
+
+        Assert.NotNull(wholeSnapshot);
+        Assert.NotNull(projectedSnapshot);
+        AssertSameSnapshot(wholeSnapshot!, projectedSnapshot!);
+    }
+
+    [Fact]
+    public async Task Counter_merge_on_the_running_row_exercises_the_tracker_and_the_counters()
+    {
+        // Guards the fixture itself: every pinned column must have a visible effect on this write,
+        // otherwise the equivalence above would pass vacuously.
+        var harness = new Harness(FullRow(RowKind.RunningWhiteGlovePart2), honourSelect: true);
+
+        var snapshot = await Increment(harness.Sut);
+
+        var written = harness.Written!;
+        Assert.Equal(2 + 3, written.GetInt32("EventCount"));
+        Assert.Equal(3 + 1, written.GetInt32("PlatformScriptCount"));
+        Assert.Equal(1 + 1, written.GetInt32("RemediationScriptCount"));
+        Assert.Equal(1 + 1, written.GetInt32("RebootCount"));
+        Assert.False(written.ContainsKey("StartedAt"), "a later earliest event must not move the anchor");
+        Assert.False(written.ContainsKey("LastEventAt"), "an older latest event must not move LastEventAt");
+        Assert.True(written.ContainsKey(TableStorageService.ServerTimeStateColumn), "the tracker continues on a running row");
+        Assert.False(written.ContainsKey("StartedAtServer"), "an unchanged twin is not rewritten");
+        Assert.Equal("2516000000000000000_" + SessionId, harness.IndexWritten!.RowKey); // the stored IndexRowKey, not a recomputed one
+        Assert.Equal(SessionStatus.InProgress, snapshot!.Status);
+        Assert.Equal(5, snapshot.EventCount);
+        Assert.Equal("Succeeded", snapshot.AdminMarkedAction);
+        Assert.Equal("2.0.1478", snapshot.AgentVersion);
+        Assert.Equal("1.99.1", snapshot.ImeAgentVersion);
+        Assert.True(snapshot.DurationSeconds >= 600, "WhiteGlove Part 2 running: stored Part 1 plus the running Part 2");
+    }
+
+    [Fact]
+    public async Task Counter_merge_on_the_completed_row_leaves_the_tracker_alone()
+    {
+        var harness = new Harness(FullRow(RowKind.Completed), honourSelect: true);
+
+        var snapshot = await Increment(harness.Sut);
+
+        Assert.False(harness.Written!.ContainsKey(TableStorageService.ServerTimeStateColumn), "CompletedAt stops the tracker");
+        Assert.Equal(SessionStatus.Succeeded, snapshot!.Status);
+        Assert.Equal(2400, snapshot.DurationSeconds);
+        Assert.Equal(new DateTime(2026, 1, 1, 17, 40, 0, DateTimeKind.Utc), snapshot.CompletedAt);
+    }
+
+    // ===================================================================== mapper ====
+
+    [Theory]
+    [MemberData(nameof(RowKinds))]
+    public void Ingest_slice_mapper_matches_the_canonical_mapper_and_needs_only_its_projection(RowKind kind)
+    {
+        var sut = new Harness(FullRow(kind), honourSelect: true).Sut;
+        var row = FullRow(kind);
+
+        var fromSlice = sut.MapToIngestSnapshot(row);
+        var fromSummary = SessionIngestSnapshot.From(sut.MapToSessionSummary(row));
+        var fromProjected = sut.MapToIngestSnapshot(Project(row, SessionRowProjections.IngestSnapshot));
+
+        AssertSameSnapshot(fromSummary, fromSlice);
+        AssertSameSnapshot(fromSlice, fromProjected);
+    }
+
+    [Theory]
+    [MemberData(nameof(RowKinds))]
+    public async Task Signal_only_snapshot_read_uses_the_same_projection_and_mapping(RowKind kind)
+    {
+        var harness = new Harness(FullRow(kind), honourSelect: true);
+
+        var snapshot = await harness.Sut.GetSessionIngestSnapshotAsync(TenantId, SessionId);
+
+        Assert.Equal(SessionRowProjections.IngestSnapshot, harness.LastSelect!);
+        Assert.NotNull(snapshot);
+        AssertSameSnapshot(harness.Sut.MapToIngestSnapshot(FullRow(kind)), snapshot!);
+    }
+
+    // ================================================================== guard row ====
+
+    [Fact]
+    public void Guard_row_readers_see_the_same_off_the_projected_row()
+    {
+        var full = FullRow(RowKind.RunningWhiteGlovePart2);
+        var projected = Project(full, SessionRowProjections.GuardRow.Concat(SessionDeletionGuard.LockColumns));
+
+        Assert.True(projected.Count < full.Count, "the guard projection must drop columns");
+        Assert.Equal(SessionRowProjections.TryReadStatus(full), SessionRowProjections.TryReadStatus(projected));
+        Assert.Equal(SessionStatus.InProgress, SessionRowProjections.TryReadStatus(projected));
+        Assert.Equal(full.GetString("SerialNumber"), projected.GetString("SerialNumber"));
+        Assert.Equal(full.GetString("DeletionState"), projected.GetString("DeletionState"));
+    }
+
+    [Theory]
+    [MemberData(nameof(OwnerScenarios))]
+    public void Owner_policy_decides_the_same_off_the_projected_row(string scenario, SessionOwner? rowOwner, SecurityValidationResult caller)
+    {
+        var full = FullRow(RowKind.RunningWhiteGlovePart2, rowOwner);
+        var projected = Project(full, SessionRowProjections.GuardRow.Concat(SessionDeletionGuard.LockColumns));
+        var now = ServerStart.AddHours(1);
+
+        var fromFull = SessionOwnershipPolicy.Evaluate(full, caller, now);
+        var fromProjected = SessionOwnershipPolicy.Evaluate(projected, caller, now);
+
+        Assert.True(fromFull.Outcome == fromProjected.Outcome, $"{scenario}: {fromFull.Outcome} vs {fromProjected.Outcome}");
+        Assert.Equal(fromFull.Rejected, fromProjected.Rejected);
+        Assert.Equal(fromFull.OwnerToStamp?.Kind, fromProjected.OwnerToStamp?.Kind);
+        Assert.Equal(fromFull.OwnerToStamp?.Thumbprint, fromProjected.OwnerToStamp?.Thumbprint);
+        Assert.Equal(fromFull.OwnerToStamp?.DeviceId, fromProjected.OwnerToStamp?.DeviceId);
+        Assert.Equal(fromFull.OwnerToStamp?.BootstrapCode, fromProjected.OwnerToStamp?.BootstrapCode);
+        Assert.Equal(fromFull.OwnerToStamp?.Serial, fromProjected.OwnerToStamp?.Serial);
+    }
+
+    public static IEnumerable<object?[]> OwnerScenarios()
+    {
+        yield return new object?[] { "legacy row, cert caller with the row's serial", null, Cert(Thumb1, Dev1) };
+        yield return new object?[] { "cert-owned row, the same cert", CertOwner(Thumb1, Dev1), Cert(Thumb1, Dev1) };
+        yield return new object?[] { "cert-owned row, a foreign cert", CertOwner(Thumb1, Dev1), Cert(Thumb2, "0f4e1c2d-2222-4aaa-8bbb-000000000002", "SN-OTHER") };
+        yield return new object?[] { "bootstrap-owned row, the cert that follows", BootstrapOwner("ABCD1234"), Cert(Thumb1, Dev1) };
+        yield return new object?[] { "cert-owned row, a bootstrap caller", CertOwner(Thumb1, Dev1), Bootstrap("ABCD1234") };
+    }
+
+    [Fact]
+    public void Lock_check_works_on_the_projected_row()
+    {
+        var full = FullRow(RowKind.RunningWhiteGlovePart2);
+        full["DeletionState"] = SessionDeletionState.Preparing;
+        full["PendingDeletionManifestId"] = "MANIFEST-1";
+        var projected = Project(full, SessionRowProjections.GuardRow.Concat(SessionDeletionGuard.LockColumns));
+        var guard = new SessionDeletionGuard(new Mock<ISessionDeletionInventoryReader>().Object, NullLogger<SessionDeletionGuard>.Instance);
+
+        var ex = Assert.Throws<SessionDeletionLockedException>(() => guard.ThrowIfLocked(projected, "V2.IngestTelemetry"));
+
+        Assert.Equal(SessionDeletionState.Preparing, ex.CurrentState);
+        Assert.Equal("MANIFEST-1", ex.ManifestId);
+    }
+
+    [Fact]
+    public async Task Projected_guard_read_asks_for_the_caller_columns_plus_the_lock_columns()
+    {
+        var reader = new Mock<ISessionDeletionInventoryReader>();
+        IEnumerable<string>? asked = null;
+        reader.Setup(r => r.GetSessionRowAsync(TenantId, SessionId, It.IsAny<IEnumerable<string>>(), It.IsAny<CancellationToken>()))
+              .Returns<string, string, IEnumerable<string>, CancellationToken>((_, _, select, _) =>
+              {
+                  asked = select.ToList();
+                  return Task.FromResult<TableEntity?>(Project(FullRow(RowKind.RunningWhiteGlovePart2), select));
+              });
+        var guard = new SessionDeletionGuard(reader.Object, NullLogger<SessionDeletionGuard>.Instance);
+
+        var row = await guard.EnsureWritableAndGetRowAsync(TenantId, SessionId, "V2.IngestTelemetry", SessionRowProjections.GuardRow);
+
+        Assert.NotNull(asked);
+        Assert.Equal(
+            SessionRowProjections.GuardRow.Concat(SessionDeletionGuard.LockColumns).OrderBy(c => c, StringComparer.Ordinal),
+            asked!.OrderBy(c => c, StringComparer.Ordinal));
+        Assert.NotNull(row);
+        Assert.Equal(SessionStatus.InProgress, SessionRowProjections.TryReadStatus(row));
+        Assert.Equal("SN-1", row!.GetString("SerialNumber"));
+        Assert.False(row.ContainsKey("DeviceName"), "the projected row carries only the requested columns");
+        reader.Verify(r => r.GetSessionRowAsync(It.IsAny<string>(), It.IsAny<string>(), It.IsAny<CancellationToken>()), Times.Never);
+    }
+
+    [Fact]
+    public async Task Projected_guard_read_still_refuses_a_locked_row_and_consults_the_tombstone_when_absent()
+    {
+        var locked = new Mock<ISessionDeletionInventoryReader>();
+        locked.Setup(r => r.GetSessionRowAsync(TenantId, SessionId, It.IsAny<IEnumerable<string>>(), It.IsAny<CancellationToken>()))
+              .ReturnsAsync(new TableEntity(TenantId, SessionId)
+              {
+                  ["DeletionState"] = SessionDeletionState.Queued,
+                  ["PendingDeletionManifestId"] = "MANIFEST-2",
+              });
+        var lockedGuard = new SessionDeletionGuard(locked.Object, NullLogger<SessionDeletionGuard>.Instance);
+        var ex = await Assert.ThrowsAsync<SessionDeletionLockedException>(
+            () => lockedGuard.EnsureWritableAndGetRowAsync(TenantId, SessionId, "RegisterSession", SessionRowProjections.GuardRow));
+        Assert.Equal("MANIFEST-2", ex.ManifestId);
+
+        var absent = new Mock<ISessionDeletionInventoryReader>();
+        absent.Setup(r => r.GetSessionRowAsync(TenantId, SessionId, It.IsAny<IEnumerable<string>>(), It.IsAny<CancellationToken>()))
+              .ReturnsAsync((TableEntity?)null);
+        absent.Setup(r => r.GetActiveSessionTombstoneAsync(TenantId, SessionId, It.IsAny<CancellationToken>()))
+              .ReturnsAsync((TableEntity?)null);
+        var absentGuard = new SessionDeletionGuard(absent.Object, NullLogger<SessionDeletionGuard>.Instance);
+        var row = await absentGuard.EnsureWritableAndGetRowAsync(TenantId, SessionId, "RegisterSession", SessionRowProjections.GuardRow);
+        Assert.Null(row);
+        absent.Verify(r => r.GetActiveSessionTombstoneAsync(TenantId, SessionId, It.IsAny<CancellationToken>()), Times.Once);
+    }
+
+    [Fact]
+    public void Projection_lists_carry_what_their_consumers_name()
+    {
+        Assert.Contains("Status", SessionRowProjections.GuardRow);
+        Assert.Contains("SerialNumber", SessionRowProjections.GuardRow);
+        Assert.All(SessionOwner.Columns.All, c => Assert.Contains(c, SessionRowProjections.GuardRow));
+        Assert.Equal(SessionRowProjections.GuardRow.Length, SessionRowProjections.GuardRow.Distinct(StringComparer.Ordinal).Count());
+
+        Assert.Contains(TableStorageService.ServerTimeStateColumn, SessionRowProjections.IngestSnapshot);
+        Assert.Contains(TableStorageService.StartedAtServerColumn, SessionRowProjections.IngestSnapshot);
+        Assert.Contains(TableStorageService.ResumedAtServerColumn, SessionRowProjections.IngestSnapshot);
+        Assert.Contains("CompletedAt", SessionRowProjections.IngestSnapshot);
+        Assert.Contains("IndexRowKey", SessionRowProjections.IngestSnapshot);
+        Assert.Equal(SessionRowProjections.IngestSnapshot.Length, SessionRowProjections.IngestSnapshot.Distinct(StringComparer.Ordinal).Count());
+    }
+
+    // ==================================================================== helpers ====
+
+    /// <summary>The counter merge of a second upload: later events, an older latest event, one of each counter.</summary>
+    private static Task<SessionIngestSnapshot?> Increment(TableStorageService sut) =>
+        sut.IncrementSessionEventCountAsync(
+            TenantId, SessionId, increment: 3,
+            earliestEventTimestamp: DeviceStart.AddMinutes(1),   // later than StartedAt: no anchor shift
+            latestEventTimestamp: DeviceStart.AddMinutes(19),    // older than LastEventAt: no move
+            currentPhase: null,
+            platformScriptIncrement: 1, remediationScriptIncrement: 1, rebootIncrement: 1,
+            upload: SecondUpload());
+
+    /// <summary>
+    /// A Sessions row with every column a real row carries. The running kind is WhiteGlove Part 2
+    /// in flight (stored Part 1 duration + ResumedAt drive the live duration) with the tracker
+    /// state of its first upload and the matching twin; the completed kind has CompletedAt, a
+    /// stored duration and an admin mark.
+    /// </summary>
+    private static TableEntity FullRow(RowKind kind, SessionOwner? owner = null)
+    {
+        var tracker = new ServerTime.Tracker();
+        tracker.Absorb(FirstUpload().Select(ServerTime.Event.From).ToList());
+
+        var row = new TableEntity(TenantId, SessionId)
+        {
+            ["StartedAt"] = new DateTimeOffset(DeviceStart),
+            ["LastEventAt"] = new DateTimeOffset(DeviceStart.AddMinutes(20)),
+            ["LastIngestAt"] = new DateTimeOffset(ServerStart.AddMinutes(20)),
+            ["EventCount"] = 2,
+            ["PlatformScriptCount"] = 3,
+            ["RemediationScriptCount"] = 1,
+            ["RebootCount"] = 1,
+            ["IndexRowKey"] = "2516000000000000000_" + SessionId, // deliberately not the key StartedAt would give
+            ["CurrentPhase"] = 4,
+            ["CurrentPhaseDetail"] = "Installing apps",
+            ["FailureReason"] = string.Empty,
+            ["DiagnosticsBlobName"] = "diag-" + SessionId + ".zip",
+            ["DiagnosticsBlobDestination"] = "platform",
+            ["IsPreProvisioned"] = true,
+            ["IsHybridJoin"] = false,
+            ["IsUserDriven"] = false,
+            ["IsSelfDeployingProfile"] = false,
+            ["IsCloudPc"] = false,
+            ["EnrollmentType"] = "v2",
+            ["PendingActionsJson"] = "[{\"Type\":\"collect_diagnostics\",\"Reason\":\"rule\"}]",
+            ["AgentVersion"] = "2.0.1478",
+            ["ImeAgentVersion"] = "1.99.1",
+            [TableStorageService.ServerTimeStateColumn] = tracker.ToJson(),
+            ["StartedAtServer"] = new DateTimeOffset(tracker.StartedAtServer!.Value),
+            // Columns the hot path never needs — the projection must drop them without a trace.
+            ["DeviceName"] = "DESKTOP-TEST01",
+            ["SerialNumber"] = "SN-1",
+            ["Manufacturer"] = "Contoso",
+            ["Model"] = "Model X",
+            ["OsName"] = "Windows 11 Enterprise",
+            ["OsBuild"] = "26100.4000",
+            ["OsDisplayVersion"] = "24H2",
+            ["GeoCountry"] = "DE",
+            ["GeoRegion"] = "HE",
+            ["GeoCity"] = "Frankfurt",
+            ["GeoLoc"] = "50.11,8.68",
+            ["ConnectionType"] = "Ethernet",
+            ["ValidatedBy"] = "AutopilotV1",
+            ["VerdictPath"] = "ime_complete",
+            ["FailureSnapshotJson"] = new string('x', 2048),
+            ["AvgApiLatencyMs"] = 123.4,
+            ["ApiRequestCount"] = 17,
+            ["DeletionState"] = SessionDeletionState.None,
+        };
+
+        if (kind == RowKind.RunningWhiteGlovePart2)
+        {
+            row["Status"] = "InProgress";
+            row["ResumedAt"] = new DateTimeOffset(DeviceStart.AddMinutes(10));
+            row["DurationSeconds"] = 600; // Part 1, stored
+            row["AdminMarkedAction"] = "Succeeded";
+        }
+        else
+        {
+            row["Status"] = "Succeeded";
+            row["CompletedAt"] = new DateTimeOffset(DeviceStart.AddMinutes(40));
+            row["CompletedAtServer"] = new DateTimeOffset(ServerStart.AddMinutes(40));
+            row["DurationSeconds"] = 2400;
+            row["AdminMarkedAction"] = "Succeeded";
+            row["ReconcileReason"] = "ime_complete";
+            row["CompletionSource"] = "agent";
+        }
+
+        if (owner != null)
+            SessionOwnershipPolicy.ApplyTo(row, owner);
+
+        row.ETag = new ETag("0xEXISTING");
+        return row;
+    }
+
+    private static List<EnrollmentEvent> FirstUpload() => new()
+    {
+        Event(1, "agent_started", "Agent", DeviceStart, sent: DeviceStart.AddSeconds(10), received: ServerStart.AddSeconds(10)),
+        Event(2, "performance_snapshot", "PerformanceCollector", DeviceStart.AddSeconds(5), sent: DeviceStart.AddSeconds(10), received: ServerStart.AddSeconds(10)),
+    };
+
+    private static List<EnrollmentEvent> SecondUpload() => new()
+    {
+        Event(3, "app_install_started", "ImeLogTracker", DeviceStart.AddSeconds(30), sent: DeviceStart.AddSeconds(60), received: ServerStart.AddSeconds(60)),
+        Event(4, "performance_snapshot", "PerformanceCollector", DeviceStart.AddSeconds(40), sent: DeviceStart.AddSeconds(60), received: ServerStart.AddSeconds(60)),
+    };
+
+    private static EnrollmentEvent Event(long sequence, string type, string source, DateTime time, DateTime? sent, DateTime received) => new()
+    {
+        EventId = $"e{sequence}",
+        TenantId = TenantId,
+        SessionId = SessionId,
+        Sequence = sequence,
+        EventType = type,
+        Source = source,
+        Timestamp = time,
+        SentAt = sent,
+        ReceivedAt = received,
+        Message = string.Empty,
+        Data = new Dictionary<string, object>(),
+    };
+
+    private static SecurityValidationResult Cert(string thumb, string? deviceId, string? serial = "SN-1") => new()
+    {
+        IsValid = true,
+        CertificateThumbprint = thumb,
+        IntuneDeviceId = deviceId,
+        SerialNumber = serial,
+        ValidatedBy = ValidatorType.AutopilotV1,
+    };
+
+    private static SecurityValidationResult Bootstrap(string code, string? serial = "SN-1") => new()
+    {
+        IsValid = true,
+        IsBootstrapAuth = true,
+        BootstrapShortCode = code,
+        SerialNumber = serial,
+        ValidatedBy = ValidatorType.Bootstrap,
+    };
+
+    private static SessionOwner CertOwner(string thumb, string? deviceId, string serial = "SN-1") =>
+        new() { Kind = SessionOwner.Kinds.Cert, Thumbprint = thumb, DeviceId = deviceId, Serial = serial, BoundAt = ServerStart };
+
+    private static SessionOwner BootstrapOwner(string code, string serial = "SN-1") =>
+        new() { Kind = SessionOwner.Kinds.Bootstrap, BootstrapCode = code, Serial = serial, BoundAt = ServerStart };
+
+    /// <summary>What a select does on the service: keys, ETag and timestamp survive, every other column is gone.</summary>
+    private static TableEntity Project(TableEntity row, IEnumerable<string>? select)
+    {
+        if (select == null)
+            return row;
+        var keep = new HashSet<string>(select, StringComparer.Ordinal);
+        var projected = new TableEntity(row.PartitionKey, row.RowKey) { ETag = row.ETag, Timestamp = row.Timestamp };
+        foreach (var kv in row)
+        {
+            if (keep.Contains(kv.Key))
+                projected[kv.Key] = kv.Value;
+        }
+        return projected;
+    }
+
+    private static readonly string[] SystemKeys = { "PartitionKey", "RowKey", "Timestamp", "odata.etag" };
+
+    private static void AssertSameEntity(TableEntity expected, TableEntity actual)
+    {
+        var expectedKeys = expected.Keys.Except(SystemKeys).OrderBy(k => k, StringComparer.Ordinal).ToList();
+        var actualKeys = actual.Keys.Except(SystemKeys).OrderBy(k => k, StringComparer.Ordinal).ToList();
+        Assert.Equal(expectedKeys, actualKeys);
+        Assert.Equal(expected.PartitionKey, actual.PartitionKey);
+        Assert.Equal(expected.RowKey, actual.RowKey);
+        foreach (var key in expectedKeys)
+        {
+            if (key == "LastIngestAt")
+            {
+                var delta = expected.GetDateTimeOffset(key)!.Value - actual.GetDateTimeOffset(key)!.Value;
+                Assert.True(Math.Abs(delta.TotalSeconds) < 10, $"{key}: {delta}");
+                continue;
+            }
+            Assert.True(Equals(expected[key], actual[key]), $"{key}: {expected[key]} vs {actual[key]}");
+        }
+    }
+
+    private static void AssertSameSnapshot(SessionIngestSnapshot expected, SessionIngestSnapshot actual)
+    {
+        Assert.Equal(expected.Status, actual.Status);
+        Assert.Equal(expected.CurrentPhase, actual.CurrentPhase);
+        Assert.Equal(expected.CurrentPhaseDetail, actual.CurrentPhaseDetail);
+        Assert.Equal(expected.FailureReason, actual.FailureReason);
+        Assert.Equal(expected.EventCount, actual.EventCount);
+        Assert.Equal(expected.CompletedAt, actual.CompletedAt);
+        Assert.Equal(expected.DiagnosticsBlobName, actual.DiagnosticsBlobName);
+        Assert.Equal(expected.IsPreProvisioned, actual.IsPreProvisioned);
+        Assert.Equal(expected.AdminMarkedAction, actual.AdminMarkedAction);
+        Assert.Equal(expected.PendingActionsJson, actual.PendingActionsJson);
+        Assert.Equal(expected.AgentVersion, actual.AgentVersion);
+        Assert.Equal(expected.ImeAgentVersion, actual.ImeAgentVersion);
+        // The live duration of a running session is computed from the clock; the two mappings run
+        // milliseconds apart.
+        Assert.NotNull(expected.DurationSeconds);
+        Assert.NotNull(actual.DurationSeconds);
+        Assert.True(Math.Abs(expected.DurationSeconds!.Value - actual.DurationSeconds!.Value) <= 5,
+            $"DurationSeconds: {expected.DurationSeconds} vs {actual.DurationSeconds}");
+    }
+
+    /// <summary>
+    /// SDK-mock harness over the real <see cref="TableStorageService"/>. With <c>honourSelect</c>
+    /// the Sessions point read returns only the columns the caller selected (what the service
+    /// does); without it the whole row (what the other session harnesses do). Merges on Sessions
+    /// and SessionsIndex are captured.
+    /// </summary>
+    private sealed class Harness
+    {
+        public TableStorageService Sut { get; }
+        public TableEntity? Written { get; private set; }
+        public ETag WrittenETag { get; private set; }
+        public TableEntity? IndexWritten { get; private set; }
+        public IEnumerable<string>? LastSelect { get; private set; }
+        public int LastReadColumnCount { get; private set; }
+
+        public Harness(TableEntity existing, bool honourSelect)
+        {
+            var sessions = new Mock<TableClient>();
+            sessions.Setup(t => t.GetEntityIfExistsAsync<TableEntity>(
+                    TenantId, SessionId, It.IsAny<IEnumerable<string>>(), It.IsAny<CancellationToken>()))
+                .Returns<string, string, IEnumerable<string>, CancellationToken>((_, _, select, _) =>
+                {
+                    LastSelect = select?.ToList();
+                    var served = honourSelect ? Project(existing, select) : existing;
+                    LastReadColumnCount = served.Count;
+                    var response = new Mock<NullableResponse<TableEntity>>();
+                    response.SetupGet(r => r.HasValue).Returns(true);
+                    response.SetupGet(r => r.Value).Returns(served);
+                    return Task.FromResult(response.Object);
+                });
+            sessions.Setup(t => t.UpdateEntityAsync(
+                    It.IsAny<TableEntity>(), It.IsAny<ETag>(), It.IsAny<TableUpdateMode>(), It.IsAny<CancellationToken>()))
+                .Returns<TableEntity, ETag, TableUpdateMode, CancellationToken>((e, etag, _, _) =>
+                {
+                    Written = e;
+                    WrittenETag = etag;
+                    return Task.FromResult(new Mock<Response>().Object);
+                });
+
+            var index = new Mock<TableClient>();
+            index.Setup(t => t.UpdateEntityAsync(
+                    It.IsAny<TableEntity>(), It.IsAny<ETag>(), It.IsAny<TableUpdateMode>(), It.IsAny<CancellationToken>()))
+                .Returns<TableEntity, ETag, TableUpdateMode, CancellationToken>((e, _, _, _) =>
+                {
+                    IndexWritten = e;
+                    return Task.FromResult(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.SessionsIndex)).Returns(index.Object);
+            Sut = new TableStorageService(serviceClient.Object, NullLogger<TableStorageService>.Instance);
+        }
+    }
+}
@@ -252,6 +252,9 @@ public class TableLifecycleBucketTests
        public Task<TableEntity?> GetSessionRowAsync(string tenantId, string sessionId, CancellationToken cancellationToken = default)
            => Task.FromResult<TableEntity?>(new TableEntity(tenantId, sessionId) { ["IndexRowKey"] = IndexRowKey });

+        public Task<TableEntity?> GetSessionRowAsync(string tenantId, string sessionId, IEnumerable<string> select, CancellationToken cancellationToken = default)
+            => GetSessionRowAsync(tenantId, sessionId, cancellationToken);
+
        public Task<TableEntity?> GetSessionsIndexRowAsync(string tenantId, string indexRowKey, CancellationToken cancellationToken = default)
            => Task.FromResult<TableEntity?>(new TableEntity(tenantId, indexRowKey) { ["SessionId"] = SessionId });

@@ -33,6 +33,9 @@ namespace AutopilotMonitor.Functions.DataAccess.TableStorage
        public Task<SessionSummary?> GetSessionAsync(string tenantId, string sessionId)
            => _storage.GetSessionAsync(tenantId, sessionId);

+        public Task<SessionIngestSnapshot?> GetSessionIngestSnapshotAsync(string tenantId, string sessionId)
+            => _storage.GetSessionIngestSnapshotAsync(tenantId, sessionId);
+
        public Task<string?> ResolveSessionTenantIdAsync(string sessionId)
            => _storage.ResolveSessionTenantIdAsync(sessionId);

@@ -146,7 +149,7 @@ namespace AutopilotMonitor.Functions.DataAccess.TableStorage
        public Task<List<ServerAction>> FetchAndClearPendingActionsAsync(string tenantId, string sessionId)
            => _storage.FetchAndClearPendingActionsAsync(tenantId, sessionId);

-        public Task<SessionSummary?> IncrementSessionEventCountAsync(
+        public Task<SessionIngestSnapshot?> IncrementSessionEventCountAsync(
            string tenantId, string sessionId, int increment,
            DateTime? earliestEventTimestamp = null, DateTime? latestEventTimestamp = null,
            EnrollmentPhase? currentPhase = null,
@@ -237,7 +237,7 @@ namespace AutopilotMonitor.Functions.Functions.Ingest
                Azure.Data.Tables.TableEntity? guardSessionRow;
                try
                {
-                    guardSessionRow = await _deletionGuard.EnsureWritableAndGetRowAsync(bodyTenantId, sessionId, "V2.IngestTelemetry");
+                    guardSessionRow = await _deletionGuard.EnsureWritableAndGetRowAsync(bodyTenantId, sessionId, "V2.IngestTelemetry", SessionRowProjections.GuardRow);
                }
                catch (SessionDeletionLockedException locked)
                {
@@ -262,9 +262,9 @@ namespace AutopilotMonitor.Functions.Functions.Ingest
                        return AsOutput(await WriteDeviceBlockedAsync(req, rowSerialVerdict));
                }

-                // Reuse the guard's full-row read: its Status feeds the stall-heal check inside
-                // EventIngestProcessor, saving one Sessions point-read per batch.
-                var preFetchedStatus = TryReadSessionStatus(guardSessionRow);
+                // Reuse the guard's row (projected to SessionRowProjections.GuardRow): its Status feeds
+                // the stall-heal check inside EventIngestProcessor, saving one Sessions point-read per batch.
+                var preFetchedStatus = SessionRowProjections.TryReadStatus(guardSessionRow);

                // SESSION-OWNER-BINDING: is the device behind this certificate/token the device this
                // session belongs to? Same guard row, zero extra reads. A foreign identity gets 403
@@ -449,26 +449,6 @@ namespace AutopilotMonitor.Functions.Functions.Ingest
            });
        }

-        /// <summary>
-        /// Extracts <c>Status</c> from a Sessions row the deletion guard already loaded. Sessions
-        /// writes Status as a STRING (<c>status.ToString()</c> in UpdateSessionStatusAsync — never
-        /// an int), so this mirrors the canonical mapper's parse: <c>Enum.TryParse</c>,
-        /// case-insensitive. Returns null for missing/unparseable values (incl. a defensive int
-        /// fallback for any legacy numeric shape) — callers then fall back to their own read.
-        /// </summary>
-        internal static SessionStatus? TryReadSessionStatus(Azure.Data.Tables.TableEntity? sessionRow)
-        {
-            if (sessionRow == null || !sessionRow.TryGetValue("Status", out var statusValue))
-                return null;
-
-            return statusValue switch
-            {
-                string s when Enum.TryParse<SessionStatus>(s, ignoreCase: true, out var parsed) => parsed,
-                int i when Enum.IsDefined(typeof(SessionStatus), i) => (SessionStatus)i,
-                _ => null,
-            };
-        }
-
        /// <summary>
        /// Routes every item of the batch by <see cref="TelemetryItemDto.Kind"/> into its storage
        /// shape. Pure and total: an item whose Kind is not a <see cref="TelemetryItemKind"/> name
@@ -596,7 +576,7 @@ namespace AutopilotMonitor.Functions.Functions.Ingest
        private async Task<(string? adminAction, List<ServerAction>? pendingActions)>
            ReadControlSignalsAsync(string tenantId, string sessionId)
        {
-            var session = await _sessionRepo.GetSessionAsync(tenantId, sessionId);
+            var session = await _sessionRepo.GetSessionIngestSnapshotAsync(tenantId, sessionId);
            if (session == null) return (null, null);

            // AdminAction carries the portal-button signal only. The old logic (Status ==
@@ -4,6 +4,7 @@ using System.Globalization;
 using System.Linq;
 using System.Net;
 using System.Text.Json;
+using AutopilotMonitor.Functions.Helpers;
 using AutopilotMonitor.Functions.Security;
 using AutopilotMonitor.Functions.Services;
 using AutopilotMonitor.Functions.Services.Deletion;
@@ -467,7 +468,7 @@ namespace AutopilotMonitor.Functions.Functions.Ingest
            if (deletionLocked)
                return new(OpsEventSeverity.Info, false, false, "deleting", "session is being deleted — nothing written");

-            var status = IngestTelemetryFunction.TryReadSessionStatus(sessionRow);
+            var status = SessionRowProjections.TryReadStatus(sessionRow);
            var statusText = status?.ToString() ?? "unknown";
            return status.HasValue && Helpers.DeviceJourneyCalculator.IsTerminal(status.Value)
                ? new(OpsEventSeverity.Info, true, true, statusText, $"late cleanup — session already {statusText} when the break arrived")
@@ -167,7 +167,7 @@ namespace AutopilotMonitor.Functions.Functions.Sessions
            try
            {
                guardSessionRow = await _deletionGuard.EnsureWritableAndGetRowAsync(
-                    registration.TenantId, registration.SessionId, "RegisterSession");
+                    registration.TenantId, registration.SessionId, "RegisterSession", SessionRowProjections.GuardRow);
            }
            catch (SessionDeletionLockedException locked)
            {
@@ -180,14 +180,14 @@ namespace AutopilotMonitor.Functions.Functions.Sessions
                };
            }

-            // Pre-read the session row so we can distinguish three lifecycle entries:
-            //   - existing == null      → first-time registration (fresh enrollment)
-            //   - existing.Pending      → WhiteGlove Part 2 resume (user-driven phase)
-            //   - everything else       → agent restart / terminal re-register (no start signal)
+            // The guard's row (projected to SessionRowProjections.GuardRow) distinguishes three
+            // lifecycle entries — no second read of the Sessions row:
+            //   - row absent          → first-time registration (fresh enrollment)
+            //   - row Pending         → WhiteGlove Part 2 resume (user-driven phase)
+            //   - everything else     → agent restart / terminal re-register (no start signal)
            // Used below to scope the opt-in "enrollment started" webhook to real start events.
-            var preExistingSession = await _sessionRepo.GetSessionAsync(registration.TenantId, registration.SessionId);
-            bool isFreshRegistration = preExistingSession == null;
-            bool isWhiteGloveResume = preExistingSession?.Status == SessionStatus.Pending;
+            bool isFreshRegistration = guardSessionRow == null;
+            bool isWhiteGloveResume = SessionRowProjections.TryReadStatus(guardSessionRow) == SessionStatus.Pending;

            // SESSION-OWNER-BINDING: compare the caller's validated identity against the owner
            // already on the row (reusing the guard's read). A foreign identity is refused with
@@ -0,0 +1,67 @@
+using System;
+using System.Linq;
+using Azure.Data.Tables;
+using AutopilotMonitor.Functions.Services;
+using AutopilotMonitor.Shared.Models;
+
+namespace AutopilotMonitor.Functions.Helpers
+{
+    /// <summary>
+    /// Column lists for the projected point reads of the Sessions row on the agent hot path, and
+    /// the reader of the status they carry. Every upload reads the row twice (deletion guard,
+    /// counter merge); read whole, the row's size set the ingest latency (backend 1.5.1256:
+    /// +35 ms p50 from ~400 bytes of new columns, the storage server unchanged). Each list is
+    /// pinned by <c>SessionRowProjectionTests</c> against a select-strict table: a column a
+    /// consumer reads but the list lacks fails there, not in production as a silent null.
+    /// </summary>
+    public static class SessionRowProjections
+    {
+        /// <summary>
+        /// What the agent routes read off the deletion guard's row: the status prefetch
+        /// (<see cref="TryReadStatus"/>, the registration's lifecycle entry), the registered serial
+        /// for the kill switch, and the SESSION-OWNER-BINDING columns
+        /// (<c>SessionOwnershipPolicy.Evaluate</c>). The guard adds its own lock columns.
+        /// </summary>
+        public static readonly string[] GuardRow =
+            new[] { "Status", "SerialNumber" }.Concat(SessionOwner.Columns.All).ToArray();
+
+        /// <summary>
+        /// What the counter merge (<c>IncrementSessionEventCountAsync</c>) reads for its own
+        /// read-modify-write and what the <c>SessionIngestSnapshot</c> it returns carries.
+        /// </summary>
+        public static readonly string[] IngestSnapshot =
+        {
+            // Counter read-modify-write, anchor alignment, index dual-write.
+            "EventCount", "StartedAt", "LastEventAt", "PlatformScriptCount", "RemediationScriptCount", "RebootCount",
+            "IndexRowKey",
+            // Server-time tracker (StageServerTimeUpload / StageServerTimeTwin).
+            "CompletedAt", TableStorageService.ServerTimeStateColumn,
+            TableStorageService.StartedAtServerColumn, TableStorageService.ResumedAtServerColumn,
+            // The snapshot: SignalR delta, control signals, pattern health, shutdown close, live duration
+            // (ComputeEffectiveDuration reads DurationSeconds, IsPreProvisioned, ResumedAt).
+            "Status", "CurrentPhase", "CurrentPhaseDetail", "FailureReason", "DurationSeconds",
+            "DiagnosticsBlobName", "IsPreProvisioned", "ResumedAt", "AdminMarkedAction", "PendingActionsJson",
+            "AgentVersion", "ImeAgentVersion",
+        };
+
+        /// <summary>
+        /// <c>Status</c> off a Sessions row (the guard's row). Sessions writes Status as a STRING
+        /// (<c>status.ToString()</c> in UpdateSessionStatusAsync — never an int), so this mirrors the
+        /// canonical mapper's parse: <c>Enum.TryParse</c>, case-insensitive. Null for missing or
+        /// unparseable values (incl. a defensive int fallback for any legacy numeric shape) — callers
+        /// then fall back to their own read.
+        /// </summary>
+        public static SessionStatus? TryReadStatus(TableEntity? sessionRow)
+        {
+            if (sessionRow == null || !sessionRow.TryGetValue("Status", out var statusValue))
+                return null;
+
+            return statusValue switch
+            {
+                string s when Enum.TryParse<SessionStatus>(s, ignoreCase: true, out var parsed) => parsed,
+                int i when Enum.IsDefined(typeof(SessionStatus), i) => (SessionStatus)i,
+                _ => null,
+            };
+        }
+    }
+}
@@ -20,6 +20,13 @@ namespace AutopilotMonitor.Functions.Services.Deletion
        /// </summary>
        Task<TableEntity?> GetSessionRowAsync(string tenantId, string sessionId, CancellationToken cancellationToken = default);

+        /// <summary>
+        /// <see cref="GetSessionRowAsync(string, string, CancellationToken)"/> projected to
+        /// <paramref name="select"/> — the hot-path form (the agent routes read the guard's row this
+        /// way, see <c>SessionRowProjections.GuardRow</c>). Returns null when the row does not exist.
+        /// </summary>
+        Task<TableEntity?> GetSessionRowAsync(string tenantId, string sessionId, IEnumerable<string> select, CancellationToken cancellationToken = default);
+
        /// <summary>
        /// Loads a SessionsIndex row by its (tenantId, indexRowKey) coordinate. Used for the
        /// FINAL tombstone step's row dump. Returns null when missing (treated as already-gone).
@@ -1,3 +1,5 @@
+using System.Collections.Generic;
+using System.Linq;
 using System.Threading;
 using System.Threading.Tasks;
 using Azure.Data.Tables;
@@ -80,6 +82,26 @@ namespace AutopilotMonitor.Functions.Services.Deletion
        public async Task<TableEntity?> EnsureWritableAndGetRowAsync(string tenantId, string sessionId, string callerContext, CancellationToken cancellationToken = default)
        {
            var sessionRow = await _reader.GetSessionRowAsync(tenantId, sessionId, cancellationToken);
+            return await GuardRowAsync(sessionRow, tenantId, sessionId, callerContext, cancellationToken);
+        }
+
+        /// <summary>Columns the guard reads off the row itself; a projected read adds them to the caller's list.</summary>
+        public static readonly string[] LockColumns = { "DeletionState", "PendingDeletionManifestId" };
+
+        /// <summary>
+        /// <see cref="EnsureWritableAndGetRowAsync(string, string, string, CancellationToken)"/> with the
+        /// row projected to <paramref name="select"/> plus <see cref="LockColumns"/> — the agent hot
+        /// path's form (see <c>SessionRowProjections.GuardRow</c>). The returned row carries only
+        /// those columns.
+        /// </summary>
+        public async Task<TableEntity?> EnsureWritableAndGetRowAsync(string tenantId, string sessionId, string callerContext, IReadOnlyCollection<string> select, CancellationToken cancellationToken = default)
+        {
+            var sessionRow = await _reader.GetSessionRowAsync(tenantId, sessionId, select.Concat(LockColumns).Distinct(), cancellationToken);
+            return await GuardRowAsync(sessionRow, tenantId, sessionId, callerContext, cancellationToken);
+        }
+
+        private async Task<TableEntity?> GuardRowAsync(TableEntity? sessionRow, string tenantId, string sessionId, string callerContext, CancellationToken cancellationToken)
+        {
            if (sessionRow != null)
            {
                ThrowIfLocked(sessionRow, callerContext);
@@ -1,4 +1,5 @@
 using AutopilotMonitor.Shared;
+using AutopilotMonitor.Shared.DataAccess;
 using System.Linq;
 using AutopilotMonitor.Functions.Services.Notifications;
 using AutopilotMonitor.Shared.Models;
@@ -134,19 +135,21 @@ namespace AutopilotMonitor.Functions.Services
            });

        private SignalRMessageAction[] BuildSignalRMessages(
-            IngestEventsRequest request, SessionSummary? updatedSession, int processedCount,
+            IngestEventsRequest request, SessionIngestSnapshot? snapshot, int processedCount,
            List<RuleResult> newRuleResults)
        {
-            object? sessionDelta = updatedSession != null ? new {
-                updatedSession.CurrentPhase,
-                updatedSession.CurrentPhaseDetail,
-                updatedSession.Status,
-                updatedSession.FailureReason,
-                updatedSession.EventCount,
-                updatedSession.DurationSeconds,
-                updatedSession.CompletedAt,
-                updatedSession.DiagnosticsBlobName,
-                updatedSession.IsPreProvisioned
+            // The delta's member names are the portal's live-update contract (sessionUpdate); the
+            // snapshot carries exactly these fields, projected off the row.
+            object? sessionDelta = snapshot != null ? new {
+                snapshot.CurrentPhase,
+                snapshot.CurrentPhaseDetail,
+                snapshot.Status,
+                snapshot.FailureReason,
+                snapshot.EventCount,
+                snapshot.DurationSeconds,
+                snapshot.CompletedAt,
+                snapshot.DiagnosticsBlobName,
+                snapshot.IsPreProvisioned
            } : null;

            var summaryMessage = new SignalRMessageAction(Constants.SignalRMessages.NewEvents)
@@ -219,14 +219,15 @@ namespace AutopilotMonitor.Functions.Services
                || classification.EspFailureEvent != null
                || classification.GatherCompletionEvent != null;

-            // The increment's post-merge snapshot serves as the "updatedSession" for the common
-            // case (non-terminal batch, no diagnostics upload) — it already reflects this batch's
-            // status transition (written above) plus the counter merge, saving the follow-up
-            // GetSessionAsync that used to run on every batch.
-            SessionSummary? updatedSession = null;
+            // The increment's post-merge snapshot (the ingest slice of the row, read projected —
+            // SessionRowProjections.IngestSnapshot) serves the common case (non-terminal batch, no
+            // transition, no diagnostics upload): it already reflects this batch's status transition
+            // (written above) plus the counter merge. The whole row is read only below, where a
+            // consumer needs it.
+            SessionIngestSnapshot? snapshot = null;
            if (processedCount > 0)
            {
-                updatedSession = await _sessionRepo.IncrementSessionEventCountAsync(
+                snapshot = await _sessionRepo.IncrementSessionEventCountAsync(
                    request.TenantId,
                    request.SessionId,
                    processedCount,
@@ -253,7 +254,7 @@ namespace AutopilotMonitor.Functions.Services
                if (patternHitsEvent.Data!.TryGetValue("imeVersion", out var imeVersionObj))
                    hitsImeVersion = imeVersionObj?.ToString();
                if (string.IsNullOrWhiteSpace(hitsImeVersion))
-                    hitsImeVersion = updatedSession?.ImeAgentVersion;
+                    hitsImeVersion = snapshot?.ImeAgentVersion;

                _ = _imePatternHealth.RecordSessionHitsAsync(hitsImeVersion, patternHitsEvent.Data, request.TenantId, request.SessionId);
            }
@@ -274,7 +275,7 @@ namespace AutopilotMonitor.Functions.Services
                // 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);
+                        snapshot?.AgentVersion, skewScan);
            }

            // Observation end: the batch carrying agent_shutting_down is the agent's last word.
@@ -284,10 +285,10 @@ namespace AutopilotMonitor.Functions.Services
            // and the agent keeps watching after enrollment_complete). Non-terminal sessions
            // (reboot shutdown, WhiteGlove part 1) keep their rows open. Idempotent + fail-soft.
            if (classification.AgentShutdownEvent != null
-                && updatedSession != null
-                && (updatedSession.Status == SessionStatus.Succeeded
-                    || updatedSession.Status == SessionStatus.Failed
-                    || updatedSession.Status == SessionStatus.Incomplete))
+                && snapshot != null
+                && (snapshot.Status == SessionStatus.Succeeded
+                    || snapshot.Status == SessionStatus.Failed
+                    || snapshot.Status == SessionStatus.Incomplete))
            {
                await _metricsRepo.CloseOpenAppInstallsForSessionAsync(request.TenantId, request.SessionId);
            }
@@ -430,12 +431,18 @@ namespace AutopilotMonitor.Functions.Services
                }
            }

-            // Re-read only when a write AFTER the increment made the snapshot stale: terminal
-            // batches (ReconcileSessionCountersAsync) and diagnostics uploads (blob fields) —
-            // or when no increment ran / it returned null (missing row, exhausted ETag retries).
-            if (isTerminalBatch || classification.DiagnosticsUploadedEvent != null)
-                updatedSession = null;
-            updatedSession ??= await _sessionRepo.GetSessionAsync(request.TenantId, request.SessionId);
+            // The whole row is read only where a consumer needs it: terminal batches
+            // (ReconcileSessionCountersAsync wrote after the increment), diagnostics uploads (blob
+            // fields), status or WhiteGlove transitions (webhooks, SLA) — and when no increment ran
+            // or it returned null (missing row, exhausted ETag retries). Every other batch works off
+            // the projected snapshot; here it is derived from the full read so both paths agree.
+            SessionSummary? updatedSession = null;
+            if (isTerminalBatch || classification.DiagnosticsUploadedEvent != null
+                || statusTransitioned || whiteGloveStatusTransitioned || snapshot == null)
+            {
+                updatedSession = await _sessionRepo.GetSessionAsync(request.TenantId, request.SessionId);
+                snapshot = updatedSession != null ? SessionIngestSnapshot.From(updatedSession) : null;
+            }

            // NOTE: long-running InProgress sessions are handled authoritatively by
            // MaintenanceService.MarkStalledSessionsAsTimedOutAsync (Stalled at 2h agent-silence,
@@ -443,13 +450,13 @@ namespace AutopilotMonitor.Functions.Services
            // here was pure observability noise (fired on every ingest of a >4h session, strictly
            // later than maintenance's first action) and was removed.

-            if (classification.WhiteGloveEvent != null && updatedSession?.IsPreProvisioned != true)
+            if (classification.WhiteGloveEvent != null && snapshot?.IsPreProvisioned != true)
            {
                _logger.LogError(
                    "{SessionPrefix} WhiteGlove status update not persisted after retries and fallback. " +
                    "IsPreProvisioned={IsPreProvisioned}, Status={Status}. " +
                    "Proceeding with 200 to allow agent spool drain.",
-                    sessionPrefix, updatedSession?.IsPreProvisioned, updatedSession?.Status);
+                    sessionPrefix, snapshot?.IsPreProvisioned, snapshot?.Status);
            }

            await SendWebhookNotificationsAsync(
@@ -468,7 +475,7 @@ namespace AutopilotMonitor.Functions.Services
            // fired falsely for every post-completion agent event (agent_shutting_down,
            // diagnostics_uploaded, enrollment_summary_shown) — making the agent believe an
            // admin had clicked Mark-Succeeded after its own completion.
-            string? adminAction = updatedSession?.AdminMarkedAction;
+            string? adminAction = snapshot?.AdminMarkedAction;
            if (!string.IsNullOrEmpty(adminAction))
            {
                _logger.LogInformation(
@@ -477,7 +484,7 @@ namespace AutopilotMonitor.Functions.Services
            }

            List<ServerAction>? pendingActions = null;
-            if (updatedSession != null && !string.IsNullOrEmpty(updatedSession.PendingActionsJson))
+            if (snapshot != null && !string.IsNullOrEmpty(snapshot.PendingActionsJson))
            {
                var fetched = await _sessionRepo.FetchAndClearPendingActionsAsync(request.TenantId, request.SessionId);
                if (fetched.Count > 0)
@@ -502,7 +509,7 @@ namespace AutopilotMonitor.Functions.Services
                }
            }

-            var signalRMessages = BuildSignalRMessages(request, updatedSession, processedCount, newRuleResults);
+            var signalRMessages = BuildSignalRMessages(request, snapshot, processedCount, newRuleResults);

            return new EventIngestResult
            {
@@ -56,6 +56,20 @@ namespace AutopilotMonitor.Functions.Services
            }
        }

+        public virtual async Task<TableEntity?> GetSessionRowAsync(string tenantId, string sessionId, IEnumerable<string> select, CancellationToken cancellationToken = default)
+        {
+            var tableClient = _tableServiceClient.GetTableClient(Shared.Constants.TableNames.Sessions);
+            try
+            {
+                var response = await tableClient.GetEntityAsync<TableEntity>(tenantId, sessionId, select: select, cancellationToken: cancellationToken);
+                return response.Value;
+            }
+            catch (RequestFailedException ex) when (ex.Status == 404)
+            {
+                return null;
+            }
+        }
+
        public async Task<TableEntity?> GetSessionsIndexRowAsync(string tenantId, string indexRowKey, CancellationToken cancellationToken = default)
        {
            var tableClient = _tableServiceClient.GetTableClient(Shared.Constants.TableNames.SessionsIndex);
@@ -7,6 +7,7 @@ using AutopilotMonitor.Functions.Pagination;
 using AutopilotMonitor.Functions.Security;
 using AutopilotMonitor.Functions.Services.Caching;
 using AutopilotMonitor.Shared;
+using AutopilotMonitor.Shared.DataAccess;
 using AutopilotMonitor.Shared.Models;
 using AutopilotMonitor.Shared.Models.Deletion;
 using AutopilotMonitor.Shared.Pagination;
@@ -1687,6 +1688,26 @@ namespace AutopilotMonitor.Functions.Services
            }
        }

+        /// <summary>
+        /// The ingest slice of the session row, read projected to
+        /// <see cref="SessionRowProjections.IngestSnapshot"/> (the signal-only upload's control
+        /// signals). Null when the row is missing; fail-soft like <see cref="GetSessionAsync"/>.
+        /// </summary>
+        public async Task<SessionIngestSnapshot?> GetSessionIngestSnapshotAsync(string tenantId, string sessionId)
+        {
+            try
+            {
+                var tableClient = _tableServiceClient.GetTableClient(Constants.TableNames.Sessions);
+                var entity = await tableClient.GetEntityIfExistsAsync<TableEntity>(tenantId, sessionId, select: SessionRowProjections.IngestSnapshot);
+                return entity.HasValue ? MapToIngestSnapshot(entity.Value!) : null;
+            }
+            catch (Exception ex)
+            {
+                _logger.LogError(ex, $"Failed to get the ingest snapshot of session {sessionId}");
+                return null;
+            }
+        }
+
        /// <summary>
        /// Open (non-terminal) sessions of the same physical device within one tenant: Pending,
        /// InProgress, Stalled or AwaitingUser rows matching the SerialNumber. Server-side
@@ -2570,8 +2591,10 @@ namespace AutopilotMonitor.Functions.Services
        /// Uses Merge mode to safely handle concurrent updates.
        /// The caller provides earliestEventTimestamp from the current batch;
        /// no redundant Events-table scan is performed here.
-        /// Returns the post-merge snapshot (RMW read + applied increments) so the ingest hot path
-        /// can skip its follow-up GetSessionAsync; null on missing row / exhausted retries / error.
+        /// The read is projected to <see cref="SessionRowProjections.IngestSnapshot"/> (the hot
+        /// path never pays for the whole row) and the post-merge values of that slice come back as
+        /// a <see cref="SessionIngestSnapshot"/> so the ingest hot path can skip its follow-up read;
+        /// null on missing row / exhausted retries / error.
        /// <para>
        /// Retry semantics: these read-modify-write increments are NOT idempotent under the
        /// agent's at-least-once retry (rows dedupe, counters double). EventCount and RebootCount
@@ -2584,7 +2607,7 @@ namespace AutopilotMonitor.Functions.Services
        /// tracker in the same write, see <see cref="StageServerTimeUpload"/>.
        /// </para>
        /// </summary>
-        public async Task<SessionSummary?> IncrementSessionEventCountAsync(string tenantId, string sessionId, int increment, DateTime? earliestEventTimestamp = null, DateTime? latestEventTimestamp = null, EnrollmentPhase? currentPhase = null, int platformScriptIncrement = 0, int remediationScriptIncrement = 0, int rebootIncrement = 0, IReadOnlyList<EnrollmentEvent>? upload = null)
+        public async Task<SessionIngestSnapshot?> IncrementSessionEventCountAsync(string tenantId, string sessionId, int increment, DateTime? earliestEventTimestamp = null, DateTime? latestEventTimestamp = null, EnrollmentPhase? currentPhase = null, int platformScriptIncrement = 0, int remediationScriptIncrement = 0, int rebootIncrement = 0, IReadOnlyList<EnrollmentEvent>? upload = null)
        {
            SecurityValidator.EnsureValidGuid(tenantId, nameof(tenantId));
            SecurityValidator.EnsureValidGuid(sessionId, nameof(sessionId));
@@ -2597,7 +2620,7 @@ namespace AutopilotMonitor.Functions.Services
                try
                {
                    var tableClient = _tableServiceClient.GetTableClient(Constants.TableNames.Sessions);
-                    var entityResponse = await tableClient.GetEntityIfExistsAsync<TableEntity>(tenantId, sessionId);
+                    var entityResponse = await tableClient.GetEntityIfExistsAsync<TableEntity>(tenantId, sessionId, select: SessionRowProjections.IngestSnapshot);
                    if (!entityResponse.HasValue)
                    {
                        _logger.LogWarning("Session {SessionId} not found when incrementing event count — session may have been cleaned up or not yet registered", sessionId);
@@ -2672,15 +2695,15 @@ namespace AutopilotMonitor.Functions.Services
                    // Dual-write: keep SessionsIndex in sync (StartedAt-shift → full upsert, else merge).
                    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.
+                    // Apply the merged fields onto the projected RMW read and map the ingest slice —
+                    // the post-merge snapshot the caller would otherwise re-read.
                    foreach (var kvp in update)
                    {
                        if (kvp.Key is "PartitionKey" or "RowKey" or "Timestamp" or "odata.etag")
                            continue;
                        entity[kvp.Key] = kvp.Value;
                    }
-                    return MapToSessionSummary(entity);
+                    return MapToIngestSnapshot(entity);
                }
                catch (Azure.RequestFailedException ex) when (ex.Status == 412)
                {
@@ -3550,6 +3573,43 @@ namespace AutopilotMonitor.Functions.Services
        // internal (not private) so UsageMetricsProjectionEquivalenceTests can pin that a Sessions
        // row carrying only UsageMetricsSessionProjection maps to the same usage-relevant fields
        // as a full row.
+        /// <summary>Stored Status (a string; "InProgress" when absent); Unknown, with a warning, when it does not parse.</summary>
+        private SessionStatus ParseStoredStatus(TableEntity entity, string sessionId)
+        {
+            var statusString = entity.GetString("Status") ?? "InProgress";
+            if (Enum.TryParse<SessionStatus>(statusString, ignoreCase: true, out var status))
+                return status;
+            _logger.LogWarning($"Failed to parse status '{statusString}' for session {sessionId}, defaulting to Unknown");
+            return SessionStatus.Unknown;
+        }
+
+        /// <summary>
+        /// The ingest slice of a Sessions row (<see cref="SessionIngestSnapshot"/>): read with
+        /// <see cref="SessionRowProjections.IngestSnapshot"/>, mapped exactly as
+        /// <see cref="MapToSessionSummary(TableEntity)"/> maps the same fields (pinned by
+        /// SessionRowProjectionTests).
+        /// </summary>
+        internal SessionIngestSnapshot MapToIngestSnapshot(TableEntity entity)
+        {
+            var startedAt = SafeGetDateTime(entity, "StartedAt") ?? DateTime.UtcNow;
+            var completedAt = SafeGetDateTime(entity, "CompletedAt");
+            var status = ParseStoredStatus(entity, entity.RowKey);
+            return new SessionIngestSnapshot(
+                status,
+                SafeGetInt32(entity, "CurrentPhase") ?? 0,
+                entity.GetString("CurrentPhaseDetail") ?? string.Empty,
+                entity.GetString("FailureReason") ?? string.Empty,
+                SafeGetInt32(entity, "EventCount") ?? 0,
+                ComputeEffectiveDuration(entity, status, startedAt, completedAt),
+                completedAt,
+                entity.GetString("DiagnosticsBlobName"),
+                entity.GetBoolean("IsPreProvisioned") ?? false,
+                entity.GetString("AdminMarkedAction"),
+                entity.GetString("PendingActionsJson") ?? string.Empty,
+                entity.GetString("AgentVersion") ?? string.Empty,
+                entity.GetString("ImeAgentVersion") ?? string.Empty);
+        }
+
        internal SessionSummary MapToSessionSummary(TableEntity entity)
            => MapToSessionSummary(entity, entity.RowKey);

@@ -3566,13 +3626,7 @@ namespace AutopilotMonitor.Functions.Services
            var startedAt = SafeGetDateTime(entity, "StartedAt") ?? DateTime.UtcNow;
            var completedAt = SafeGetDateTime(entity, "CompletedAt");

-            // Parse status with error handling and case-insensitivity
-            var statusString = entity.GetString("Status") ?? "InProgress";
-            if (!Enum.TryParse<SessionStatus>(statusString, ignoreCase: true, out var status))
-            {
-                _logger.LogWarning($"Failed to parse status '{statusString}' for session {sessionId}, defaulting to Unknown");
-                status = SessionStatus.Unknown;
-            }
+            var status = ParseStoredStatus(entity, sessionId);

            return new SessionSummary
            {
@@ -21,6 +21,13 @@ namespace AutopilotMonitor.Shared.DataAccess
        Task UpdateSessionOwnerAsync(string tenantId, string sessionId, SessionOwner owner);
        Task<SessionSummary?> GetSessionAsync(string tenantId, string sessionId);
        /// <summary>
+        /// The ingest slice of the session row (<see cref="SessionIngestSnapshot"/>), read projected
+        /// to its columns; null when the row is missing or unreadable. The signal-only upload reads
+        /// its control signals (AdminMarkedAction, pending ServerActions) through this instead of a
+        /// full <see cref="GetSessionAsync"/>.
+        /// </summary>
+        Task<SessionIngestSnapshot?> GetSessionIngestSnapshotAsync(string tenantId, string sessionId);
+        /// <summary>
        /// Resolves the tenantId owning <paramref name="sessionId"/> — point-read on the
        /// SessionTenantLookup table with a legacy SessionsIndex-scan fallback that self-heals
        /// the lookup row. Null when the session is unknown. Used for global-scope cross-tenant
@@ -111,14 +118,15 @@ namespace AutopilotMonitor.Shared.DataAccess
            string? failureSnapshotJson = null, bool allowTerminalReclassification = false,
            bool espSoftFailure = false, string? completionSource = null);
        /// <summary>
-        /// Increments per-session counters via read-modify-write. Returns the post-merge session
-        /// snapshot (the RMW read with the applied increments) so hot-path callers can skip a
-        /// follow-up <see cref="GetSessionAsync"/>; null when the row is missing or the update
-        /// could not be applied (caller falls back to its own read).
+        /// Increments per-session counters via read-modify-write. The read is projected to the
+        /// ingest slice of the row and its post-merge values come back as a
+        /// <see cref="SessionIngestSnapshot"/> (the RMW read with the applied increments) so
+        /// hot-path callers can skip a follow-up read; null when the row is missing or the update
+        /// could not be applied (caller falls back to its own <see cref="GetSessionAsync"/>).
        /// <paramref name="upload"/> (the batch's stored events) feeds the session's server-time
        /// tracker in the same write (StartedAtServer / ResumedAtServer while the session runs).
        /// </summary>
-        Task<SessionSummary?> IncrementSessionEventCountAsync(
+        Task<SessionIngestSnapshot?> IncrementSessionEventCountAsync(
            string tenantId, string sessionId, int increment,
            DateTime? earliestEventTimestamp = null, DateTime? latestEventTimestamp = null,
            EnrollmentPhase? currentPhase = null,
@@ -0,0 +1,62 @@
+using System;
+using AutopilotMonitor.Shared.Models;
+
+namespace AutopilotMonitor.Shared.DataAccess
+{
+    /// <summary>
+    /// The slice of a Sessions row that a non-terminal ingest batch reads: the counter merge
+    /// returns it as the post-merge state (SignalR delta, control signals to the agent, IME
+    /// pattern health, the shutdown close) and the signal-only upload reads it directly.
+    /// Deliberately not a <see cref="SessionSummary"/>: the hot path reads the row projected to
+    /// exactly these columns (read whole, the row's size — not the storage server — set the
+    /// ingest latency), so a summary built from that read would carry silent defaults in every
+    /// other field. Batches that need the whole row (terminal, status or WhiteGlove transition,
+    /// diagnostics upload) read a full summary and derive this via <see cref="From"/>.
+    /// </summary>
+    public sealed class SessionIngestSnapshot
+    {
+        public SessionIngestSnapshot(
+            SessionStatus status, int currentPhase, string currentPhaseDetail, string failureReason,
+            int eventCount, int? durationSeconds, DateTime? completedAt, string? diagnosticsBlobName,
+            bool isPreProvisioned, string? adminMarkedAction, string pendingActionsJson,
+            string agentVersion, string imeAgentVersion)
+        {
+            Status = status;
+            CurrentPhase = currentPhase;
+            CurrentPhaseDetail = currentPhaseDetail;
+            FailureReason = failureReason;
+            EventCount = eventCount;
+            DurationSeconds = durationSeconds;
+            CompletedAt = completedAt;
+            DiagnosticsBlobName = diagnosticsBlobName;
+            IsPreProvisioned = isPreProvisioned;
+            AdminMarkedAction = adminMarkedAction;
+            PendingActionsJson = pendingActionsJson;
+            AgentVersion = agentVersion;
+            ImeAgentVersion = imeAgentVersion;
+        }
+
+        public SessionStatus Status { get; }
+        public int CurrentPhase { get; }
+        public string CurrentPhaseDetail { get; }
+        public string FailureReason { get; }
+        public int EventCount { get; }
+        public int? DurationSeconds { get; }
+        public DateTime? CompletedAt { get; }
+        public string? DiagnosticsBlobName { get; }
+        public bool IsPreProvisioned { get; }
+        /// <summary>Portal-button signal to the agent (set only by the Mark-Succeeded/Failed functions).</summary>
+        public string? AdminMarkedAction { get; }
+        /// <summary>Queued ServerActions (JSON); empty when none are pending.</summary>
+        public string PendingActionsJson { get; }
+        public string AgentVersion { get; }
+        public string ImeAgentVersion { get; }
+
+        /// <summary>The same slice taken off a full summary (the paths that read the whole row).</summary>
+        public static SessionIngestSnapshot From(SessionSummary session) => new SessionIngestSnapshot(
+            session.Status, session.CurrentPhase, session.CurrentPhaseDetail, session.FailureReason,
+            session.EventCount, session.DurationSeconds, session.CompletedAt, session.DiagnosticsBlobName,
+            session.IsPreProvisioned, session.AdminMarkedAction, session.PendingActionsJson,
+            session.AgentVersion, session.ImeAgentVersion);
+    }
+}