fix(backend): cascade-delete codex follow-ups (F1 V2 guard + upsert-revival, F2 partial-restore key-level re-increment, F3 legacy delete sessions-first, F4 restore 400 EntityAlreadyExists)

F1 — V2 ingest writes were landing past the cascade lock. Wired SessionDeletionGuard.EnsureWritableAsync into IngestTelemetryFunction before PersistItemsAsync so Event / Signal / Transition writes are refused with 410 Gone when DeletionState is Preparing / Queued / Running / Poisoned. Switched UpdateSessionImeAgentVersionAsync from UpsertEntityAsync to UpdateEntityAsync(ETag.All, Merge) with 404 swallow so a tombstoned Sessions row can no longer be silently resurrected by an in-flight ingest. F2 — Partial-restore could leave SoftwareInventory at -N drift when the cascade poisoned mid-AGGREGATE-loop: per-key decrement was persisted into AggregateDecrementsApplied BEFORE the underlying decrement, but the gate used CompletedSteps.Contains(aggregate.Order) — which is only set AFTER the whole loop. Switched the gate to AggregateDecrementsApplied as the authoritative ledger and pass a filtered DeletionStep into RunInventoryReIncrementWithIdempotencyAsync so only the keys that were actually decremented are re-incremented. F3 — Legacy direct-delete path deleted Events / RuleResults / AppInstallSummaries before the ETag-CAS-guarded DeleteSessionAsync, leaving a TOCTOU window where a V2 producer could CAS-claim DeletionState=Preparing after the pre-read but before our side- table mutations — landing the producer's manifest snapshot on top of already-half-purged state. Reordered both admin (DeleteSessionFunction) and retention (SessionRetentionFanoutService) paths to Sessions-CAS first; side-tables only after the Sessions row is gone (no producer can claim a missing row). F4 — Restore per-row Add fallback only caught HTTP 409 for duplicate- row conflicts; Azurite and some legacy Azure deployments surface duplicate-row as 400 with ErrorCode=EntityAlreadyExists, breaking idempotent restore-retry after partial-failed restore. Centralised the check via IsAlreadyExistsStatus(rfe) helper accepting 409 OR 400 + ErrorCode=EntityAlreadyExists; non-duplicate 400s still propagate. Tests: - SessionRestoreServiceTests: 2 partial-restore tests refit to the new AggregateDecrementsApplied contract + new explicit Codex-scenario test (only-decremented-keys re-incremented, untouched key skipped). - SessionDeletionPr4cCodexFixesTests F4: harness setup adds full AggregateDecrementsApplied so the per-key idempotency assertion reflects the new gate. - SessionRetentionFanoutServiceTests: pre-read tests replaced by two tests covering "DeleteSessionAsync refuses → no side tables" and "DeleteSessionAsync succeeds → side tables run after". - TableStorageRestoreBatchTests: two new F4 tests — 400 EntityAlreadyExists treated as Skipped, 400 with non-duplicate ErrorCode still propagates. - UpdateSessionImeAgentVersionTests (NEW): asserts UpdateEntityAsync is used (not Upsert), 404 swallowed silently, other storage errors non-fatal. Suites: Backend xUnit 1510, Agent V1 530, Agent V2 1042 — all green. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
2026-10-24 20:53:06 +00:00 · 2026-05-13 10:44:54 +02:00
parent 4c676d076a
commit 921994e845
11 changed files with 410 additions and 132 deletions
@@ -226,10 +226,16 @@ public class SessionDeletionPr4cCodexFixesTests
    {
        var harness = new RestoreHarness();
        harness.SetPoisonedCascade();
-        // Simulate a prior partial restore that already re-incremented Office + Acrobat but
-        // crashed before the CAS Poisoned → None (the exact Codex F4 scenario).
-        var aggregateOrder = harness.Manifest.Steps.First(s => s.Class == DeletionStepClass.Aggregate).Order;
-        harness.Progress.CompletedSteps.Add(aggregateOrder);
+        // Scenario: forward cascade decremented all 3 keys → AggregateDecrementsApplied carries
+        // the full set (authoritative ledger). A prior partial-restore re-incremented Office +
+        // Acrobat (RestoreReIncrementsApplied) but crashed before CAS Poisoned → None. On retry,
+        // only Firefox should be re-incremented — the other two are already idempotent.
+        harness.Progress.AggregateDecrementsApplied = new HashSet<string>(StringComparer.Ordinal)
+        {
+            "Microsoft:Office:16.0",
+            "Adobe:Acrobat:23.1",
+            "Mozilla:Firefox:120.0",
+        };
        harness.Progress.RestoreReIncrementsApplied = new HashSet<string>(StringComparer.Ordinal)
        {
            "Microsoft:Office:16.0",
@@ -225,13 +225,14 @@ public class SessionRestoreServiceTests
    }

    [Fact]
-    public async Task Restore_partial_skips_inventory_re_increment_when_decrement_never_ran()
+    public async Task Restore_partial_skips_inventory_re_increment_when_no_decrements_applied()
    {
        var harness = new Harness();
        harness.SetPoisonedCascade();
-        // CompletedSteps does NOT include the AGGREGATE step's order (16).
+        // No per-key decrements were ever persisted. Restore must not blindly re-increment.
+        harness.Progress.AggregateDecrementsApplied = null;
        harness.Progress.CompletedSteps.Clear();
-        harness.Progress.CompletedSteps.Add(1);  // poisoned right after step 1
+        harness.Progress.CompletedSteps.Add(1);  // poisoned right after step 1, before AGGREGATE entry

        await harness.Sut.RestoreAsync(TenantId, SessionId, ManifestId, dryRun: false, actor: "ga@example.com");

@@ -241,12 +242,20 @@ public class SessionRestoreServiceTests
    }

    [Fact]
-    public async Task Restore_partial_re_increments_inventory_when_decrement_already_ran()
+    public async Task Restore_partial_re_increments_inventory_for_keys_in_AggregateDecrementsApplied()
    {
        var harness = new Harness();
        harness.SetPoisonedCascade();
-        // CompletedSteps includes the AGGREGATE step's order (16).
-        harness.Progress.CompletedSteps.UnionWith(new[] { 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15, 16 });
+        // The handler persists each key into AggregateDecrementsApplied BEFORE the underlying
+        // decrement runs. Restore must re-increment exactly those keys — even when the AGGREGATE
+        // step's Order is NOT yet in CompletedSteps (poison fell between two keys).
+        harness.Progress.CompletedSteps.UnionWith(new[] { 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15 });
+        harness.Progress.AggregateDecrementsApplied = new HashSet<string>(StringComparer.Ordinal)
+        {
+            "Microsoft:Office:16.0",
+            "Adobe:Acrobat:23.1",
+            "Mozilla:Firefox:120.0",
+        };

        await harness.Sut.RestoreAsync(TenantId, SessionId, ManifestId, dryRun: false, actor: "ga@example.com");

@@ -255,6 +264,35 @@ public class SessionRestoreServiceTests
            Times.Exactly(3));
    }

+    [Fact]
+    public async Task Restore_partial_re_increments_only_keys_that_were_actually_decremented()
+    {
+        // The Codex finding scenario: the cascade poisoned mid-AGGREGATE-loop. 2 of 3 keys had
+        // been persisted into AggregateDecrementsApplied (and their counters decremented) before
+        // the crash; the AGGREGATE step never reached CompletedSteps. Restore must re-increment
+        // only the 2 keys that were touched — re-incrementing the 3rd would create +1 drift.
+        var harness = new Harness();
+        harness.SetPoisonedCascade();
+        harness.Progress.AggregateDecrementsApplied = new HashSet<string>(StringComparer.Ordinal)
+        {
+            "Microsoft:Office:16.0",
+            "Adobe:Acrobat:23.1",
+            // "Mozilla:Firefox:120.0" was never reached
+        };
+
+        await harness.Sut.RestoreAsync(TenantId, SessionId, ManifestId, dryRun: false, actor: "ga@example.com");
+
+        harness.Storage.Verify(s => s.RestoreSoftwareInventoryEntryByKeyAsync(
+            TenantId, It.Is<DeletionDecrementKey>(k => k.Vendor == "Microsoft" && k.Name == "Office"),
+            It.IsAny<CancellationToken>()), Times.Once);
+        harness.Storage.Verify(s => s.RestoreSoftwareInventoryEntryByKeyAsync(
+            TenantId, It.Is<DeletionDecrementKey>(k => k.Vendor == "Adobe" && k.Name == "Acrobat"),
+            It.IsAny<CancellationToken>()), Times.Once);
+        harness.Storage.Verify(s => s.RestoreSoftwareInventoryEntryByKeyAsync(
+            TenantId, It.Is<DeletionDecrementKey>(k => k.Vendor == "Mozilla" && k.Name == "Firefox"),
+            It.IsAny<CancellationToken>()), Times.Never);
+    }
+
    [Theory]
    [InlineData(SessionDeletionState.Running)]
    [InlineData(SessionDeletionState.Preparing)]
@@ -139,50 +139,49 @@ public class SessionRetentionFanoutServiceTests
    }

    // ────────────────────────────────────────────────────────────────────────── PR6 follow-up F1 ─
+    // Codex-followup: the ordering changed from "side-tables first, Sessions-CAS last" to
+    // "Sessions-CAS first, side-tables after". DeleteSessionAsync's internal ETag-CAS is now the
+    // single arbiter against in-flight V2 cascades — if it returns false (locked / 412 / 404),
+    // side-tables stay untouched. This closes the TOCTOU window the pre-read couldn't.

    [Fact]
-    public async Task RunAsync_legacy_path_skips_locked_session_WITHOUT_touching_side_tables()
+    public async Task RunAsync_legacy_path_skips_side_tables_when_DeleteSessionAsync_refuses_locked_row()
    {
-        // PR6 follow-up F1: pre-read DeletionState before any side-table mutation. Without this
-        // gate the legacy path would delete Events / RuleResults / AppSummaries and only discover
-        // the V2 cascade lock at the tombstone CAS — leaving the V2 cascade's manifest claiming
-        // rows that were silently removed in the meantime.
+        // DeleteSessionAsync detects the V2 cascade lock internally and returns false. The fanout
+        // must record SessionsSkipped + leave side-tables intact so the V2 worker owns cleanup.
        var harness = new Harness();
        harness.WithTenant(TenantA, retentionDays: 30, v2: false, sessions: new[] { Old("s1", 60) });
-        // Override the default fresh-read: this session is locked by a V2 cascade.
-        harness.SessionRepo.Setup(r => r.GetSessionAsync(TenantA, "s1"))
-            .ReturnsAsync(new SessionSummary
-            {
-                TenantId = TenantA,
-                SessionId = "s1",
-                DeletionState = SessionDeletionState.Queued,
-                PendingDeletionManifestId = "EXISTING-MANIFEST",
-            });
+        harness.SessionRepo.Setup(r => r.DeleteSessionAsync(TenantA, "s1")).ReturnsAsync(false);

        var result = await harness.Sut.RunAsync(CancellationToken.None);

        Assert.Equal(1, result.SessionsSkipped);
        Assert.Equal(0, result.SessionsLegacyDeleted);
-        // Critical assertion: NONE of the side-table delete helpers were invoked.
+        harness.SessionRepo.Verify(r => r.DeleteSessionAsync(TenantA, "s1"), Times.Once);
+        // Side-tables MUST stay intact when DeleteSessionAsync refused.
        harness.MaintenanceRepo.Verify(m => m.DeleteSessionEventsAsync(It.IsAny<string>(), It.IsAny<string>()), Times.Never);
        harness.MaintenanceRepo.Verify(m => m.DeleteSessionRuleResultsAsync(It.IsAny<string>(), It.IsAny<string>()), Times.Never);
        harness.MaintenanceRepo.Verify(m => m.DeleteSessionAppInstallSummariesAsync(It.IsAny<string>(), It.IsAny<string>()), Times.Never);
-        harness.SessionRepo.Verify(r => r.DeleteSessionAsync(It.IsAny<string>(), It.IsAny<string>()), Times.Never);
    }

    [Fact]
-    public async Task RunAsync_legacy_path_skips_when_session_already_removed_during_preread()
+    public async Task RunAsync_legacy_path_runs_side_tables_only_after_DeleteSessionAsync_succeeds()
    {
-        // Edge case: row disappeared between GetSessionsOlderThan and the pre-read. Treat as skip
-        // (no side-table work needed); no error.
+        // Sessions-first ordering: side-table helpers fire ONLY after DeleteSessionAsync returns
+        // true. No producer can claim a missing Sessions row, so side-table cleanup is race-free.
        var harness = new Harness();
        harness.WithTenant(TenantA, retentionDays: 30, v2: false, sessions: new[] { Old("s1", 60) });
-        harness.SessionRepo.Setup(r => r.GetSessionAsync(TenantA, "s1")).ReturnsAsync((SessionSummary?)null);
+        harness.SessionRepo.Setup(r => r.DeleteSessionAsync(TenantA, "s1")).ReturnsAsync(true);
+        harness.MaintenanceRepo.Setup(m => m.DeleteSessionEventsAsync(TenantA, "s1")).ReturnsAsync(7);

        var result = await harness.Sut.RunAsync(CancellationToken.None);

-        Assert.Equal(1, result.SessionsSkipped);
-        harness.MaintenanceRepo.Verify(m => m.DeleteSessionEventsAsync(It.IsAny<string>(), It.IsAny<string>()), Times.Never);
+        Assert.Equal(1, result.SessionsLegacyDeleted);
+        Assert.Equal(0, result.SessionsSkipped);
+        harness.SessionRepo.Verify(r => r.DeleteSessionAsync(TenantA, "s1"), Times.Once);
+        harness.MaintenanceRepo.Verify(m => m.DeleteSessionEventsAsync(TenantA, "s1"), Times.Once);
+        harness.MaintenanceRepo.Verify(m => m.DeleteSessionRuleResultsAsync(TenantA, "s1"), Times.Once);
+        harness.MaintenanceRepo.Verify(m => m.DeleteSessionAppInstallSummariesAsync(TenantA, "s1"), Times.Once);
    }

    // ────────────────────────────────────────────────────────────────────────── PR6 follow-up F2 ─
@@ -130,6 +130,47 @@ public class TableStorageRestoreBatchTests
                TableName, new[] { MakeDump("pkA", "rk_0") }, RestoreMode.Full));
    }

+    [Fact]
+    public async Task Restore_perRow_treats_400_EntityAlreadyExists_as_Skipped()
+    {
+        // Codex-followup F4: Azurite and some legacy Azure deployments map duplicate-row conflict
+        // to HTTP 400 (ErrorCode=EntityAlreadyExists) instead of 409. The batch catch already
+        // tolerated 400, but the per-row fallback's `status == 409`-only filter rethrew the 400,
+        // breaking idempotent restore-retry after partial-failed restore.
+        var harness = new Harness
+        {
+            BatchBehavior = _ => throw new RequestFailedException(400, "EntityAlreadyExists", "EntityAlreadyExists", innerException: null),
+            PerRowAddBehavior = _ => throw new RequestFailedException(400, "EntityAlreadyExists", "EntityAlreadyExists", innerException: null),
+        };
+
+        var result = await harness.Sut.RestoreRowsByExactKeysInBatchesAsync(
+            TableName, new[] { MakeDump("pkA", "rk_dup") }, RestoreMode.Partial);
+
+        Assert.Equal(1, result.Attempted);
+        Assert.Equal(0, result.Restored);
+        Assert.Equal(1, result.Skipped);
+    }
+
+    [Fact]
+    public async Task Restore_perRow_400_without_EntityAlreadyExists_propagates()
+    {
+        // Negative case: a 400 that is NOT a duplicate-row conflict (e.g. validation error from
+        // an oversized property) must still propagate. The new IsAlreadyExistsStatus helper keys
+        // off ErrorCode = "EntityAlreadyExists" to distinguish.
+        var harness = new Harness
+        {
+            BatchBehavior = _ => throw new RequestFailedException(400, "PropertyValueTooLarge", "PropertyValueTooLarge", innerException: null),
+            PerRowAddBehavior = _ => throw new RequestFailedException(400, "PropertyValueTooLarge", "PropertyValueTooLarge", innerException: null),
+        };
+
+        var ex = await Assert.ThrowsAsync<RequestFailedException>(() =>
+            harness.Sut.RestoreRowsByExactKeysInBatchesAsync(
+                TableName, new[] { MakeDump("pkA", "rk_x") }, RestoreMode.Partial));
+
+        Assert.Equal(400, ex.Status);
+        Assert.Equal("PropertyValueTooLarge", ex.ErrorCode);
+    }
+
    [Fact]
    public async Task Restore_throws_on_null_args()
    {
@@ -0,0 +1,95 @@
+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 Microsoft.Extensions.Logging.Abstractions;
+using Moq;
+using Xunit;
+
+namespace AutopilotMonitor.Functions.Tests;
+
+/// <summary>
+/// Codex-followup F1 (sub-finding): <c>UpdateSessionImeAgentVersionAsync</c> previously used
+/// <c>UpsertEntityAsync</c>, which would silently recreate a partial Sessions row after the
+/// cascade-delete worker had tombstoned it — breaking the manifest-snapshot invariant. The fix
+/// switches to <c>UpdateEntityAsync(entity, ETag.All, Merge)</c>: 404 means "row already gone"
+/// and is the correct no-op outcome.
+/// </summary>
+public class UpdateSessionImeAgentVersionTests
+{
+    private const string TenantId  = "11111111-1111-1111-1111-111111111111";
+    private const string SessionId = "22222222-2222-2222-2222-222222222222";
+
+    [Fact]
+    public async Task UpdateSessionImeAgentVersionAsync_uses_UpdateEntityAsync_not_UpsertEntityAsync()
+    {
+        var harness = new Harness();
+        harness.Sessions
+            .Setup(t => t.UpdateEntityAsync(It.IsAny<ITableEntity>(), It.IsAny<ETag>(), It.IsAny<TableUpdateMode>(), It.IsAny<CancellationToken>()))
+            .ReturnsAsync(Mock.Of<Response>());
+
+        await harness.Sut.UpdateSessionImeAgentVersionAsync(TenantId, SessionId, "1.2.3.4");
+
+        harness.Sessions.Verify(t => t.UpdateEntityAsync(
+            It.Is<ITableEntity>(e => e.PartitionKey == TenantId && e.RowKey == SessionId),
+            It.IsAny<ETag>(),
+            TableUpdateMode.Merge,
+            It.IsAny<CancellationToken>()),
+            Times.Once);
+        // The critical invariant: the old Upsert path is GONE — a tombstoned row must not be
+        // resurrected by an in-flight ingest landing past the cascade lock.
+        harness.Sessions.Verify(t => t.UpsertEntityAsync(
+            It.IsAny<ITableEntity>(), It.IsAny<TableUpdateMode>(), It.IsAny<CancellationToken>()),
+            Times.Never);
+    }
+
+    [Fact]
+    public async Task UpdateSessionImeAgentVersionAsync_swallows_404_when_row_is_tombstoned()
+    {
+        // The exact tombstone-revival scenario: cascade worker removed the Sessions row, an
+        // in-flight ingest then tries to stamp ImeAgentVersion. UpdateEntityAsync surfaces 404 →
+        // helper must return silently, NOT recreate the row.
+        var harness = new Harness();
+        harness.Sessions
+            .Setup(t => t.UpdateEntityAsync(It.IsAny<ITableEntity>(), It.IsAny<ETag>(), It.IsAny<TableUpdateMode>(), It.IsAny<CancellationToken>()))
+            .ThrowsAsync(new RequestFailedException(404, "Not Found"));
+
+        await harness.Sut.UpdateSessionImeAgentVersionAsync(TenantId, SessionId, "1.2.3.4");
+
+        // No Upsert anywhere — 404 is the silent success path.
+        harness.Sessions.Verify(t => t.UpsertEntityAsync(
+            It.IsAny<ITableEntity>(), It.IsAny<TableUpdateMode>(), It.IsAny<CancellationToken>()),
+            Times.Never);
+    }
+
+    [Fact]
+    public async Task UpdateSessionImeAgentVersionAsync_swallows_other_storage_exceptions_as_warning()
+    {
+        // Method contract: failures are non-fatal. A 503 / 500 from storage logs a warning but
+        // does not throw — preserves the pre-fix "don't block ingest" behaviour.
+        var harness = new Harness();
+        harness.Sessions
+            .Setup(t => t.UpdateEntityAsync(It.IsAny<ITableEntity>(), It.IsAny<ETag>(), It.IsAny<TableUpdateMode>(), It.IsAny<CancellationToken>()))
+            .ThrowsAsync(new RequestFailedException(503, "ServiceUnavailable"));
+
+        await harness.Sut.UpdateSessionImeAgentVersionAsync(TenantId, SessionId, "1.2.3.4");
+        // No exception thrown.
+    }
+
+    private sealed class Harness
+    {
+        public Mock<TableClient> Sessions { get; }
+        public TableStorageService Sut { get; }
+
+        public Harness()
+        {
+            Sessions = new Mock<TableClient>();
+            var mockServiceClient = new Mock<TableServiceClient>();
+            mockServiceClient.Setup(s => s.GetTableClient(Constants.TableNames.Sessions)).Returns(Sessions.Object);
+            Sut = new TableStorageService(mockServiceClient.Object, NullLogger<TableStorageService>.Instance);
+        }
+    }
+}
@@ -2,6 +2,7 @@ using System.IO;
 using System.Net;
 using AutopilotMonitor.Functions.Security;
 using AutopilotMonitor.Functions.Services;
+using AutopilotMonitor.Functions.Services.Deletion;
 using AutopilotMonitor.Functions.Services.Indexing;
 using AutopilotMonitor.Shared.DataAccess;
 using AutopilotMonitor.Shared.Models;
@@ -52,6 +53,7 @@ namespace AutopilotMonitor.Functions.Functions.Ingest
        private readonly BootstrapSessionService _bootstrapSessionService;
        private readonly BlockedDeviceService _blockedDeviceService;
        private readonly BlockedVersionService _blockedVersionService;
+        private readonly SessionDeletionGuard _deletionGuard;

        public IngestTelemetryFunction(
            ILogger<IngestTelemetryFunction> logger,
@@ -67,7 +69,8 @@ namespace AutopilotMonitor.Functions.Functions.Ingest
            DeviceAssociationValidator deviceAssociationValidator,
            BootstrapSessionService bootstrapSessionService,
            BlockedDeviceService blockedDeviceService,
-            BlockedVersionService blockedVersionService)
+            BlockedVersionService blockedVersionService,
+            SessionDeletionGuard deletionGuard)
        {
            _logger = logger;
            _sessionRepo = sessionRepo;
@@ -83,6 +86,7 @@ namespace AutopilotMonitor.Functions.Functions.Ingest
            _bootstrapSessionService = bootstrapSessionService;
            _blockedDeviceService = blockedDeviceService;
            _blockedVersionService = blockedVersionService;
+            _deletionGuard = deletionGuard;
        }

        [Function("IngestTelemetry")]
@@ -190,6 +194,24 @@ namespace AutopilotMonitor.Functions.Functions.Ingest
                        "All telemetry items in a batch must share the same PartitionKey"));
                }

+                // Cascade-delete guard: refuse the batch with 410 Gone when a V2 cascade owns the
+                // Sessions row (states Preparing/Queued/Running/Poisoned). Without this check, the
+                // hot-path writers below (StoreEventsBatchAsync, signal/transition StoreBatchAsync,
+                // UpdateSessionImeAgentVersionAsync upsert, …) would land rows past the lock and
+                // leave orphan data the manifest cannot describe. One read per batch; absent
+                // Sessions row → silent pass (caller handles session-not-found in its own write).
+                try
+                {
+                    await _deletionGuard.EnsureWritableAsync(bodyTenantId, sessionId, "V2.IngestTelemetry");
+                }
+                catch (SessionDeletionLockedException locked)
+                {
+                    _logger.LogInformation(
+                        "IngestTelemetry: refused batch — cascade in flight tenant={Tenant} session={Session} state={State} manifestId={ManifestId}",
+                        bodyTenantId, sessionId, locked.CurrentState, locked.ManifestId);
+                    return AsOutput(await WriteSessionLockedAsync(req, locked));
+                }
+
                // Partition + persist. Events are routed through EventIngestProcessor which runs the
                // full pipeline (rule engine / app-install aggregation / SignalR / webhooks / ...);
                // Signal + Transition go straight to their repositories.
@@ -294,6 +316,26 @@ namespace AutopilotMonitor.Functions.Functions.Ingest
            return response;
        }

+        /// <summary>
+        /// Responds 410 Gone when the V2 cascade-delete guard refuses the batch. Per plan §5 PR3
+        /// wiring table, telemetry ingest of a locked session is a terminal condition for the
+        /// agent: the session is being torn down server-side and any further writes would create
+        /// orphan rows. 410 is the documented status; the body shape mirrors the device-blocked
+        /// response so existing agent code paths (which already short-circuit on Success=false)
+        /// drop the batch without retry.
+        /// </summary>
+        private static async Task<HttpResponseData> WriteSessionLockedAsync(HttpRequestData req, SessionDeletionLockedException locked)
+        {
+            var response = req.CreateResponse(HttpStatusCode.Gone);
+            await response.WriteAsJsonAsync(new IngestEventsResponse
+            {
+                Success = false,
+                Message = $"Session is being deleted by an administrator (state={locked.CurrentState}); further telemetry will be rejected.",
+                ProcessedAt = DateTime.UtcNow,
+            });
+            return response;
+        }
+
        /// <summary>
        /// Partitions the incoming batch by <see cref="TelemetryItemDto.Kind"/> and persists each
        /// kind through its destination path. Events go through <see cref="EventIngestProcessor"/>
@@ -245,12 +245,68 @@ namespace AutopilotMonitor.Functions.Functions.Sessions
        };

        /// <summary>
-        /// Legacy direct-delete path — preserved unchanged for the flag-OFF default. Removed by
+        /// Legacy direct-delete path — preserved for the flag-OFF default. Removed by
        /// plan §5 PR7 once V2 has been stable on all tenants for at least two weeks.
+        /// <para>
+        /// <b>Ordering</b> (Sessions-first): the ETag-CAS-guarded <c>DeleteSessionAsync</c> runs
+        /// <i>before</i> any side-table delete. This closes the TOCTOU window where a V2
+        /// cascade producer could CAS-claim <c>DeletionState=Preparing</c> after our pre-read
+        /// but before our side-table mutations — leaving the producer's manifest snapshot
+        /// inconsistent against an already-half-purged side-table state. Side-table cleanup
+        /// after the Sessions row is tombstoned is safe: no producer can claim a missing row.
+        /// </para>
        /// </summary>
        private async Task<HttpResponseData> RunLegacyDirectDeletePathAsync(
            HttpRequestData req, string tenantId, string sessionId, string userIdentifier)
        {
+            // Sessions-first: tombstone (and release the writer-lock semantics) before any
+            // side-table mutation. DeleteSessionAsync is ETag-CAS-guarded (PR5 finding 2)
+            // against in-flight V2 cascades.
+            var sessionDeleted = await _sessionRepo.DeleteSessionAsync(tenantId, sessionId);
+
+            if (!sessionDeleted)
+            {
+                // DeleteSessionAsync returned false. Either 404 (already gone) or a concurrent
+                // V2 producer CAS-claimed the row (ETag 412 / non-None DeletionState). Re-read
+                // to disambiguate and pick the right status code.
+                var refreshed = await _sessionRepo.GetSessionAsync(tenantId, sessionId);
+                if (refreshed == null)
+                {
+                    _logger.LogWarning($"Session {sessionId} not found at final delete (race with concurrent caller)");
+                    return await WriteResponse(req, HttpStatusCode.NotFound, new
+                    {
+                        success = false,
+                        message = $"Session {sessionId} not found",
+                    });
+                }
+                if (SessionDeletionState.IsLocked(refreshed.DeletionState))
+                {
+                    _logger.LogWarning(
+                        "Legacy delete refused: V2 cascade owns the session. " +
+                        "tenant={TenantId} session={SessionId} state={State}",
+                        tenantId, sessionId, refreshed.DeletionState);
+                    var poisonedHint = string.Equals(refreshed.DeletionState, SessionDeletionState.Poisoned, StringComparison.Ordinal);
+                    return await WriteResponse(req, HttpStatusCode.Conflict, new
+                    {
+                        success = false,
+                        message = poisonedHint
+                            ? "Cascade is poisoned; recover via POST /api/admin/sessions/{id}/restore before retrying delete."
+                            : "A V2 cascade owns this session; retry later or use the cascade-delete path.",
+                        deletionState = refreshed.DeletionState,
+                        manifestId = refreshed.PendingDeletionManifestId,
+                        hint = poisonedHint ? "cascade_poisoned_use_restore" : "cascade_already_in_flight",
+                    });
+                }
+                _logger.LogError($"Legacy delete failed for session {sessionId} without an identifiable race cause");
+                return await WriteResponse(req, HttpStatusCode.InternalServerError, new
+                {
+                    success = false,
+                    message = "Internal server error",
+                });
+            }
+
+            // Sessions row is gone — no V2 producer can claim it now. Safe to cascade-clean
+            // side tables without ordering races.
            var eventsDeleted = await _maintenanceRepo.DeleteSessionEventsAsync(tenantId, sessionId);
            _logger.LogInformation($"Deleted {eventsDeleted} events for session {sessionId}");

@@ -260,72 +316,27 @@ namespace AutopilotMonitor.Functions.Functions.Sessions
            var appSummariesDeleted = await _maintenanceRepo.DeleteSessionAppInstallSummariesAsync(tenantId, sessionId);
            _logger.LogInformation($"Deleted {appSummariesDeleted} app install summaries for session {sessionId}");

-            var sessionDeleted = await _sessionRepo.DeleteSessionAsync(tenantId, sessionId);
-
-            if (sessionDeleted)
-            {
-                await _maintenanceRepo.LogAuditEntryAsync(
-                    tenantId,
-                    "DELETE",
-                    "Session",
-                    sessionId,
-                    userIdentifier,
-                    new Dictionary<string, string>
-                    {
-                        { "EventsDeleted", eventsDeleted.ToString() },
-                        { "RuleResultsDeleted", ruleResultsDeleted.ToString() },
-                        { "AppInstallSummariesDeleted", appSummariesDeleted.ToString() }
-                    });
-
-                _logger.LogInformation($"Successfully deleted session {sessionId}");
-                return await WriteResponse(req, HttpStatusCode.OK, new
+            await _maintenanceRepo.LogAuditEntryAsync(
+                tenantId,
+                "DELETE",
+                "Session",
+                sessionId,
+                userIdentifier,
+                new Dictionary<string, string>
                {
-                    success = true,
-                    message = $"Session {sessionId} deleted successfully",
-                    eventsDeleted,
-                    ruleResultsDeleted,
-                    appInstallSummariesDeleted = appSummariesDeleted,
+                    { "EventsDeleted", eventsDeleted.ToString() },
+                    { "RuleResultsDeleted", ruleResultsDeleted.ToString() },
+                    { "AppInstallSummariesDeleted", appSummariesDeleted.ToString() }
                });
-            }

-            // DeleteSessionAsync returned false. PR5 finding 2: this is no longer just a 404
-            // race — it can also mean a concurrent V2 producer CAS-claimed the row (ETag 412)
-            // or the row's DeletionState is now non-None. Re-read to disambiguate and pick the
-            // right status code so the UI shows the right message.
-            var refreshed = await _sessionRepo.GetSessionAsync(tenantId, sessionId);
-            if (refreshed == null)
+            _logger.LogInformation($"Successfully deleted session {sessionId}");
+            return await WriteResponse(req, HttpStatusCode.OK, new
            {
-                _logger.LogWarning($"Session {sessionId} not found at final delete (race with concurrent caller)");
-                return await WriteResponse(req, HttpStatusCode.NotFound, new
-                {
-                    success = false,
-                    message = $"Session {sessionId} not found",
-                });
-            }
-            if (SessionDeletionState.IsLocked(refreshed.DeletionState))
-            {
-                _logger.LogWarning(
-                    "Legacy delete refused: V2 cascade claimed the session between pre-read and tombstone. " +
-                    "tenant={TenantId} session={SessionId} state={State}",
-                    tenantId, sessionId, refreshed.DeletionState);
-                var poisonedHint = string.Equals(refreshed.DeletionState, SessionDeletionState.Poisoned, StringComparison.Ordinal);
-                return await WriteResponse(req, HttpStatusCode.Conflict, new
-                {
-                    success = false,
-                    message = poisonedHint
-                        ? "Cascade is poisoned; recover via POST /api/admin/sessions/{id}/restore before retrying delete."
-                        : "A V2 cascade claimed this session while the legacy delete was running; retry later.",
-                    deletionState = refreshed.DeletionState,
-                    manifestId = refreshed.PendingDeletionManifestId,
-                    hint = poisonedHint ? "cascade_poisoned_use_restore" : "cascade_already_in_flight",
-                });
-            }
-            // Row still exists with DeletionState=None but the delete failed — generic error.
-            _logger.LogError($"Legacy delete failed for session {sessionId} without an identifiable race cause");
-            return await WriteResponse(req, HttpStatusCode.InternalServerError, new
-            {
-                success = false,
-                message = "Internal server error",
+                success = true,
+                message = $"Session {sessionId} deleted successfully",
+                eventsDeleted,
+                ruleResultsDeleted,
+                appInstallSummariesDeleted = appSummariesDeleted,
            });
        }

@@ -346,11 +346,13 @@ namespace AutopilotMonitor.Functions.Services.Deletion
            DeletionManifest manifest, DeletionProgress progress, string progressEtag,
            SessionRestoreResult result, CancellationToken ct)
        {
-            // Determine whether the AGGREGATE decrement step had already run before the cascade
-            // poisoned. If yes → re-increment to undo it. If no → leave counters alone (avoid
-            // double +1).
-            var aggregate = manifest.Steps.FirstOrDefault(s => s.Class == DeletionStepClass.Aggregate);
-            var decrementAlreadyRan = aggregate != null && progress.CompletedSteps.Contains(aggregate.Order);
+            // Re-increment policy in partial-restore: the AGGREGATE step's CompletedSteps marker
+            // is set AFTER the per-key loop in SessionDeletionHandler. If the cascade poisoned
+            // mid-loop, some keys are in AggregateDecrementsApplied (and were actually
+            // decremented) but the step is NOT in CompletedSteps — using CompletedSteps as the
+            // gate would skip the entire re-increment block and leave -N permanent drift.
+            // Authoritative truth is AggregateDecrementsApplied (handler persists it BEFORE the
+            // decrement call) — re-increment exactly those keys.

            // Pass 1: every non-FINAL step in REVERSE cascade order.
            foreach (var step in manifest.Steps
@@ -362,10 +364,31 @@ namespace AutopilotMonitor.Functions.Services.Deletion
                switch (step.Class)
                {
                    case DeletionStepClass.Aggregate:
-                        if (!decrementAlreadyRan) break;
+                    {
+                        var appliedKeys = progress.AggregateDecrementsApplied;
+                        if (step.Decrements == null || step.Decrements.Count == 0
+                            || appliedKeys == null || appliedKeys.Count == 0)
+                        {
+                            break;
+                        }
+
+                        var decrementsToReverse = step.Decrements
+                            .Where(k => appliedKeys.Contains(BuildAggregateCompositeKey(k)))
+                            .ToList();
+                        if (decrementsToReverse.Count == 0) break;
+
+                        var partialStep = new DeletionStep
+                        {
+                            Order = step.Order,
+                            Table = step.Table,
+                            Step = step.Step,
+                            Class = step.Class,
+                            Decrements = decrementsToReverse,
+                        };
                        (progress, progressEtag) = await RunInventoryReIncrementWithIdempotencyAsync(
-                            tenantId, sessionId, manifestId, step, progress, progressEtag, result, ct).ConfigureAwait(false);
+                            tenantId, sessionId, manifestId, partialStep, progress, progressEtag, result, ct).ConfigureAwait(false);
                        break;
+                    }

                    case DeletionStepClass.PkBySession:
                    case DeletionStepClass.PkRkExact:
@@ -307,41 +307,29 @@ namespace AutopilotMonitor.Functions.Services.Deletion
            var tenantId = session.TenantId;
            var sessionId = session.SessionId;

-            // F1: gate side-table mutations on DeletionState. The CAS guard inside DeleteSessionAsync
-            // is the storage-atomic backstop (PR5 finding 2), but it fires AFTER the side-table
-            // deletes. By the time it returns false we've already orphaned events from the V2
-            // cascade's planned manifest. Reading the Sessions row up-front is cheap (one Get) and
-            // skips ~3 side-table deletes when the lock is held.
-            var refreshed = await _sessionRepo.GetSessionAsync(tenantId, sessionId).ConfigureAwait(false);
-            if (refreshed == null)
-            {
-                // Row already gone — nothing to clean up. Treat as skip (no events were ours to delete either).
-                return false;
-            }
-            if (SessionDeletionState.IsLocked(refreshed.DeletionState))
+            // Sessions-first: the ETag-CAS-guarded DeleteSessionAsync (PR5 finding 2) is the lock
+            // arbiter against in-flight V2 cascades. Running it BEFORE side-table mutations means
+            // a concurrent V2 producer either claims the row first (we get false → skip side
+            // tables, V2 owns the cascade) or we win (Sessions gone → no producer can claim →
+            // side-tables safe to cascade-clean). The previous side-tables-first ordering left
+            // a TOCTOU window where V2's manifest snapshot landed on top of an already-half-
+            // purged side-table state.
+            var deleted = await _sessionRepo.DeleteSessionAsync(tenantId, sessionId).ConfigureAwait(false);
+            if (!deleted)
            {
                _logger.LogInformation(
-                    "Retention fanout (legacy): session {SessionId} is locked by a V2 cascade (state={State}, manifestId={ManifestId}) — skipping ALL deletes, V2 worker owns the cascade",
-                    sessionId, refreshed.DeletionState, refreshed.PendingDeletionManifestId);
+                    "Retention fanout (legacy): session {SessionId} not tombstoned — either already gone or owned by a V2 cascade. Side-tables left intact; V2 worker (or next retention run) will absorb",
+                    sessionId);
                return false;
            }

+            // Sessions row is gone — V2 producers cannot claim a missing row. Side-table cleanup
+            // is now race-free.
            _ = await _maintenanceRepo.DeleteSessionEventsAsync(tenantId, sessionId).ConfigureAwait(false);
            _ = await _maintenanceRepo.DeleteSessionRuleResultsAsync(tenantId, sessionId).ConfigureAwait(false);
            _ = await _maintenanceRepo.DeleteSessionAppInstallSummariesAsync(tenantId, sessionId).ConfigureAwait(false);

-            // CAS guard inside DeleteSessionAsync (PR5 finding 2) refuses if DeletionState is locked
-            // — backstop for the T1-T3 race after our pre-read above. A `false` here means either
-            // 404 (already gone) is treated as success internally, OR a concurrent V2 cascade
-            // claimed the row in the brief window between our pre-read and the tombstone.
-            var deleted = await _sessionRepo.DeleteSessionAsync(tenantId, sessionId).ConfigureAwait(false);
-            if (!deleted)
-            {
-                _logger.LogWarning(
-                    "Retention fanout (legacy): session {SessionId} could not be tombstoned — likely claimed by a concurrent V2 cascade. Side tables already deleted; cascade is idempotent so V2 will absorb the difference. Will retry next run",
-                    sessionId);
-            }
-            return deleted;
+            return true;
        }
    }
 }
@@ -119,7 +119,7 @@ namespace AutopilotMonitor.Functions.Services
                        await tableClient.SubmitTransactionAsync(actions, cancellationToken);
                        restored += chunk.Count;
                    }
-                    catch (RequestFailedException ex) when (ex.Status == 409 || ex.Status == 400)
+                    catch (RequestFailedException ex) when (IsAlreadyExistsStatus(ex))
                    {
                        // Azure rolls back the entire transaction when any Add hits an existing
                        // row. Fall back to per-row AddEntity so we can distinguish "row already
@@ -138,8 +138,14 @@ namespace AutopilotMonitor.Functions.Services
                                await tableClient.AddEntityAsync(ConvertDumpToEntity(dump, tableName), cancellationToken);
                                restored++;
                            }
-                            catch (RequestFailedException rfe) when (rfe.Status == 409)
+                            catch (RequestFailedException rfe) when (IsAlreadyExistsStatus(rfe))
                            {
+                                // Mirror the batch-level relaxation: some Azure deployments and
+                                // the Azurite emulator surface duplicate-row conflict as HTTP 400
+                                // with ErrorCode=EntityAlreadyExists instead of 409. Counting only
+                                // 409 here left the per-row fallback non-idempotent on those
+                                // backends — restore-after-partial would throw on the leftover
+                                // rows that the previous attempt had already inserted.
                                skipped++;
                            }
                        }
@@ -242,5 +248,20 @@ namespace AutopilotMonitor.Functions.Services
        // Azure Tables batch transactions cap at 100 actions per submission, all sharing a
        // PartitionKey. Mirrors the const in the deletion partial.
        private const int RestoreBatchActionLimit = 100;
+
+        /// <summary>
+        /// True when the Azure SDK exception encodes a "row already exists" outcome — the only
+        /// case where we treat the failed Add as Skipped during restore. Production Azure Tables
+        /// reports this as HTTP 409 Conflict (ErrorCode <c>EntityAlreadyExists</c>), but the
+        /// Azurite emulator and some older service versions report HTTP 400 Bad Request with the
+        /// same ErrorCode. Anything else propagates so we don't silently swallow genuine errors.
+        /// </summary>
+        private static bool IsAlreadyExistsStatus(RequestFailedException ex)
+        {
+            if (ex.Status == 409) return true;
+            if (ex.Status == 400 && string.Equals(ex.ErrorCode, "EntityAlreadyExists", StringComparison.Ordinal))
+                return true;
+            return false;
+        }
    }
 }
@@ -2000,6 +2000,14 @@ namespace AutopilotMonitor.Functions.Services
        /// <summary>
        /// Stores the IME agent version on an existing session (Merge-mode, single field update).
        /// Non-fatal: failures are logged as warnings and do not block ingest.
+        /// <para>
+        /// Uses <see cref="TableClient.UpdateEntityAsync(ITableEntity, ETag, TableUpdateMode, System.Threading.CancellationToken)"/>
+        /// with <see cref="ETag.All"/>, not <c>UpsertEntityAsync</c>: a tombstoned Sessions row
+        /// must stay tombstoned. The previous Upsert would silently recreate a partial Sessions
+        /// row (PK/RK + ImeAgentVersion only) after the cascade-delete worker had removed it,
+        /// breaking the manifest-snapshot invariant. 404 here means "row already gone" and is
+        /// the correct no-op outcome.
+        /// </para>
        /// </summary>
        public async Task UpdateSessionImeAgentVersionAsync(string tenantId, string sessionId, string version)
        {
@@ -2010,7 +2018,13 @@ namespace AutopilotMonitor.Functions.Services
                {
                    ["ImeAgentVersion"] = version
                };
-                await tableClient.UpsertEntityAsync(entity, TableUpdateMode.Merge);
+                await tableClient.UpdateEntityAsync(entity, ETag.All, TableUpdateMode.Merge);
+            }
+            catch (RequestFailedException ex) when (ex.Status == 404)
+            {
+                _logger.LogDebug(
+                    "UpdateSessionImeAgentVersion no-op: Sessions row {Tenant}/{Session} is absent (tombstoned or never registered)",
+                    tenantId, sessionId);
            }
            catch (Exception ex)
            {