feat(backend): SoftwareInventory contributions side-row + cascade delete helpers (PR2/7)

PR2 of the safe cascade-delete plan
(.claude/plans/session-deletion-cascade-safe.md). Adds the per-table
delete primitives the cascade worker will use in PR4 plus the §17
SessionInventoryContributions side-row that makes per-session inventory
counter increments at-most-once and exactly reversible. No production
caller invokes the cascade primitives yet; the inventory rewrite IS a
call-site change in VulnerabilityCorrelationService and ships in this PR.

== Cascade delete helpers (TableStorageService.Deletion.cs) ==
DeleteByExactKeysInBatchesAsync(tableName, keys) is the load-bearing
helper: groups by PartitionKey (Azure batch transactions are
partition-scoped), chunks at 100 actions per batch, falls back to
per-row DeleteEntityAsync with 404-ignore on whole-batch rollback.
Returns (Attempted, DeletedNow, AlreadyMissing) for accurate progress
accounting on idempotent re-runs of partial cascades. Plus 9 thin
wrappers per the §3 taxonomy: 7 keyed (EventTypeIndex, CveIndex,
SessionsByTerminal, SessionsByStage, DeadEndsByReason,
ClassifierVerdictsByIdLevel, SignalsByKind), 1 single-row
(DeviceSnapshot), and 2 enumerate-then-delete (Signals,
DecisionTransitions for the PK_BY_SESSION class). All fail-loud per
memory feedback_storage_helpers_fail_soft.

== SoftwareInventory side-row (TableStorageService.Inventory.cs) ==
Plan §17 design: each inventory-correlated session contributes ONE
SessionInventoryContributions row (PK=tenantId, RK=sessionId) with the
normalized software keys it added. Increments to SoftwareInventory
counters are now driven by deltas against this row, so:
  - WG sessions that fire software_inventory_analysis 4× no longer
    over-count by 4× per software per session
  - Cascade decrement is exact: the side-row is the source of truth
    for which counters this session contributed to
  - Pre-side-row sessions (created before this deploy) have no
    contributions row → cascade skips inventory entirely for them
    (drift accepted on the historical backlog, ages out via 90d
    retention).

Helpers (all fail-loud, all bounded retry per §12-Q10 = 10 attempts /
60s wall-clock cap, all ETag-CAS):
  - GetSessionInventoryContributionsAsync: returns prior key list or
    null on 404
  - UpsertSessionInventoryContributionsAsync: side-row CAS that
    computes (added, removed) delta against the new key set
  - DeleteSessionInventoryContributionsAsync: 404-ignore tombstone
  - IncrementSoftwareInventoryEntryAsync: ETag-CAS +1 with first-time
    AddEntity that writes full metadata
  - DecrementSoftwareInventoryEntryAsync: ETag-CAS -1, clamp >= 0,
    deletes the row when count hits zero

Codec: SoftwareKeysJsonCodec encodes the key list as raw JSON below
30 KB, gzip+Base64 above. Same shape DeletionManifestBuilder (PR1)
already reads at preflight, so the cascade decrement reads back what
the increment wrote with no schema gap. Plus deterministic ComputeDelta
(case-insensitive composite key) so concurrent correlations on the
same session collapse to a single counter update.

== VulnerabilityCorrelationService rewrite ==
The Task.Run fire-and-forget at the old line 246 now calls the new
ApplyInventoryContributionsAsync helper instead of
UpsertSoftwareInventoryAsync. Each Increment / Decrement is wrapped
in its OWN try/catch so one counter failure does NOT abort the rest
of the loop (Plan §17.3 + §17.8): without that, the side-row would
already promise the new key set while subsequent runs see an empty
delta and never retry the under-counted keys. Each individual failure
emits an `inventory_counter_write_failed` audit entry with operation
+ vendor + name + version + exception type/message via the existing
LogAuditEntryAsync. IMaintenanceRepository was added as an OPTIONAL
constructor param (positional default null) so existing 8-arg test
factories keep compiling unchanged.

ParseSoftwareInventoryItems exposed as internal static for testability;
preserves the same dedup + cpe-mapping behaviour as the old
UpsertSoftwareInventoryAsync path. The legacy method is left in place
on TableStorageService.Rules.cs for now (no remaining callers; PR7
cleanup can remove if desired).

== Tests (39 new) ==
- SoftwareKeysJsonCodecTests (12): raw vs gzip+Base64 paths,
  determinism, malformed-input rejection, delta cases (added /
  removed / identity / case-insensitive), composite-key shape.
- TableStorageDeletionBatchTests (8, SDK-mocked via internal test
  ctor): empty input, PK grouping (one batch per PK), chunking at
  100, batch 404 + all per-row 404, mixed batch 404 (2 succeed +
  3 missing) with accurate accounting, non-404 propagates, no
  over-deletion, ETag.All on every action.
- TableStorageInventoryCasTests (9): Increment first-create with
  full metadata, CAS +1 on existing row, 412 retry, 409 first-create
  race fall back to Update, empty-triple skip; Decrement 404 no-op,
  clamp-at-zero deletes the row, count>1 subtract, 0-count clamp,
  412 retry.
- VulnerabilityCorrelationServiceContributionsTests (10): Parser
  correctness, end-to-end add/remove flow with Mock.Verify counts,
  empty-input no-op, side-row exhaustion swallowed, per-counter
  failure does not abort + audits the failed key (both increment
  and decrement variants).

Backend suite: 1290/1290 green.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

src/Backend/AutopilotMonitor.Functions.Tests/SoftwareKeysJsonCodecTests.cs

@@ -0,0 +1,223 @@
+using System.Collections.Generic;
+using System.Linq;
+using AutopilotMonitor.Shared.Models.Deletion;
+using Xunit;
+
+namespace AutopilotMonitor.Functions.Tests;
+
+/// <summary>
+/// Pure-logic tests for the SoftwareKeysJson codec used by the inventory contributions side-row
+/// (Plan §17). Encoding determinism + delta computation are the load-bearing properties — the
+/// cascade decrement step reads back the side-row's keys and is byte-identical to what the
+/// increment path wrote.
+/// </summary>
+public class SoftwareKeysJsonCodecTests
+{
+    [Fact]
+    public void Encode_uses_raw_json_below_compression_threshold()
+    {
+        var keys = new List<DeletionDecrementKey>
+        {
+            new DeletionDecrementKey { Vendor = "Contoso", Name = "Widget", Version = "1.0" },
+            new DeletionDecrementKey { Vendor = "Fabrikam", Name = "Gadget", Version = "2.0" },
+        };
+
+        var encoded = SoftwareKeysJsonCodec.Encode(keys);
+
+        Assert.False(encoded.IsCompressed);
+        Assert.Equal(2, encoded.KeyCount);
+        Assert.Contains("Contoso", encoded.Encoded);
+        Assert.Contains("Fabrikam", encoded.Encoded);
+    }
+
+    [Fact]
+    public void Encode_uses_gzip_base64_above_compression_threshold()
+    {
+        // Generate >30 KB of raw JSON: 1000 distinct keys with ~30-char vendor each.
+        var keys = new List<DeletionDecrementKey>();
+        for (var i = 0; i < 1000; i++)
+        {
+            keys.Add(new DeletionDecrementKey
+            {
+                Vendor = $"VendorWithALongerNameForBulk-{i:D6}",
+                Name = $"Product-{i:D6}",
+                Version = $"1.{i}.0",
+            });
+        }
+
+        var encoded = SoftwareKeysJsonCodec.Encode(keys);
+
+        Assert.True(encoded.IsCompressed);
+        Assert.Equal(1000, encoded.KeyCount);
+        // Compressed payload should be much smaller than raw (heavy redundancy).
+        var rawSize = System.Text.Json.JsonSerializer.SerializeToUtf8Bytes(keys, DeletionManifestJson.SerializerOptions).Length;
+        Assert.True(encoded.Encoded.Length * 3 / 4 < rawSize / 3,
+            $"Compression should shrink payload by >3x. Got base64 {encoded.Encoded.Length} chars vs raw {rawSize} bytes.");
+    }
+
+    [Fact]
+    public void Encode_decode_round_trips_keys_in_deterministic_order()
+    {
+        var keys = new List<DeletionDecrementKey>
+        {
+            new DeletionDecrementKey { Vendor = "Tailspin", Name = "Service", Version = "3.0" },
+            new DeletionDecrementKey { Vendor = "Contoso", Name = "Widget", Version = "1.0" },
+            new DeletionDecrementKey { Vendor = "Fabrikam", Name = "Gadget", Version = "2.0" },
+        };
+
+        var encoded = SoftwareKeysJsonCodec.Encode(keys);
+        var decoded = SoftwareKeysJsonCodec.Decode(encoded.Encoded, encoded.IsCompressed);
+
+        Assert.Equal(3, decoded.Count);
+        // Builder sorts by (Vendor, Name, Version) so the side-row content is byte-stable
+        // across builds — critical for manifest-determinism (Plan §18.6).
+        Assert.Equal("Contoso", decoded[0].Vendor);
+        Assert.Equal("Fabrikam", decoded[1].Vendor);
+        Assert.Equal("Tailspin", decoded[2].Vendor);
+    }
+
+    [Fact]
+    public void Encode_with_compression_round_trips_through_gzip_base64()
+    {
+        // Force the compressed path with a key set that just barely exceeds the threshold.
+        var keys = new List<DeletionDecrementKey>();
+        for (var i = 0; i < 500; i++)
+        {
+            keys.Add(new DeletionDecrementKey
+            {
+                Vendor = "VendorWithALongerNameForBulk-" + i.ToString("D6"),
+                Name = "Product-" + i.ToString("D6"),
+                Version = "1." + i + ".0",
+            });
+        }
+
+        var encoded = SoftwareKeysJsonCodec.Encode(keys);
+        var decoded = SoftwareKeysJsonCodec.Decode(encoded.Encoded, encoded.IsCompressed);
+
+        Assert.True(encoded.IsCompressed);
+        Assert.Equal(500, decoded.Count);
+        Assert.Equal(keys[0].Vendor, decoded.First(k => k.Vendor == keys[0].Vendor).Vendor);
+    }
+
+    [Fact]
+    public void Decode_empty_string_returns_empty_list()
+    {
+        Assert.Empty(SoftwareKeysJsonCodec.Decode("", false));
+        Assert.Empty(SoftwareKeysJsonCodec.Decode("", true));
+    }
+
+    [Fact]
+    public void Decode_invalid_base64_throws_invalid_data()
+    {
+        var ex = Assert.Throws<System.IO.InvalidDataException>(
+            () => SoftwareKeysJsonCodec.Decode("!!! not base64 !!!", isCompressed: true));
+        Assert.Contains("Base64", ex.Message);
+    }
+
+    [Fact]
+    public void Decode_malformed_json_throws_invalid_data()
+    {
+        Assert.Throws<System.IO.InvalidDataException>(
+            () => SoftwareKeysJsonCodec.Decode("{not-an-array}", isCompressed: false));
+    }
+
+    [Fact]
+    public void ComputeDelta_returns_all_added_when_oldKeys_is_empty()
+    {
+        var newKeys = new[]
+        {
+            new DeletionDecrementKey { Vendor = "Contoso", Name = "Widget", Version = "1.0" },
+            new DeletionDecrementKey { Vendor = "Fabrikam", Name = "Gadget", Version = "2.0" },
+        };
+
+        var (added, removed) = SoftwareKeysJsonCodec.ComputeDelta(System.Array.Empty<DeletionDecrementKey>(), newKeys);
+
+        Assert.Equal(2, added.Count);
+        Assert.Empty(removed);
+    }
+
+    [Fact]
+    public void ComputeDelta_returns_one_added_when_newKeys_adds_a_phase2_app()
+    {
+        var oldKeys = new[]
+        {
+            new DeletionDecrementKey { Vendor = "Contoso", Name = "Widget", Version = "1.0" },
+        };
+        var newKeys = new[]
+        {
+            new DeletionDecrementKey { Vendor = "Contoso", Name = "Widget", Version = "1.0" },
+            new DeletionDecrementKey { Vendor = "Fabrikam", Name = "NewApp", Version = "2.0" },
+        };
+
+        var (added, removed) = SoftwareKeysJsonCodec.ComputeDelta(oldKeys, newKeys);
+
+        Assert.Single(added);
+        Assert.Contains(SoftwareKeysJsonCodec.CompositeKey("Fabrikam", "NewApp", "2.0"), added);
+        Assert.Empty(removed);
+    }
+
+    [Fact]
+    public void ComputeDelta_returns_one_removed_when_uninstall_happens_between_phases()
+    {
+        var oldKeys = new[]
+        {
+            new DeletionDecrementKey { Vendor = "Contoso", Name = "Widget", Version = "1.0" },
+            new DeletionDecrementKey { Vendor = "Fabrikam", Name = "OldApp", Version = "0.9" },
+        };
+        var newKeys = new[]
+        {
+            new DeletionDecrementKey { Vendor = "Contoso", Name = "Widget", Version = "1.0" },
+        };
+
+        var (added, removed) = SoftwareKeysJsonCodec.ComputeDelta(oldKeys, newKeys);
+
+        Assert.Empty(added);
+        Assert.Single(removed);
+        Assert.Contains(SoftwareKeysJsonCodec.CompositeKey("Fabrikam", "OldApp", "0.9"), removed);
+    }
+
+    [Fact]
+    public void ComputeDelta_is_empty_when_keys_are_identical()
+    {
+        var keys = new[]
+        {
+            new DeletionDecrementKey { Vendor = "Contoso", Name = "Widget", Version = "1.0" },
+            new DeletionDecrementKey { Vendor = "Fabrikam", Name = "Gadget", Version = "2.0" },
+        };
+
+        var (added, removed) = SoftwareKeysJsonCodec.ComputeDelta(keys, keys);
+
+        Assert.Empty(added);
+        Assert.Empty(removed);
+    }
+
+    [Fact]
+    public void ComputeDelta_is_case_insensitive()
+    {
+        // Upstream normalization is best-effort; "Contoso" and "contoso" must collapse so a case-
+        // flip between two correlation events on the same session doesn't trigger spurious deltas.
+        var oldKeys = new[]
+        {
+            new DeletionDecrementKey { Vendor = "Contoso", Name = "Widget", Version = "1.0" },
+        };
+        var newKeys = new[]
+        {
+            new DeletionDecrementKey { Vendor = "contoso", Name = "widget", Version = "1.0" },
+        };
+
+        var (added, removed) = SoftwareKeysJsonCodec.ComputeDelta(oldKeys, newKeys);
+
+        Assert.Empty(added);
+        Assert.Empty(removed);
+    }
+
+    [Fact]
+    public void CompositeKey_builds_three_part_separated_value()
+    {
+        var ck = SoftwareKeysJsonCodec.CompositeKey("Contoso", "Widget", "1.0");
+        Assert.NotEqual("ContosoWidget1.0", ck); // separator IS present
+        // Different orderings of the same string components produce different composites.
+        var ck2 = SoftwareKeysJsonCodec.CompositeKey("Widget", "Contoso", "1.0");
+        Assert.NotEqual(ck, ck2);
+    }
+}

src/Backend/AutopilotMonitor.Functions.Tests/TableStorageDeletionBatchTests.cs

@@ -0,0 +1,241 @@
+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.Services;
+using AutopilotMonitor.Shared;
+using Microsoft.Extensions.Logging.Abstractions;
+using Moq;
+using Xunit;
+
+namespace AutopilotMonitor.Functions.Tests;
+
+/// <summary>
+/// SDK-mocked behaviour tests for <see cref="TableStorageService.DeleteByExactKeysInBatchesAsync"/>.
+/// This is the safety layer the PR4 cascade worker depends on for idempotent re-runs: groups by
+/// PartitionKey, chunks at 100 actions per batch (Azure Tables' transaction limit), falls back
+/// to per-row delete with 404-ignore when whole-batch rollback hides which row was already
+/// missing. Tests use Moq against the Azure SDK's virtual surface — no Azurite, matching repo
+/// convention.
+/// </summary>
+public class TableStorageDeletionBatchTests
+{
+    private const string TableName = Constants.TableNames.EventTypeIndex;
+
+    [Fact]
+    public async Task Delete_returns_empty_result_for_empty_key_list()
+    {
+        var harness = new Harness();
+        var result = await harness.Sut.DeleteByExactKeysInBatchesAsync(TableName, new List<(string, string)>());
+
+        Assert.Equal(0, result.Attempted);
+        Assert.Equal(0, result.DeletedNow);
+        Assert.Equal(0, result.AlreadyMissing);
+        Assert.Empty(harness.SubmittedBatches);
+    }
+
+    [Fact]
+    public async Task Delete_groups_keys_by_partition_so_one_batch_per_pk()
+    {
+        var harness = new Harness();
+        var keys = new List<(string Pk, string Rk)>
+        {
+            ("pkA", "rk1"), ("pkA", "rk2"),
+            ("pkB", "rk3"), ("pkB", "rk4"), ("pkB", "rk5"),
+            ("pkA", "rk6"),
+        };
+
+        var result = await harness.Sut.DeleteByExactKeysInBatchesAsync(TableName, keys);
+
+        Assert.Equal(6, result.Attempted);
+        Assert.Equal(6, result.DeletedNow);
+        Assert.Equal(0, result.AlreadyMissing);
+
+        // Two partitions → two batch transactions. Azure Tables forbids mixing PKs in one batch.
+        Assert.Equal(2, harness.SubmittedBatches.Count);
+        var batchPks = harness.SubmittedBatches
+            .Select(b => b.Select(a => a.Entity.PartitionKey).Distinct().Single())
+            .OrderBy(p => p)
+            .ToList();
+        Assert.Equal(new[] { "pkA", "pkB" }, batchPks);
+    }
+
+    [Fact]
+    public async Task Delete_chunks_each_partition_into_batches_of_100()
+    {
+        // 250 rows in one PK → 3 batch transactions (100, 100, 50).
+        var harness = new Harness();
+        var keys = Enumerable.Range(0, 250).Select(i => ("pkA", $"rk_{i:D4}")).ToList();
+
+        var result = await harness.Sut.DeleteByExactKeysInBatchesAsync(TableName, keys);
+
+        Assert.Equal(250, result.Attempted);
+        Assert.Equal(250, result.DeletedNow);
+        Assert.Equal(3, harness.SubmittedBatches.Count);
+        Assert.Equal(100, harness.SubmittedBatches[0].Count);
+        Assert.Equal(100, harness.SubmittedBatches[1].Count);
+        Assert.Equal(50,  harness.SubmittedBatches[2].Count);
+    }
+
+    [Fact]
+    public async Task Delete_falls_back_to_per_row_on_batch_404_and_counts_all_already_missing()
+    {
+        // Idempotent re-run scenario: batch transaction fails 404 (Azure rolls back the whole
+        // batch on any 404 inside it). Per-row fallback finds every row already missing → result
+        // is everything in AlreadyMissing, nothing in DeletedNow, no throws.
+        var harness = new Harness
+        {
+            BatchBehavior = _ => throw new RequestFailedException(404, "ResourceNotFound"),
+            PerRowBehavior = _ => throw new RequestFailedException(404, "ResourceNotFound"),
+        };
+
+        var keys = Enumerable.Range(0, 5).Select(i => ("pkA", $"rk_{i}")).ToList();
+        var result = await harness.Sut.DeleteByExactKeysInBatchesAsync(TableName, keys);
+
+        Assert.Equal(5, result.Attempted);
+        Assert.Equal(0, result.DeletedNow);
+        Assert.Equal(5, result.AlreadyMissing);
+        Assert.Single(harness.SubmittedBatches);
+        Assert.Equal(5, harness.PerRowDeletes.Count);
+    }
+
+    [Fact]
+    public async Task Delete_mixed_batch_404_fallback_counts_partial_deletes_and_misses()
+    {
+        // 5 rows; batch fails 404; per-row fallback finds 2 exist and 3 are missing.
+        var perRowState = new Dictionary<string, bool>
+        {
+            ["rk_0"] = true, ["rk_1"] = false, ["rk_2"] = true, ["rk_3"] = false, ["rk_4"] = false,
+        };
+        var harness = new Harness
+        {
+            BatchBehavior = _ => throw new RequestFailedException(404, "ResourceNotFound"),
+            PerRowBehavior = rk =>
+            {
+                if (perRowState.TryGetValue(rk, out var exists) && !exists)
+                    throw new RequestFailedException(404, "ResourceNotFound");
+                return new Mock<Response>().Object;
+            },
+        };
+
+        var keys = perRowState.Keys.Select(rk => ("pkA", rk)).ToList();
+        var result = await harness.Sut.DeleteByExactKeysInBatchesAsync(TableName, keys);
+
+        Assert.Equal(5, result.Attempted);
+        Assert.Equal(2, result.DeletedNow);
+        Assert.Equal(3, result.AlreadyMissing);
+    }
+
+    [Fact]
+    public async Task Delete_propagates_non_404_storage_errors()
+    {
+        // 500 / 503 / quota — real failures the cascade worker must surface so the queue retry can fire.
+        var harness = new Harness
+        {
+            BatchBehavior = _ => throw new RequestFailedException(500, "Internal Server Error"),
+        };
+
+        await Assert.ThrowsAsync<RequestFailedException>(
+            () => harness.Sut.DeleteByExactKeysInBatchesAsync(
+                TableName, new List<(string, string)> { ("pkA", "rk_0") }));
+    }
+
+    [Fact]
+    public async Task Delete_only_targets_keys_in_the_input_list_no_over_deletion()
+    {
+        var harness = new Harness();
+        var keys = new List<(string, string)>
+        {
+            ("pkA", "rk_target_1"), ("pkA", "rk_target_2"), ("pkA", "rk_target_3"),
+        };
+
+        await harness.Sut.DeleteByExactKeysInBatchesAsync(TableName, keys);
+
+        var deletedKeys = harness.SubmittedBatches
+            .SelectMany(b => b)
+            .Select(a => a.Entity.RowKey)
+            .OrderBy(k => k)
+            .ToList();
+        Assert.Equal(new[] { "rk_target_1", "rk_target_2", "rk_target_3" }, deletedKeys);
+
+        // Every action is a Delete (no accidental Upsert / Update).
+        foreach (var batch in harness.SubmittedBatches)
+        {
+            foreach (var action in batch)
+            {
+                Assert.Equal(TableTransactionActionType.Delete, action.ActionType);
+            }
+        }
+    }
+
+    [Fact]
+    public async Task Delete_uses_eTag_All_so_concurrent_writes_dont_block_cascade()
+    {
+        var harness = new Harness();
+        await harness.Sut.DeleteByExactKeysInBatchesAsync(
+            TableName, new List<(string, string)> { ("pkA", "rk_0") });
+
+        var batch = harness.SubmittedBatches.Single();
+        Assert.Equal(ETag.All, batch.Single().Entity.ETag);
+    }
+
+    // ---------------------------------------------------------------- Harness ----
+
+    /// <summary>
+    /// SDK-mock harness: <see cref="BatchBehavior"/> and <see cref="PerRowBehavior"/> are
+    /// per-test overrides for the SubmitTransactionAsync / DeleteEntityAsync calls. Captured
+    /// invocations are accumulated in <see cref="SubmittedBatches"/> and <see cref="PerRowDeletes"/>.
+    /// </summary>
+    private sealed class Harness
+    {
+        public List<List<TableTransactionAction>> SubmittedBatches { get; } = new List<List<TableTransactionAction>>();
+        public List<(string Pk, string Rk)> PerRowDeletes { get; } = new List<(string, string)>();
+
+        /// <summary>If set, called instead of the default success response on SubmitTransactionAsync.</summary>
+        public Func<List<TableTransactionAction>, Response<IReadOnlyList<Response>>>? BatchBehavior { get; set; }
+
+        /// <summary>If set, called instead of the default success response on per-row DeleteEntityAsync.</summary>
+        public Func<string, Response>? PerRowBehavior { get; set; }
+
+        public TableStorageService Sut { get; }
+
+        public Harness()
+        {
+            var mockTableClient = new Mock<TableClient>();
+
+            mockTableClient
+                .Setup(c => c.SubmitTransactionAsync(It.IsAny<IEnumerable<TableTransactionAction>>(), It.IsAny<CancellationToken>()))
+                .Returns<IEnumerable<TableTransactionAction>, CancellationToken>((actions, _) =>
+                {
+                    var snapshot = actions.ToList();
+                    SubmittedBatches.Add(snapshot);
+                    if (BatchBehavior != null)
+                    {
+                        return Task.FromResult(BatchBehavior(snapshot));
+                    }
+                    var inner = (IReadOnlyList<Response>)snapshot.Select(_ => new Mock<Response>().Object).ToList();
+                    return Task.FromResult(Response.FromValue(inner, new Mock<Response>().Object));
+                });
+
+            mockTableClient
+                .Setup(c => c.DeleteEntityAsync(It.IsAny<string>(), It.IsAny<string>(), It.IsAny<ETag>(), It.IsAny<CancellationToken>()))
+                .Returns<string, string, ETag, CancellationToken>((pk, rk, _, _) =>
+                {
+                    PerRowDeletes.Add((pk, rk));
+                    if (PerRowBehavior != null)
+                    {
+                        return Task.FromResult(PerRowBehavior(rk));
+                    }
+                    return Task.FromResult(new Mock<Response>().Object);
+                });
+
+            var mockServiceClient = new Mock<TableServiceClient>();
+            mockServiceClient.Setup(s => s.GetTableClient(It.IsAny<string>())).Returns(mockTableClient.Object);
+
+            Sut = new TableStorageService(mockServiceClient.Object, NullLogger<TableStorageService>.Instance);
+        }
+    }
+}

src/Backend/AutopilotMonitor.Functions.Tests/TableStorageInventoryCasTests.cs

@@ -0,0 +1,374 @@
+using System;
+using System.Collections.Generic;
+using System.Threading;
+using System.Threading.Tasks;
+using Azure;
+using Azure.Data.Tables;
+using AutopilotMonitor.Functions.Services;
+using AutopilotMonitor.Shared;
+using AutopilotMonitor.Shared.Models.Deletion;
+using Microsoft.Extensions.Logging.Abstractions;
+using Moq;
+using Xunit;
+
+namespace AutopilotMonitor.Functions.Tests;
+
+/// <summary>
+/// SDK-mocked behaviour tests for the SoftwareInventory side-row + counter ETag-CAS helpers
+/// in <see cref="TableStorageService"/>. These are the writer-side correctness primitives that
+/// the cascade decrement step (PR4) and the new VulnerabilityCorrelationService increment path
+/// (PR2) depend on. Tests exercise the actual retry loops + clamp + 404 idempotency without
+/// hitting Azure, matching repo convention.
+/// </summary>
+public class TableStorageInventoryCasTests
+{
+    private const string TenantId  = "11111111-1111-1111-1111-111111111111";
+    private const string SessionId = "22222222-2222-2222-2222-222222222222";
+
+    // ============================================================ IncrementSoftwareInventoryEntryAsync ====
+
+    [Fact]
+    public async Task Increment_creates_row_with_full_metadata_when_no_row_exists()
+    {
+        // First encounter for a (tenant, key) pair → AddEntity with all the display + normalization
+        // metadata, SessionCount=1, FirstSeenAt/LastSeenAt populated.
+        var harness = new Harness();
+        harness.SoftwareInventory.Setup(t =>
+            t.GetEntityAsync<TableEntity>(TenantId, It.IsAny<string>(), It.IsAny<IEnumerable<string>>(), It.IsAny<CancellationToken>()))
+            .ThrowsAsync(new RequestFailedException(404, "ResourceNotFound"));
+
+        TableEntity? addedEntity = null;
+        harness.SoftwareInventory.Setup(t => t.AddEntityAsync(It.IsAny<TableEntity>(), It.IsAny<CancellationToken>()))
+            .Returns<TableEntity, CancellationToken>((e, _) =>
+            {
+                addedEntity = e;
+                return Task.FromResult(new Mock<Response>().Object);
+            });
+
+        var item = new SoftwareInventoryItem
+        {
+            NormalizedVendor = "contoso", NormalizedName = "widget", NormalizedVersion = "1.0",
+            DisplayName = "Contoso Widget", Publisher = "Contoso Ltd.",
+            RegistrySource = "x64", NormalizationConfidence = "high",
+            CpeUri = "cpe:2.3:a:contoso:widget:*:*:*:*:*:*:*:*",
+        };
+
+        await harness.Sut.IncrementSoftwareInventoryEntryAsync(TenantId, item, SessionId);
+
+        Assert.NotNull(addedEntity);
+        Assert.Equal(TenantId, addedEntity!.PartitionKey);
+        Assert.Equal(1, addedEntity.GetInt32("SessionCount"));
+        Assert.Equal("Contoso Widget", addedEntity.GetString("DisplayName"));
+        Assert.Equal("widget", addedEntity.GetString("NormalizedName"));
+        Assert.Equal("contoso", addedEntity.GetString("NormalizedVendor"));
+        Assert.Equal("1.0", addedEntity.GetString("NormalizedVersion"));
+        Assert.Equal("cpe:2.3:a:contoso:widget:*:*:*:*:*:*:*:*", addedEntity.GetString("CpeUri"));
+        Assert.Equal(SessionId, addedEntity.GetString("FirstSessionId"));
+        Assert.Equal(SessionId, addedEntity.GetString("LastSessionId"));
+
+        harness.SoftwareInventory.Verify(t => t.AddEntityAsync(It.IsAny<TableEntity>(), It.IsAny<CancellationToken>()), Times.Once);
+        harness.SoftwareInventory.Verify(t => t.UpdateEntityAsync(It.IsAny<TableEntity>(), It.IsAny<ETag>(), It.IsAny<TableUpdateMode>(), It.IsAny<CancellationToken>()), Times.Never);
+    }
+
+    [Fact]
+    public async Task Increment_does_etag_cas_plus_one_when_row_exists()
+    {
+        var existing = new TableEntity(TenantId, "contoso:widget:1.0")
+        {
+            ["DisplayName"]   = "Contoso Widget",
+            ["NormalizedName"] = "widget",
+            ["SessionCount"]  = 3,
+            ["LastSeenAt"]    = DateTime.UtcNow.AddDays(-1).ToString("o"),
+            ["CpeUri"]        = "cpe:2.3:a:contoso:widget:*:*:*:*:*:*:*:*",
+        };
+        existing.ETag = new ETag("0xABC");
+
+        var harness = new Harness();
+        harness.SoftwareInventory.Setup(t =>
+            t.GetEntityAsync<TableEntity>(TenantId, It.IsAny<string>(), It.IsAny<IEnumerable<string>>(), It.IsAny<CancellationToken>()))
+            .ReturnsAsync(Response.FromValue(existing, new Mock<Response>().Object));
+
+        TableEntity? written = null;
+        harness.SoftwareInventory.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;
+                Assert.Equal(new ETag("0xABC"), etag);
+                return Task.FromResult(new Mock<Response>().Object);
+            });
+
+        var item = new SoftwareInventoryItem
+        {
+            NormalizedVendor = "contoso", NormalizedName = "widget", NormalizedVersion = "1.0",
+        };
+        await harness.Sut.IncrementSoftwareInventoryEntryAsync(TenantId, item, SessionId);
+
+        Assert.NotNull(written);
+        Assert.Equal(4, written!.GetInt32("SessionCount"));
+        Assert.Equal(SessionId, written.GetString("LastSessionId"));
+        harness.SoftwareInventory.Verify(t => t.AddEntityAsync(It.IsAny<TableEntity>(), It.IsAny<CancellationToken>()), Times.Never);
+    }
+
+    [Fact]
+    public async Task Increment_retries_on_412_etag_conflict_and_succeeds_on_second_attempt()
+    {
+        var existing = new TableEntity(TenantId, "contoso:widget:1.0")
+        {
+            ["SessionCount"] = 2,
+        };
+        existing.ETag = new ETag("0xSTALE");
+
+        var harness = new Harness();
+        harness.SoftwareInventory.Setup(t =>
+            t.GetEntityAsync<TableEntity>(TenantId, It.IsAny<string>(), It.IsAny<IEnumerable<string>>(), It.IsAny<CancellationToken>()))
+            .ReturnsAsync(Response.FromValue(existing, new Mock<Response>().Object));
+
+        var updateCalls = 0;
+        harness.SoftwareInventory.Setup(t => t.UpdateEntityAsync(
+                It.IsAny<TableEntity>(), It.IsAny<ETag>(), It.IsAny<TableUpdateMode>(), It.IsAny<CancellationToken>()))
+            .Returns<TableEntity, ETag, TableUpdateMode, CancellationToken>((e, _, _, _) =>
+            {
+                updateCalls++;
+                if (updateCalls == 1)
+                {
+                    throw new RequestFailedException(412, "Precondition Failed");
+                }
+                return Task.FromResult(new Mock<Response>().Object);
+            });
+
+        var item = new SoftwareInventoryItem
+        {
+            NormalizedVendor = "contoso", NormalizedName = "widget", NormalizedVersion = "1.0",
+        };
+
+        await harness.Sut.IncrementSoftwareInventoryEntryAsync(TenantId, item, SessionId);
+
+        Assert.Equal(2, updateCalls);
+    }
+
+    [Fact]
+    public async Task Increment_retries_on_first_create_race_409_and_falls_back_to_update_path()
+    {
+        // Race: two callers see no row, both try AddEntity. First wins (no 409). Second gets 409,
+        // re-reads on next attempt, finds the row, and goes through the Update path.
+        var harness = new Harness();
+        var getCalls = 0;
+        var existingAfterRace = new TableEntity(TenantId, "contoso:widget:1.0")
+        {
+            ["SessionCount"] = 1,
+        };
+        existingAfterRace.ETag = new ETag("0xPOSTRACE");
+
+        harness.SoftwareInventory.Setup(t =>
+            t.GetEntityAsync<TableEntity>(TenantId, It.IsAny<string>(), It.IsAny<IEnumerable<string>>(), It.IsAny<CancellationToken>()))
+            .Returns(() =>
+            {
+                getCalls++;
+                if (getCalls == 1)
+                {
+                    return Task.FromException<Response<TableEntity>>(new RequestFailedException(404, "ResourceNotFound"));
+                }
+                return Task.FromResult(Response.FromValue(existingAfterRace, new Mock<Response>().Object));
+            });
+
+        harness.SoftwareInventory.Setup(t => t.AddEntityAsync(It.IsAny<TableEntity>(), It.IsAny<CancellationToken>()))
+            .ThrowsAsync(new RequestFailedException(409, "EntityAlreadyExists"));
+
+        var updateCalls = 0;
+        harness.SoftwareInventory.Setup(t => t.UpdateEntityAsync(
+                It.IsAny<TableEntity>(), It.IsAny<ETag>(), It.IsAny<TableUpdateMode>(), It.IsAny<CancellationToken>()))
+            .Returns<TableEntity, ETag, TableUpdateMode, CancellationToken>((e, _, _, _) =>
+            {
+                updateCalls++;
+                return Task.FromResult(new Mock<Response>().Object);
+            });
+
+        var item = new SoftwareInventoryItem
+        {
+            NormalizedVendor = "contoso", NormalizedName = "widget", NormalizedVersion = "1.0",
+        };
+        await harness.Sut.IncrementSoftwareInventoryEntryAsync(TenantId, item, SessionId);
+
+        Assert.Equal(2, getCalls);
+        Assert.Equal(1, updateCalls);
+    }
+
+    [Fact]
+    public async Task Increment_skips_empty_normalized_triple_silently()
+    {
+        var harness = new Harness();
+        var item = new SoftwareInventoryItem(); // all empty
+        await harness.Sut.IncrementSoftwareInventoryEntryAsync(TenantId, item, SessionId);
+
+        harness.SoftwareInventory.Verify(t =>
+            t.GetEntityAsync<TableEntity>(It.IsAny<string>(), It.IsAny<string>(), It.IsAny<IEnumerable<string>>(), It.IsAny<CancellationToken>()),
+            Times.Never);
+    }
+
+    // ============================================================ DecrementSoftwareInventoryEntryAsync ====
+
+    [Fact]
+    public async Task Decrement_is_no_op_when_row_missing_404()
+    {
+        // Idempotent: re-running the cascade on a session whose counters are already drained
+        // must not throw. The helper returns silently.
+        var harness = new Harness();
+        harness.SoftwareInventory.Setup(t =>
+            t.GetEntityAsync<TableEntity>(TenantId, It.IsAny<string>(), It.IsAny<IEnumerable<string>>(), It.IsAny<CancellationToken>()))
+            .ThrowsAsync(new RequestFailedException(404, "ResourceNotFound"));
+
+        await harness.Sut.DecrementSoftwareInventoryEntryAsync(TenantId, "contoso", "widget", "1.0");
+
+        harness.SoftwareInventory.Verify(t => t.UpdateEntityAsync(
+            It.IsAny<TableEntity>(), It.IsAny<ETag>(), It.IsAny<TableUpdateMode>(), It.IsAny<CancellationToken>()),
+            Times.Never);
+        harness.SoftwareInventory.Verify(t =>
+            t.DeleteEntityAsync(It.IsAny<string>(), It.IsAny<string>(), It.IsAny<ETag>(), It.IsAny<CancellationToken>()),
+            Times.Never);
+    }
+
+    [Fact]
+    public async Task Decrement_clamps_at_zero_and_deletes_the_row_when_count_drops_to_zero()
+    {
+        var existing = new TableEntity(TenantId, "contoso:widget:1.0")
+        {
+            ["SessionCount"] = 1,
+        };
+        existing.ETag = new ETag("0xLASTONE");
+
+        var harness = new Harness();
+        harness.SoftwareInventory.Setup(t =>
+            t.GetEntityAsync<TableEntity>(TenantId, It.IsAny<string>(), It.IsAny<IEnumerable<string>>(), It.IsAny<CancellationToken>()))
+            .ReturnsAsync(Response.FromValue(existing, new Mock<Response>().Object));
+
+        harness.SoftwareInventory.Setup(t =>
+            t.DeleteEntityAsync(TenantId, It.IsAny<string>(), It.IsAny<ETag>(), It.IsAny<CancellationToken>()))
+            .ReturnsAsync(new Mock<Response>().Object);
+
+        await harness.Sut.DecrementSoftwareInventoryEntryAsync(TenantId, "contoso", "widget", "1.0");
+
+        // Row is deleted (count→0), NOT updated to a row with SessionCount=0.
+        harness.SoftwareInventory.Verify(t =>
+            t.DeleteEntityAsync(TenantId, It.IsAny<string>(), new ETag("0xLASTONE"), It.IsAny<CancellationToken>()),
+            Times.Once);
+        harness.SoftwareInventory.Verify(t => t.UpdateEntityAsync(
+            It.IsAny<TableEntity>(), It.IsAny<ETag>(), It.IsAny<TableUpdateMode>(), It.IsAny<CancellationToken>()),
+            Times.Never);
+    }
+
+    [Fact]
+    public async Task Decrement_subtracts_one_when_count_is_greater_than_one()
+    {
+        var existing = new TableEntity(TenantId, "contoso:widget:1.0")
+        {
+            ["SessionCount"] = 4,
+        };
+        existing.ETag = new ETag("0xSEVERAL");
+
+        var harness = new Harness();
+        harness.SoftwareInventory.Setup(t =>
+            t.GetEntityAsync<TableEntity>(TenantId, It.IsAny<string>(), It.IsAny<IEnumerable<string>>(), It.IsAny<CancellationToken>()))
+            .ReturnsAsync(Response.FromValue(existing, new Mock<Response>().Object));
+
+        TableEntity? written = null;
+        harness.SoftwareInventory.Setup(t => t.UpdateEntityAsync(
+                It.IsAny<TableEntity>(), It.IsAny<ETag>(), It.IsAny<TableUpdateMode>(), It.IsAny<CancellationToken>()))
+            .Returns<TableEntity, ETag, TableUpdateMode, CancellationToken>((e, _, _, _) =>
+            {
+                written = e;
+                return Task.FromResult(new Mock<Response>().Object);
+            });
+
+        await harness.Sut.DecrementSoftwareInventoryEntryAsync(TenantId, "contoso", "widget", "1.0");
+
+        Assert.NotNull(written);
+        Assert.Equal(3, written!.GetInt32("SessionCount"));
+        harness.SoftwareInventory.Verify(t =>
+            t.DeleteEntityAsync(It.IsAny<string>(), It.IsAny<string>(), It.IsAny<ETag>(), It.IsAny<CancellationToken>()),
+            Times.Never);
+    }
+
+    [Fact]
+    public async Task Decrement_clamps_at_zero_even_when_existing_count_is_zero()
+    {
+        // Drift case: side-row promises a decrement but the counter is already at zero. Math.Max
+        // keeps us safe; the row is deleted (count→0 already).
+        var existing = new TableEntity(TenantId, "contoso:widget:1.0")
+        {
+            ["SessionCount"] = 0,
+        };
+        existing.ETag = new ETag("0xZERO");
+
+        var harness = new Harness();
+        harness.SoftwareInventory.Setup(t =>
+            t.GetEntityAsync<TableEntity>(TenantId, It.IsAny<string>(), It.IsAny<IEnumerable<string>>(), It.IsAny<CancellationToken>()))
+            .ReturnsAsync(Response.FromValue(existing, new Mock<Response>().Object));
+
+        harness.SoftwareInventory.Setup(t =>
+            t.DeleteEntityAsync(TenantId, It.IsAny<string>(), It.IsAny<ETag>(), It.IsAny<CancellationToken>()))
+            .ReturnsAsync(new Mock<Response>().Object);
+
+        await harness.Sut.DecrementSoftwareInventoryEntryAsync(TenantId, "contoso", "widget", "1.0");
+
+        // Clamped → row deleted, never updated to a negative count.
+        harness.SoftwareInventory.Verify(t =>
+            t.DeleteEntityAsync(TenantId, It.IsAny<string>(), It.IsAny<ETag>(), It.IsAny<CancellationToken>()),
+            Times.Once);
+    }
+
+    [Fact]
+    public async Task Decrement_retries_on_412_and_succeeds_on_second_attempt()
+    {
+        var existing = new TableEntity(TenantId, "contoso:widget:1.0")
+        {
+            ["SessionCount"] = 5,
+        };
+        existing.ETag = new ETag("0xSTALE");
+
+        var harness = new Harness();
+        harness.SoftwareInventory.Setup(t =>
+            t.GetEntityAsync<TableEntity>(TenantId, It.IsAny<string>(), It.IsAny<IEnumerable<string>>(), It.IsAny<CancellationToken>()))
+            .ReturnsAsync(Response.FromValue(existing, new Mock<Response>().Object));
+
+        var updateCalls = 0;
+        harness.SoftwareInventory.Setup(t => t.UpdateEntityAsync(
+                It.IsAny<TableEntity>(), It.IsAny<ETag>(), It.IsAny<TableUpdateMode>(), It.IsAny<CancellationToken>()))
+            .Returns<TableEntity, ETag, TableUpdateMode, CancellationToken>((e, _, _, _) =>
+            {
+                updateCalls++;
+                if (updateCalls == 1)
+                {
+                    throw new RequestFailedException(412, "Precondition Failed");
+                }
+                return Task.FromResult(new Mock<Response>().Object);
+            });
+
+        await harness.Sut.DecrementSoftwareInventoryEntryAsync(TenantId, "contoso", "widget", "1.0");
+
+        Assert.Equal(2, updateCalls);
+    }
+
+    // ============================================================ Harness ====
+
+    /// <summary>
+    /// SDK-mock harness for SoftwareInventory CAS helpers. Per-test setups configure
+    /// <see cref="SoftwareInventory"/> for the (Get / Add / Update / Delete) call sequence
+    /// the inventory partial drives.
+    /// </summary>
+    private sealed class Harness
+    {
+        public Mock<TableClient> SoftwareInventory { get; }
+        public TableStorageService Sut { get; }
+
+        public Harness()
+        {
+            SoftwareInventory = new Mock<TableClient>();
+            var mockServiceClient = new Mock<TableServiceClient>();
+            mockServiceClient
+                .Setup(s => s.GetTableClient(Constants.TableNames.SoftwareInventory))
+                .Returns(SoftwareInventory.Object);
+            Sut = new TableStorageService(mockServiceClient.Object, NullLogger<TableStorageService>.Instance);
+        }
+    }
+}

src/Backend/AutopilotMonitor.Functions.Tests/VulnerabilityCorrelationServiceContributionsTests.cs

@@ -0,0 +1,328 @@
+using System;
+using System.Collections.Generic;
+using System.Threading;
+using System.Threading.Tasks;
+using AutopilotMonitor.Functions.Services.Vulnerability;
+using AutopilotMonitor.Shared.DataAccess;
+using AutopilotMonitor.Shared.Models.Deletion;
+using Microsoft.Extensions.Logging.Abstractions;
+using Moq;
+using Xunit;
+
+namespace AutopilotMonitor.Functions.Tests;
+
+/// <summary>
+/// Behaviour tests for the rewritten inventory-write path
+/// (<see cref="VulnerabilityCorrelationService.ApplyInventoryContributionsAsync"/>). The
+/// background <c>Task.Run</c> in <c>CorrelateAsync</c> just forwards to this method, so testing
+/// it directly is equivalent to testing the rewritten flow end-to-end without the timing
+/// uncertainty of a fire-and-forget Task.
+/// </summary>
+public class VulnerabilityCorrelationServiceContributionsTests
+{
+    private const string TenantId  = "11111111-1111-1111-1111-111111111111";
+    private const string SessionId = "22222222-2222-2222-2222-222222222222";
+
+    [Fact]
+    public async Task ParseSoftwareInventoryItems_extracts_canonical_normalized_triple_with_cpe()
+    {
+        var raw = new List<Dictionary<string, object>>
+        {
+            new Dictionary<string, object>
+            {
+                ["normalizedName"] = "edge",
+                ["normalizedPublisher"] = "microsoft",
+                ["normalizedVersion"] = "121.0.0.0",
+                ["displayName"] = "Microsoft Edge",
+                ["publisher"] = "Microsoft Corporation",
+                ["registrySource"] = "x64",
+                ["normalizationConfidence"] = "high",
+            },
+        };
+        var cpe = new Dictionary<string, string?>(StringComparer.OrdinalIgnoreCase)
+        {
+            ["edge"] = "cpe:2.3:a:microsoft:edge:*:*:*:*:*:*:*:*",
+        };
+
+        var items = VulnerabilityCorrelationService.ParseSoftwareInventoryItems(raw, cpe);
+
+        Assert.Single(items);
+        Assert.Equal("microsoft", items[0].NormalizedVendor);
+        Assert.Equal("edge", items[0].NormalizedName);
+        Assert.Equal("121.0.0.0", items[0].NormalizedVersion);
+        Assert.Equal("Microsoft Edge", items[0].DisplayName);
+        Assert.Equal("cpe:2.3:a:microsoft:edge:*:*:*:*:*:*:*:*", items[0].CpeUri);
+    }
+
+    [Fact]
+    public void ParseSoftwareInventoryItems_skips_entries_with_empty_normalized_triple()
+    {
+        var raw = new List<Dictionary<string, object>>
+        {
+            new Dictionary<string, object> { ["normalizedName"] = "good", ["normalizedPublisher"] = "pub", ["normalizedVersion"] = "1.0" },
+            new Dictionary<string, object> { ["displayName"] = "Empty triple — should be skipped" },
+        };
+
+        var items = VulnerabilityCorrelationService.ParseSoftwareInventoryItems(raw, cpeMappings: null);
+
+        Assert.Single(items);
+        Assert.Equal("good", items[0].NormalizedName);
+    }
+
+    [Fact]
+    public async Task ApplyInventoryContributionsAsync_calls_increment_per_added_item_in_delta()
+    {
+        var vulnRepo = new Mock<IVulnerabilityRepository>();
+        vulnRepo
+            .Setup(r => r.UpsertSessionInventoryContributionsAsync(
+                TenantId, SessionId, It.IsAny<IReadOnlyList<SoftwareInventoryItem>>(), It.IsAny<CancellationToken>()))
+            .ReturnsAsync(new InventoryContributionsDelta
+            {
+                FirstTime = true,
+                AddedItems = new List<SoftwareInventoryItem>
+                {
+                    new SoftwareInventoryItem { NormalizedVendor = "v1", NormalizedName = "n1", NormalizedVersion = "1.0" },
+                    new SoftwareInventoryItem { NormalizedVendor = "v2", NormalizedName = "n2", NormalizedVersion = "2.0" },
+                },
+                RemovedKeys = new List<DeletionDecrementKey>(),
+            });
+
+        var service = NewService(vulnRepo);
+        var raw = new List<Dictionary<string, object>>
+        {
+            new Dictionary<string, object> { ["normalizedPublisher"] = "v1", ["normalizedName"] = "n1", ["normalizedVersion"] = "1.0" },
+            new Dictionary<string, object> { ["normalizedPublisher"] = "v2", ["normalizedName"] = "n2", ["normalizedVersion"] = "2.0" },
+        };
+
+        await service.ApplyInventoryContributionsAsync(TenantId, SessionId, raw, cpeMappings: null);
+
+        vulnRepo.Verify(r => r.IncrementSoftwareInventoryEntryAsync(
+            TenantId,
+            It.Is<SoftwareInventoryItem>(i => i.NormalizedName == "n1"),
+            SessionId,
+            It.IsAny<CancellationToken>()), Times.Once);
+        vulnRepo.Verify(r => r.IncrementSoftwareInventoryEntryAsync(
+            TenantId,
+            It.Is<SoftwareInventoryItem>(i => i.NormalizedName == "n2"),
+            SessionId,
+            It.IsAny<CancellationToken>()), Times.Once);
+        vulnRepo.Verify(r => r.DecrementSoftwareInventoryEntryAsync(
+            It.IsAny<string>(), It.IsAny<string>(), It.IsAny<string>(), It.IsAny<string>(), It.IsAny<CancellationToken>()),
+            Times.Never);
+    }
+
+    [Fact]
+    public async Task ApplyInventoryContributionsAsync_calls_decrement_per_removed_key_in_delta()
+    {
+        var vulnRepo = new Mock<IVulnerabilityRepository>();
+        vulnRepo
+            .Setup(r => r.UpsertSessionInventoryContributionsAsync(
+                TenantId, SessionId, It.IsAny<IReadOnlyList<SoftwareInventoryItem>>(), It.IsAny<CancellationToken>()))
+            .ReturnsAsync(new InventoryContributionsDelta
+            {
+                FirstTime = false,
+                AddedItems = new List<SoftwareInventoryItem>(),
+                RemovedKeys = new List<DeletionDecrementKey>
+                {
+                    new DeletionDecrementKey { Vendor = "vOld", Name = "nOld", Version = "0.9" },
+                },
+            });
+
+        var service = NewService(vulnRepo);
+        var raw = new List<Dictionary<string, object>>
+        {
+            new Dictionary<string, object> { ["normalizedPublisher"] = "v1", ["normalizedName"] = "n1", ["normalizedVersion"] = "1.0" },
+        };
+
+        await service.ApplyInventoryContributionsAsync(TenantId, SessionId, raw, cpeMappings: null);
+
+        vulnRepo.Verify(r => r.DecrementSoftwareInventoryEntryAsync(
+            TenantId, "vOld", "nOld", "0.9", It.IsAny<CancellationToken>()), Times.Once);
+        vulnRepo.Verify(r => r.IncrementSoftwareInventoryEntryAsync(
+            It.IsAny<string>(), It.IsAny<SoftwareInventoryItem>(), It.IsAny<string>(), It.IsAny<CancellationToken>()),
+            Times.Never);
+    }
+
+    [Fact]
+    public async Task ApplyInventoryContributionsAsync_returns_without_calling_repo_for_empty_input()
+    {
+        var vulnRepo = new Mock<IVulnerabilityRepository>(MockBehavior.Strict);
+        // Strict mock — any unexpected call fails the test.
+
+        var service = NewService(vulnRepo);
+        await service.ApplyInventoryContributionsAsync(TenantId, SessionId, new List<Dictionary<string, object>>(), null);
+
+        vulnRepo.Verify(r => r.UpsertSessionInventoryContributionsAsync(
+            It.IsAny<string>(), It.IsAny<string>(), It.IsAny<IReadOnlyList<SoftwareInventoryItem>>(), It.IsAny<CancellationToken>()),
+            Times.Never);
+    }
+
+    [Fact]
+    public async Task ApplyInventoryContributionsAsync_swallows_side_row_helper_exceptions_for_non_fatal_logging()
+    {
+        // Helpers are fail-loud (per memory feedback_storage_helpers_fail_soft) but the caller
+        // intentionally logs+swallows the side-row CAS exhaustion so the primary VulnerabilityReport
+        // stays stored. Drift is the documented trade-off (§17.3 "Side-row write before counter writes").
+        var vulnRepo = new Mock<IVulnerabilityRepository>();
+        vulnRepo
+            .Setup(r => r.UpsertSessionInventoryContributionsAsync(
+                TenantId, SessionId, It.IsAny<IReadOnlyList<SoftwareInventoryItem>>(), It.IsAny<CancellationToken>()))
+            .ThrowsAsync(new InvalidOperationException("SessionInventoryContributions ETag-CAS exhausted ..."));
+
+        var service = NewService(vulnRepo);
+        var raw = new List<Dictionary<string, object>>
+        {
+            new Dictionary<string, object> { ["normalizedPublisher"] = "v1", ["normalizedName"] = "n1", ["normalizedVersion"] = "1.0" },
+        };
+
+        // Should not throw.
+        await service.ApplyInventoryContributionsAsync(TenantId, SessionId, raw, cpeMappings: null);
+
+        vulnRepo.Verify(r => r.IncrementSoftwareInventoryEntryAsync(
+            It.IsAny<string>(), It.IsAny<SoftwareInventoryItem>(), It.IsAny<string>(), It.IsAny<CancellationToken>()),
+            Times.Never);
+    }
+
+    [Fact]
+    public async Task ApplyInventoryContributionsAsync_continues_after_per_counter_failure_and_audits_each()
+    {
+        // Codex Round 2: a single Increment failure must NOT abort the rest of the loop —
+        // the side-row already promises the new key set, so subsequent runs would see an empty
+        // delta and never retry the under-counted keys. Per Plan §17.3 + §17.8, each failure
+        // emits an inventory_counter_write_failed audit + we keep processing the rest.
+        var vulnRepo = new Mock<IVulnerabilityRepository>();
+        vulnRepo
+            .Setup(r => r.UpsertSessionInventoryContributionsAsync(
+                TenantId, SessionId, It.IsAny<IReadOnlyList<SoftwareInventoryItem>>(), It.IsAny<CancellationToken>()))
+            .ReturnsAsync(new InventoryContributionsDelta
+            {
+                FirstTime = true,
+                AddedItems = new List<SoftwareInventoryItem>
+                {
+                    new SoftwareInventoryItem { NormalizedVendor = "v1", NormalizedName = "n1", NormalizedVersion = "1.0" },
+                    new SoftwareInventoryItem { NormalizedVendor = "v2", NormalizedName = "n2", NormalizedVersion = "2.0" },
+                    new SoftwareInventoryItem { NormalizedVendor = "v3", NormalizedName = "n3", NormalizedVersion = "3.0" },
+                },
+                RemovedKeys = new List<DeletionDecrementKey>
+                {
+                    new DeletionDecrementKey { Vendor = "vOld", Name = "nOld", Version = "0.9" },
+                },
+            });
+
+        // Increment of (v1,n1,1.0) succeeds, (v2,n2,2.0) THROWS, (v3,n3,3.0) succeeds.
+        vulnRepo.Setup(r => r.IncrementSoftwareInventoryEntryAsync(
+                TenantId, It.Is<SoftwareInventoryItem>(i => i.NormalizedName == "n2"),
+                SessionId, It.IsAny<CancellationToken>()))
+            .ThrowsAsync(new InvalidOperationException("SoftwareInventory increment ETag-CAS exhausted"));
+
+        var maintenanceRepo = new Mock<IMaintenanceRepository>();
+        maintenanceRepo.Setup(m => m.LogAuditEntryAsync(
+                It.IsAny<string>(), It.IsAny<string>(), It.IsAny<string>(),
+                It.IsAny<string>(), It.IsAny<string>(), It.IsAny<Dictionary<string, string>?>()))
+            .ReturnsAsync(true);
+
+        var service = NewService(vulnRepo, maintenanceRepo);
+        var raw = new List<Dictionary<string, object>>
+        {
+            new Dictionary<string, object> { ["normalizedPublisher"] = "v1", ["normalizedName"] = "n1", ["normalizedVersion"] = "1.0" },
+            new Dictionary<string, object> { ["normalizedPublisher"] = "v2", ["normalizedName"] = "n2", ["normalizedVersion"] = "2.0" },
+            new Dictionary<string, object> { ["normalizedPublisher"] = "v3", ["normalizedName"] = "n3", ["normalizedVersion"] = "3.0" },
+        };
+
+        // Should NOT throw.
+        await service.ApplyInventoryContributionsAsync(TenantId, SessionId, raw, cpeMappings: null);
+
+        // All three increments attempted.
+        vulnRepo.Verify(r => r.IncrementSoftwareInventoryEntryAsync(
+            TenantId, It.Is<SoftwareInventoryItem>(i => i.NormalizedName == "n1"),
+            SessionId, It.IsAny<CancellationToken>()), Times.Once);
+        vulnRepo.Verify(r => r.IncrementSoftwareInventoryEntryAsync(
+            TenantId, It.Is<SoftwareInventoryItem>(i => i.NormalizedName == "n2"),
+            SessionId, It.IsAny<CancellationToken>()), Times.Once);
+        vulnRepo.Verify(r => r.IncrementSoftwareInventoryEntryAsync(
+            TenantId, It.Is<SoftwareInventoryItem>(i => i.NormalizedName == "n3"),
+            SessionId, It.IsAny<CancellationToken>()), Times.Once);
+        // Decrement still runs after the loop continues past the failed Increment.
+        vulnRepo.Verify(r => r.DecrementSoftwareInventoryEntryAsync(
+            TenantId, "vOld", "nOld", "0.9", It.IsAny<CancellationToken>()), Times.Once);
+
+        // Audit emitted exactly once for the failed (v2,n2,2.0) Increment.
+        maintenanceRepo.Verify(m => m.LogAuditEntryAsync(
+            TenantId,
+            "inventory_counter_write_failed",
+            "Session",
+            SessionId,
+            "system",
+            It.Is<Dictionary<string, string>?>(d =>
+                d != null
+                && d["operation"] == "increment"
+                && d["vendor"] == "v2"
+                && d["name"] == "n2"
+                && d["version"] == "2.0")),
+            Times.Once);
+    }
+
+    [Fact]
+    public async Task ApplyInventoryContributionsAsync_audits_decrement_failures_too()
+    {
+        var vulnRepo = new Mock<IVulnerabilityRepository>();
+        vulnRepo
+            .Setup(r => r.UpsertSessionInventoryContributionsAsync(
+                TenantId, SessionId, It.IsAny<IReadOnlyList<SoftwareInventoryItem>>(), It.IsAny<CancellationToken>()))
+            .ReturnsAsync(new InventoryContributionsDelta
+            {
+                AddedItems = new List<SoftwareInventoryItem>(),
+                RemovedKeys = new List<DeletionDecrementKey>
+                {
+                    new DeletionDecrementKey { Vendor = "vDead", Name = "nDead", Version = "0.5" },
+                },
+            });
+
+        vulnRepo.Setup(r => r.DecrementSoftwareInventoryEntryAsync(
+                TenantId, "vDead", "nDead", "0.5", It.IsAny<CancellationToken>()))
+            .ThrowsAsync(new InvalidOperationException("Decrement ETag-CAS exhausted"));
+
+        var maintenanceRepo = new Mock<IMaintenanceRepository>();
+        maintenanceRepo.Setup(m => m.LogAuditEntryAsync(
+                It.IsAny<string>(), It.IsAny<string>(), It.IsAny<string>(),
+                It.IsAny<string>(), It.IsAny<string>(), It.IsAny<Dictionary<string, string>?>()))
+            .ReturnsAsync(true);
+
+        var service = NewService(vulnRepo, maintenanceRepo);
+        var raw = new List<Dictionary<string, object>>
+        {
+            new Dictionary<string, object> { ["normalizedPublisher"] = "v1", ["normalizedName"] = "n1", ["normalizedVersion"] = "1.0" },
+        };
+
+        await service.ApplyInventoryContributionsAsync(TenantId, SessionId, raw, cpeMappings: null);
+
+        maintenanceRepo.Verify(m => m.LogAuditEntryAsync(
+            TenantId,
+            "inventory_counter_write_failed",
+            "Session",
+            SessionId,
+            "system",
+            It.Is<Dictionary<string, string>?>(d =>
+                d != null
+                && d["operation"] == "decrement"
+                && d["vendor"] == "vDead"
+                && d["name"] == "nDead"
+                && d["version"] == "0.5")),
+            Times.Once);
+    }
+
+    private static VulnerabilityCorrelationService NewService(
+        Mock<IVulnerabilityRepository> vulnRepo, Mock<IMaintenanceRepository>? maintenanceRepo = null)
+    {
+        return new VulnerabilityCorrelationService(
+            vulnRepo.Object,
+            sessionRepo: Mock.Of<ISessionRepository>(),
+            nvdApi: null!,
+            kevService: null!,
+            msrcCorrelation: null!,
+            adminConfig: null!,
+            telemetryClient: null,
+            logger: NullLogger<VulnerabilityCorrelationService>.Instance,
+            maintenanceRepo: maintenanceRepo?.Object);
+    }
+}

src/Backend/AutopilotMonitor.Functions/DataAccess/TableStorage/TableVulnerabilityRepository.cs

@@ -1,7 +1,9 @@
+using System.Threading;
 using Azure;
 using Azure.Data.Tables;
 using AutopilotMonitor.Shared;
 using AutopilotMonitor.Shared.DataAccess;
+using AutopilotMonitor.Shared.Models.Deletion;
 using AutopilotMonitor.Functions.Services;
 using Microsoft.Extensions.Logging;
 
@@ -69,6 +71,27 @@ namespace AutopilotMonitor.Functions.DataAccess.TableStorage
             return results;
         }
 
+        // --- Software Inventory v2 (Plan §17 contributions side-row + delta-driven counters) ---
+
+        public Task<IReadOnlyList<DeletionDecrementKey>?> GetSessionInventoryContributionsAsync(
+            string tenantId, string sessionId, CancellationToken ct = default)
+            => _storage.GetSessionInventoryContributionsAsync(tenantId, sessionId, ct);
+
+        public Task<InventoryContributionsDelta> UpsertSessionInventoryContributionsAsync(
+            string tenantId, string sessionId, IReadOnlyList<SoftwareInventoryItem> newItems, CancellationToken ct = default)
+            => _storage.UpsertSessionInventoryContributionsAsync(tenantId, sessionId, newItems, ct);
+
+        public Task DeleteSessionInventoryContributionsAsync(string tenantId, string sessionId, CancellationToken ct = default)
+            => _storage.DeleteSessionInventoryContributionsAsync(tenantId, sessionId, ct);
+
+        public Task IncrementSoftwareInventoryEntryAsync(
+            string tenantId, SoftwareInventoryItem item, string sessionId, CancellationToken ct = default)
+            => _storage.IncrementSoftwareInventoryEntryAsync(tenantId, item, sessionId, ct);
+
+        public Task DecrementSoftwareInventoryEntryAsync(
+            string tenantId, string normalizedVendor, string normalizedName, string normalizedVersion, CancellationToken ct = default)
+            => _storage.DecrementSoftwareInventoryEntryAsync(tenantId, normalizedVendor, normalizedName, normalizedVersion, ct);
+
         public async Task UpdateSoftwareInventoryCpeAsync(string normalizedName, string cpeUri)
         {
             var tableClient = _storage.GetTableClient(Constants.TableNames.SoftwareInventory);

src/Backend/AutopilotMonitor.Functions/Services/TableStorageService.Deletion.cs

@@ -1,13 +1,39 @@
+using System;
 using System.Collections.Generic;
+using System.Linq;
 using System.Runtime.CompilerServices;
 using System.Threading;
 using System.Threading.Tasks;
 using Azure;
 using Azure.Data.Tables;
+using AutopilotMonitor.Functions.Security;
 using AutopilotMonitor.Functions.Services.Deletion;
 
 namespace AutopilotMonitor.Functions.Services
 {
+    /// <summary>
+    /// Result of <see cref="TableStorageService.DeleteByExactKeysInBatchesAsync"/> — accurate
+    /// progress accounting for the cascade worker. <see cref="Attempted"/> is the input key
+    /// count, <see cref="DeletedNow"/> is the number of rows that the helper actually removed
+    /// in this call, <see cref="AlreadyMissing"/> is the number that returned 404 (idempotent
+    /// re-run of a partially-completed cascade).
+    /// </summary>
+    public class DeletionBatchResult
+    {
+        public int Attempted { get; }
+        public int DeletedNow { get; }
+        public int AlreadyMissing { get; }
+
+        public DeletionBatchResult(int attempted, int deletedNow, int alreadyMissing)
+        {
+            Attempted = attempted;
+            DeletedNow = deletedNow;
+            AlreadyMissing = alreadyMissing;
+        }
+
+        public static readonly DeletionBatchResult Empty = new DeletionBatchResult(0, 0, 0);
+    }
+
     /// <summary>
     /// Cascade-deletion read surface. Implements <see cref="ISessionDeletionInventoryReader"/>
     /// against the existing <c>_tableServiceClient</c>. Pure delegation; fail-loud on every
@@ -68,5 +94,167 @@ namespace AutopilotMonitor.Functions.Services
                 return null;
             }
         }
+
+        // ============================================================ Delete helpers (PR2) ====
+
+        // Azure Tables batch transactions cap at 100 actions per submission, all sharing a PartitionKey.
+        private const int BatchActionLimit = 100;
+
+        /// <summary>
+        /// Idempotent batched deletion by exact <c>(PK, RK)</c> keys. Groups by PartitionKey
+        /// (batch transactions are partition-scoped), chunks each group into <see cref="BatchActionLimit"/>-row
+        /// transactions, and falls back to per-row <see cref="TableClient.DeleteEntityAsync(string, string, ETag, CancellationToken)"/>
+        /// with 404-ignore when the whole-batch rollback hides which row was already missing.
+        /// Returns <see cref="DeletionBatchResult"/> for accurate progress accounting on
+        /// re-runs of partially-completed cascades.
+        /// </summary>
+        public async Task<DeletionBatchResult> DeleteByExactKeysInBatchesAsync(
+            string tableName,
+            IReadOnlyList<(string Pk, string Rk)> keys,
+            CancellationToken cancellationToken = default)
+        {
+            if (keys == null) throw new ArgumentNullException(nameof(keys));
+            var attempted = keys.Count;
+            if (attempted == 0) return DeletionBatchResult.Empty;
+
+            var tableClient = _tableServiceClient.GetTableClient(tableName);
+            var deletedNow = 0;
+            var alreadyMissing = 0;
+
+            foreach (var group in keys.GroupBy(k => k.Pk, StringComparer.Ordinal))
+            {
+                var rowsInGroup = group.ToList();
+                for (var i = 0; i < rowsInGroup.Count; i += BatchActionLimit)
+                {
+                    var chunk = rowsInGroup.Skip(i).Take(BatchActionLimit).ToList();
+                    var actions = chunk
+                        .Select(k => new TableTransactionAction(
+                            TableTransactionActionType.Delete,
+                            new TableEntity(k.Pk, k.Rk) { ETag = ETag.All }))
+                        .ToList();
+
+                    try
+                    {
+                        await tableClient.SubmitTransactionAsync(actions, cancellationToken);
+                        deletedNow += chunk.Count;
+                    }
+                    catch (RequestFailedException ex) when (ex.Status == 404)
+                    {
+                        // Azure rolls back the entire transaction when a single Delete returns 404.
+                        // Fall back to per-row delete with 404-ignore so the helper stays idempotent.
+                        foreach (var (pk, rk) in chunk)
+                        {
+                            cancellationToken.ThrowIfCancellationRequested();
+                            try
+                            {
+                                await tableClient.DeleteEntityAsync(pk, rk, ETag.All, cancellationToken);
+                                deletedNow++;
+                            }
+                            catch (RequestFailedException rfe) when (rfe.Status == 404)
+                            {
+                                alreadyMissing++;
+                            }
+                        }
+                    }
+                }
+            }
+
+            return new DeletionBatchResult(attempted, deletedNow, alreadyMissing);
+        }
+
+        // ---------------- DISCRIMINATOR_PK_RK_SUFFIX / DISCRIMINATOR_PK_RK_EXACT --------------
+
+        public Task<DeletionBatchResult> DeleteEventTypeIndexEntriesAsync(
+            string tenantId, string sessionId, IReadOnlyList<(string Pk, string Rk)> keys, CancellationToken ct = default)
+        {
+            SecurityValidator.EnsureValidGuid(tenantId, nameof(tenantId));
+            SecurityValidator.EnsureValidGuid(sessionId, nameof(sessionId));
+            return DeleteByExactKeysInBatchesAsync(Shared.Constants.TableNames.EventTypeIndex, keys, ct);
+        }
+
+        public Task<DeletionBatchResult> DeleteCveIndexEntriesAsync(
+            string tenantId, string sessionId, IReadOnlyList<(string Pk, string Rk)> keys, CancellationToken ct = default)
+        {
+            SecurityValidator.EnsureValidGuid(tenantId, nameof(tenantId));
+            SecurityValidator.EnsureValidGuid(sessionId, nameof(sessionId));
+            return DeleteByExactKeysInBatchesAsync(Shared.Constants.TableNames.CveIndex, keys, ct);
+        }
+
+        // ---------------- DISCRIMINATOR_PK_PROP -----------------------------------------------
+
+        public Task<DeletionBatchResult> DeleteSessionsByTerminalAsync(
+            IReadOnlyList<(string Pk, string Rk)> keys, CancellationToken ct = default)
+            => DeleteByExactKeysInBatchesAsync(Shared.Constants.TableNames.SessionsByTerminal, keys, ct);
+
+        public Task<DeletionBatchResult> DeleteSessionsByStageAsync(
+            IReadOnlyList<(string Pk, string Rk)> keys, CancellationToken ct = default)
+            => DeleteByExactKeysInBatchesAsync(Shared.Constants.TableNames.SessionsByStage, keys, ct);
+
+        public Task<DeletionBatchResult> DeleteDeadEndsByReasonAsync(
+            IReadOnlyList<(string Pk, string Rk)> keys, CancellationToken ct = default)
+            => DeleteByExactKeysInBatchesAsync(Shared.Constants.TableNames.DeadEndsByReason, keys, ct);
+
+        public Task<DeletionBatchResult> DeleteClassifierVerdictsByIdLevelAsync(
+            IReadOnlyList<(string Pk, string Rk)> keys, CancellationToken ct = default)
+            => DeleteByExactKeysInBatchesAsync(Shared.Constants.TableNames.ClassifierVerdictsByIdLevel, keys, ct);
+
+        public Task<DeletionBatchResult> DeleteSignalsByKindAsync(
+            IReadOnlyList<(string Pk, string Rk)> keys, CancellationToken ct = default)
+            => DeleteByExactKeysInBatchesAsync(Shared.Constants.TableNames.SignalsByKind, keys, ct);
+
+        // ---------------- PK_RK_EXACT (single-row) --------------------------------------------
+
+        public async Task DeleteDeviceSnapshotAsync(string tenantId, string sessionId, CancellationToken ct = default)
+        {
+            SecurityValidator.EnsureValidGuid(tenantId, nameof(tenantId));
+            SecurityValidator.EnsureValidGuid(sessionId, nameof(sessionId));
+            var tableClient = _tableServiceClient.GetTableClient(Shared.Constants.TableNames.DeviceSnapshot);
+            try
+            {
+                await tableClient.DeleteEntityAsync(tenantId, sessionId, ETag.All, ct);
+            }
+            catch (RequestFailedException ex) when (ex.Status == 404)
+            {
+                // Idempotent — already gone.
+            }
+        }
+
+        // ---------------- PK_BY_SESSION (enumerate-then-delete) -------------------------------
+
+        public Task<DeletionBatchResult> DeleteSessionSignalsAsync(string tenantId, string sessionId, CancellationToken ct = default)
+            => DeletePkBySessionAsync(Shared.Constants.TableNames.Signals, tenantId, sessionId, ct);
+
+        public Task<DeletionBatchResult> DeleteSessionDecisionTransitionsAsync(string tenantId, string sessionId, CancellationToken ct = default)
+            => DeletePkBySessionAsync(Shared.Constants.TableNames.DecisionTransitions, tenantId, sessionId, ct);
+
+        /// <summary>
+        /// Enumerates every row in <paramref name="tableName"/> with
+        /// <c>PartitionKey == {tenantId}_{sessionId}</c> and deletes them via
+        /// <see cref="DeleteByExactKeysInBatchesAsync"/>. Used for PK_BY_SESSION-class cascade
+        /// targets when the caller does not yet have a manifest in hand (e.g. legacy direct-delete
+        /// paths or test fixtures); the §1 P6 cascade worker passes manifest keys to
+        /// <see cref="DeleteByExactKeysInBatchesAsync"/> directly.
+        /// </summary>
+        private async Task<DeletionBatchResult> DeletePkBySessionAsync(
+            string tableName, string tenantId, string sessionId, CancellationToken ct)
+        {
+            SecurityValidator.EnsureValidGuid(tenantId, nameof(tenantId));
+            SecurityValidator.EnsureValidGuid(sessionId, nameof(sessionId));
+
+            var tableClient = _tableServiceClient.GetTableClient(tableName);
+            var partitionKey = $"{tenantId}_{sessionId}";
+            var safePk = ODataSanitizer.EscapeValue(partitionKey);
+
+            var keys = new List<(string Pk, string Rk)>();
+            await foreach (var entity in tableClient.QueryAsync<TableEntity>(
+                filter: $"PartitionKey eq '{safePk}'",
+                select: new[] { "PartitionKey", "RowKey" },
+                cancellationToken: ct))
+            {
+                keys.Add((entity.PartitionKey, entity.RowKey));
+            }
+
+            return await DeleteByExactKeysInBatchesAsync(tableName, keys, ct);
+        }
     }
 }

src/Backend/AutopilotMonitor.Functions/Services/TableStorageService.Inventory.cs

@@ -0,0 +1,380 @@
+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.Security;
+using AutopilotMonitor.Shared;
+using AutopilotMonitor.Shared.Models.Deletion;
+
+namespace AutopilotMonitor.Functions.Services
+{
+    /// <summary>
+    /// SoftwareInventory side-row + counter helpers (Plan §17). All methods are fail-loud per
+    /// memory <c>feedback_storage_helpers_fail_soft</c> — non-404 storage errors propagate;
+    /// bounded retry exhaustion (10 attempts / 60s wall-clock per §12-Q10) throws
+    /// <see cref="InvalidOperationException"/>.
+    /// </summary>
+    public partial class TableStorageService
+    {
+        // Bounded-retry budget for every ETag-CAS / AddEntity-race loop in this file.
+        // §12-Q10: 10 attempts OR 60s wall-clock, whichever first → poison.
+        private const int InventoryCasMaxAttempts = 10;
+        private static readonly TimeSpan InventoryCasMaxWallClock = TimeSpan.FromSeconds(60);
+
+        // ============================================================ SessionInventoryContributions ====
+
+        /// <summary>
+        /// Reads the side-row by (tenantId, sessionId). Returns null on 404 (pre-side-row session
+        /// or never written). Fail-loud on any other storage error.
+        /// </summary>
+        public async Task<IReadOnlyList<DeletionDecrementKey>?> GetSessionInventoryContributionsAsync(
+            string tenantId, string sessionId, CancellationToken ct = default)
+        {
+            SecurityValidator.EnsureValidGuid(tenantId, nameof(tenantId));
+            SecurityValidator.EnsureValidGuid(sessionId, nameof(sessionId));
+
+            var tableClient = _tableServiceClient.GetTableClient(Constants.TableNames.SessionInventoryContributions);
+            try
+            {
+                var response = await tableClient.GetEntityAsync<TableEntity>(tenantId, sessionId, cancellationToken: ct);
+                var encoded = response.Value.GetString("SoftwareKeysJson") ?? string.Empty;
+                var compressed = response.Value.GetBoolean("IsCompressed") ?? false;
+                return SoftwareKeysJsonCodec.Decode(encoded, compressed);
+            }
+            catch (RequestFailedException ex) when (ex.Status == 404)
+            {
+                return null;
+            }
+        }
+
+        /// <summary>
+        /// Direct delete with 404-ignore. Cascade tombstone calls this after consuming the
+        /// side-row's keys for the decrement step (Plan §17.5).
+        /// </summary>
+        public async Task DeleteSessionInventoryContributionsAsync(
+            string tenantId, string sessionId, CancellationToken ct = default)
+        {
+            SecurityValidator.EnsureValidGuid(tenantId, nameof(tenantId));
+            SecurityValidator.EnsureValidGuid(sessionId, nameof(sessionId));
+
+            var tableClient = _tableServiceClient.GetTableClient(Constants.TableNames.SessionInventoryContributions);
+            try
+            {
+                await tableClient.DeleteEntityAsync(tenantId, sessionId, ETag.All, ct);
+            }
+            catch (RequestFailedException ex) when (ex.Status == 404)
+            {
+                // Already gone — idempotent.
+            }
+        }
+
+        /// <summary>
+        /// At-most-once-per-session, delta-update upsert. Plan §17.3 algorithm:
+        /// <list type="number">
+        ///   <item>Read existing side-row.</item>
+        ///   <item>Compute newKeys = items.Select(ToDecrementKey).</item>
+        ///   <item>If no side-row: AddEntity (full payload). On 409 → race lost, restart loop.</item>
+        ///   <item>If side-row exists: decode old keys, compute (added, removed). If both empty: no-op return.
+        ///         Otherwise UpdateEntity with ETag-Replace. On 412 → restart loop with fresh read.</item>
+        /// </list>
+        /// Bounded retry exhausted → throw <see cref="InvalidOperationException"/>.
+        /// </summary>
+        public async Task<InventoryContributionsDelta> UpsertSessionInventoryContributionsAsync(
+            string tenantId, string sessionId,
+            IReadOnlyList<SoftwareInventoryItem> newItems,
+            CancellationToken ct = default)
+        {
+            SecurityValidator.EnsureValidGuid(tenantId, nameof(tenantId));
+            SecurityValidator.EnsureValidGuid(sessionId, nameof(sessionId));
+            if (newItems == null) throw new ArgumentNullException(nameof(newItems));
+
+            var tableClient = _tableServiceClient.GetTableClient(Constants.TableNames.SessionInventoryContributions);
+
+            // Deduplicate by composite key — upstream may pass the same normalization twice.
+            var newKeys = new Dictionary<string, SoftwareInventoryItem>(StringComparer.OrdinalIgnoreCase);
+            foreach (var item in newItems)
+            {
+                if (string.IsNullOrEmpty(item.NormalizedVendor)
+                    && string.IsNullOrEmpty(item.NormalizedName)
+                    && string.IsNullOrEmpty(item.NormalizedVersion))
+                {
+                    continue; // skip empty keys; matches existing UpsertSoftwareInventory dedup
+                }
+                var composite = SoftwareKeysJsonCodec.CompositeKey(item.NormalizedVendor, item.NormalizedName, item.NormalizedVersion);
+                newKeys[composite] = item;
+            }
+            var newKeyList = newKeys.Values.Select(i => i.ToDecrementKey()).ToList();
+
+            var deadline = DateTime.UtcNow + InventoryCasMaxWallClock;
+            for (var attempt = 0; attempt < InventoryCasMaxAttempts && DateTime.UtcNow < deadline; attempt++)
+            {
+                ct.ThrowIfCancellationRequested();
+
+                TableEntity? existing = null;
+                try
+                {
+                    var response = await tableClient.GetEntityAsync<TableEntity>(tenantId, sessionId, cancellationToken: ct);
+                    existing = response.Value;
+                }
+                catch (RequestFailedException ex) when (ex.Status == 404)
+                {
+                    // First-time write path.
+                }
+
+                var encoded = SoftwareKeysJsonCodec.Encode(newKeyList);
+                var now = DateTime.UtcNow;
+
+                if (existing == null)
+                {
+                    var row = new TableEntity(tenantId, sessionId)
+                    {
+                        ["SoftwareKeysJson"] = encoded.Encoded,
+                        ["IsCompressed"]     = encoded.IsCompressed,
+                        ["CountedAt"]        = now,
+                        ["LastUpdatedAt"]    = now,
+                        ["KeyCount"]         = encoded.KeyCount,
+                    };
+                    try
+                    {
+                        await tableClient.AddEntityAsync(row, ct);
+                        return new InventoryContributionsDelta
+                        {
+                            AddedItems = newKeys.Values.ToList(),
+                            RemovedKeys = new List<DeletionDecrementKey>(),
+                            FirstTime = true,
+                        };
+                    }
+                    catch (RequestFailedException ex) when (ex.Status == 409)
+                    {
+                        continue; // race lost — another writer beat us to first-insert
+                    }
+                }
+                else
+                {
+                    var oldEncoded = existing.GetString("SoftwareKeysJson") ?? string.Empty;
+                    var oldCompressed = existing.GetBoolean("IsCompressed") ?? false;
+                    var oldKeys = SoftwareKeysJsonCodec.Decode(oldEncoded, oldCompressed);
+
+                    var (added, removed) = SoftwareKeysJsonCodec.ComputeDelta(oldKeys, newKeyList);
+                    if (added.Count == 0 && removed.Count == 0)
+                    {
+                        return new InventoryContributionsDelta
+                        {
+                            AddedItems = new List<SoftwareInventoryItem>(),
+                            RemovedKeys = new List<DeletionDecrementKey>(),
+                            FirstTime = false,
+                        };
+                    }
+
+                    existing["SoftwareKeysJson"] = encoded.Encoded;
+                    existing["IsCompressed"]     = encoded.IsCompressed;
+                    existing["LastUpdatedAt"]    = now;
+                    existing["KeyCount"]         = encoded.KeyCount;
+
+                    try
+                    {
+                        await tableClient.UpdateEntityAsync(existing, existing.ETag, TableUpdateMode.Replace, ct);
+                        var addedItems = newKeys
+                            .Where(kv => added.Contains(kv.Key))
+                            .Select(kv => kv.Value)
+                            .ToList();
+                        var removedKeys = oldKeys
+                            .Where(k => removed.Contains(SoftwareKeysJsonCodec.CompositeKey(k)))
+                            .ToList();
+                        return new InventoryContributionsDelta
+                        {
+                            AddedItems = addedItems,
+                            RemovedKeys = removedKeys,
+                            FirstTime = false,
+                        };
+                    }
+                    catch (RequestFailedException ex) when (ex.Status == 412)
+                    {
+                        continue; // ETag conflict — re-read on next attempt
+                    }
+                }
+            }
+
+            throw new InvalidOperationException(
+                $"SessionInventoryContributions ETag-CAS exhausted after {InventoryCasMaxAttempts} attempts or "
+                + $"{InventoryCasMaxWallClock.TotalSeconds:0}s wall-clock. tenant={tenantId} session={sessionId}");
+        }
+
+        // ============================================================ SoftwareInventory counters ====
+
+        /// <summary>
+        /// ETag-CAS <c>SessionCount += 1</c> for the (tenantId, normalizedKey) row, creating the
+        /// row with full <paramref name="item"/> metadata on first encounter. Bounded retry per
+        /// <see cref="InventoryCasMaxAttempts"/> / <see cref="InventoryCasMaxWallClock"/>.
+        /// </summary>
+        public async Task IncrementSoftwareInventoryEntryAsync(
+            string tenantId, SoftwareInventoryItem item, string sessionId, CancellationToken ct = default)
+        {
+            SecurityValidator.EnsureValidGuid(tenantId, nameof(tenantId));
+            if (item == null) throw new ArgumentNullException(nameof(item));
+
+            var rowKey = BuildSoftwareInventoryRowKey(item.NormalizedVendor, item.NormalizedName, item.NormalizedVersion);
+            if (string.IsNullOrWhiteSpace(rowKey) || rowKey == "::") return; // skip empty triples — matches existing UpsertSoftwareInventory dedup
+
+            var tableClient = _tableServiceClient.GetTableClient(Constants.TableNames.SoftwareInventory);
+            var deadline = DateTime.UtcNow + InventoryCasMaxWallClock;
+
+            for (var attempt = 0; attempt < InventoryCasMaxAttempts && DateTime.UtcNow < deadline; attempt++)
+            {
+                ct.ThrowIfCancellationRequested();
+                var nowIso = DateTime.UtcNow.ToString("o");
+
+                TableEntity? existing = null;
+                try
+                {
+                    var response = await tableClient.GetEntityAsync<TableEntity>(tenantId, rowKey, cancellationToken: ct);
+                    existing = response.Value;
+                }
+                catch (RequestFailedException ex) when (ex.Status == 404)
+                {
+                    // First-time create path.
+                }
+
+                if (existing == null)
+                {
+                    var row = new TableEntity(tenantId, rowKey)
+                    {
+                        ["DisplayName"]            = item.DisplayName,
+                        ["NormalizedName"]         = item.NormalizedName,
+                        ["NormalizedVendor"]       = item.NormalizedVendor,
+                        ["NormalizedVersion"]      = item.NormalizedVersion,
+                        ["Publisher"]              = item.Publisher,
+                        ["RegistrySource"]         = item.RegistrySource,
+                        ["NormalizationConfidence"] = item.NormalizationConfidence,
+                        ["FirstSeenAt"]            = nowIso,
+                        ["LastSeenAt"]             = nowIso,
+                        ["FirstSessionId"]         = sessionId,
+                        ["LastSessionId"]          = sessionId,
+                        ["SessionCount"]           = 1,
+                        ["CpeUri"]                 = item.CpeUri ?? string.Empty,
+                    };
+                    try
+                    {
+                        await tableClient.AddEntityAsync(row, ct);
+                        return;
+                    }
+                    catch (RequestFailedException ex) when (ex.Status == 409)
+                    {
+                        continue; // race lost — another writer created it; reread + increment
+                    }
+                }
+                else
+                {
+                    existing["SessionCount"]  = (existing.GetInt32("SessionCount") ?? 0) + 1;
+                    existing["LastSeenAt"]    = nowIso;
+                    existing["LastSessionId"] = sessionId;
+                    // Stamp a CpeUri if we just got one and the row didn't have one yet — preserves
+                    // existing behaviour from UpsertSoftwareInventoryAsync.
+                    if (!string.IsNullOrEmpty(item.CpeUri) && string.IsNullOrEmpty(existing.GetString("CpeUri")))
+                    {
+                        existing["CpeUri"] = item.CpeUri;
+                    }
+
+                    try
+                    {
+                        await tableClient.UpdateEntityAsync(existing, existing.ETag, TableUpdateMode.Replace, ct);
+                        return;
+                    }
+                    catch (RequestFailedException ex) when (ex.Status == 412)
+                    {
+                        continue;
+                    }
+                }
+            }
+
+            throw new InvalidOperationException(
+                $"SoftwareInventory increment ETag-CAS exhausted after {InventoryCasMaxAttempts} attempts or "
+                + $"{InventoryCasMaxWallClock.TotalSeconds:0}s wall-clock. tenant={tenantId} key={rowKey}");
+        }
+
+        /// <summary>
+        /// ETag-CAS <c>SessionCount -= 1</c> with clamp at zero. Deletes the row when count hits
+        /// zero. 404 = idempotent no-op. Bounded retry per <see cref="InventoryCasMaxAttempts"/> /
+        /// <see cref="InventoryCasMaxWallClock"/>; throws on exhaustion.
+        /// </summary>
+        public async Task DecrementSoftwareInventoryEntryAsync(
+            string tenantId, string normalizedVendor, string normalizedName, string normalizedVersion,
+            CancellationToken ct = default)
+        {
+            SecurityValidator.EnsureValidGuid(tenantId, nameof(tenantId));
+            var rowKey = BuildSoftwareInventoryRowKey(normalizedVendor, normalizedName, normalizedVersion);
+            if (string.IsNullOrWhiteSpace(rowKey) || rowKey == "::") return;
+
+            var tableClient = _tableServiceClient.GetTableClient(Constants.TableNames.SoftwareInventory);
+            var deadline = DateTime.UtcNow + InventoryCasMaxWallClock;
+
+            for (var attempt = 0; attempt < InventoryCasMaxAttempts && DateTime.UtcNow < deadline; attempt++)
+            {
+                ct.ThrowIfCancellationRequested();
+
+                TableEntity existing;
+                try
+                {
+                    var response = await tableClient.GetEntityAsync<TableEntity>(tenantId, rowKey, cancellationToken: ct);
+                    existing = response.Value;
+                }
+                catch (RequestFailedException ex) when (ex.Status == 404)
+                {
+                    return; // already gone — decrement is idempotent
+                }
+
+                var newCount = Math.Max(0, (existing.GetInt32("SessionCount") ?? 0) - 1);
+                if (newCount == 0)
+                {
+                    try
+                    {
+                        await tableClient.DeleteEntityAsync(tenantId, rowKey, existing.ETag, ct);
+                        return;
+                    }
+                    catch (RequestFailedException ex) when (ex.Status == 412)
+                    {
+                        continue;
+                    }
+                    catch (RequestFailedException ex) when (ex.Status == 404)
+                    {
+                        return;
+                    }
+                }
+
+                existing["SessionCount"] = newCount;
+                existing["LastSeenAt"]   = DateTime.UtcNow.ToString("o");
+                try
+                {
+                    await tableClient.UpdateEntityAsync(existing, existing.ETag, TableUpdateMode.Replace, ct);
+                    return;
+                }
+                catch (RequestFailedException ex) when (ex.Status == 412)
+                {
+                    continue;
+                }
+            }
+
+            throw new InvalidOperationException(
+                $"SoftwareInventory decrement ETag-CAS exhausted after {InventoryCasMaxAttempts} attempts or "
+                + $"{InventoryCasMaxWallClock.TotalSeconds:0}s wall-clock. tenant={tenantId} key={rowKey}");
+        }
+
+        // ============================================================ Helpers ====
+
+        /// <summary>
+        /// SoftwareInventory RowKey format — preserves the same shape used by the legacy
+        /// <c>UpsertSoftwareInventoryAsync</c> path so existing rows remain reachable by the
+        /// new helpers. <c>SanitizeTableKey</c> lives in TableStorageService.Rules.cs as a
+        /// private static; partial classes share scope.
+        /// </summary>
+        private static string BuildSoftwareInventoryRowKey(string vendor, string name, string version)
+        {
+            var raw = $"{vendor ?? string.Empty}:{name ?? string.Empty}:{version ?? string.Empty}";
+            var sanitized = SanitizeTableKey(raw);
+            return sanitized.Length > 512 ? sanitized.Substring(0, 512) : sanitized;
+        }
+    }
+}

src/Backend/AutopilotMonitor.Functions/Services/TableStorageService.cs

@@ -56,6 +56,17 @@ namespace AutopilotMonitor.Functions.Services
             }
         }
 
+        /// <summary>
+        /// Test seam: construct directly from a (possibly Moq'd) <see cref="TableServiceClient"/>.
+        /// Used by xUnit so the storage-touching helpers in the partial classes (Deletion,
+        /// Inventory, …) can be exercised against the SDK's virtual surface without hitting Azure.
+        /// </summary>
+        internal TableStorageService(TableServiceClient tableServiceClient, ILogger<TableStorageService> logger)
+        {
+            _tableServiceClient = tableServiceClient;
+            _logger = logger;
+        }
+
         /// <summary>
         /// Returns a TableClient for the specified table name.
         /// Used by services that need direct table access (e.g. VulnerabilityCorrelationService).

src/Backend/AutopilotMonitor.Functions/Services/Vulnerability/VulnerabilityCorrelationService.cs

@@ -7,6 +7,7 @@ using System.Threading;
 using System.Threading.Tasks;
 using AutopilotMonitor.Functions.DataAccess.TableStorage;
 using AutopilotMonitor.Shared.DataAccess;
+using AutopilotMonitor.Shared.Models.Deletion;
 using Microsoft.ApplicationInsights;
 using Microsoft.Extensions.Logging;
 
@@ -16,6 +17,7 @@ namespace AutopilotMonitor.Functions.Services.Vulnerability
     {
         private readonly IVulnerabilityRepository _vulnRepo;
         private readonly ISessionRepository _sessionRepo;
+        private readonly IMaintenanceRepository? _maintenanceRepo;
         private readonly NvdApiClient _nvdApi;
         private readonly KevDataService _kevService;
         private readonly MsrcCorrelationService _msrcCorrelation;
@@ -42,10 +44,12 @@ namespace AutopilotMonitor.Functions.Services.Vulnerability
             MsrcCorrelationService msrcCorrelation,
             AdminConfigurationService adminConfig,
             TelemetryClient? telemetryClient,
-            ILogger<VulnerabilityCorrelationService> logger)
+            ILogger<VulnerabilityCorrelationService> logger,
+            IMaintenanceRepository? maintenanceRepo = null)
         {
             _vulnRepo = vulnRepo;
             _sessionRepo = sessionRepo;
+            _maintenanceRepo = maintenanceRepo;
             _nvdApi = nvdApi;
             _kevService = kevService;
             _msrcCorrelation = msrcCorrelation;
@@ -242,18 +246,13 @@ namespace AutopilotMonitor.Functions.Services.Vulnerability
                 return ra.CompareTo(rb);
             });
 
-            // Persist software inventory (fire-and-forget, non-blocking)
-            _ = Task.Run(async () =>
-            {
-                try
-                {
-                    await _vulnRepo.UpsertSoftwareInventoryAsync(tenantId, inventoryItems, sessionId, cpeMappingsDict);
-                }
-                catch (Exception ex)
-                {
-                    _logger.LogDebug(ex, "Failed to upsert software inventory (non-fatal)");
-                }
-            });
+            // Persist software inventory via the contributions side-row (Plan §17).
+            // Fire-and-forget so the report still gets stored if the side-row write hits a
+            // transient storage failure — the report is the primary product, inventory is the
+            // secondary tenant-aggregate signal. PR3 wires SessionDeletionGuard above this
+            // layer (at VulnerabilityCorrelateHandler) so a session locked by a cascade never
+            // reaches this code path; PR2's side-row write proceeds unguarded.
+            _ = Task.Run(() => ApplyInventoryContributionsAsync(tenantId, sessionId, inventoryItems, cpeMappingsDict));
 
             // Summary log for production monitoring
             _logger.LogInformation(
@@ -957,6 +956,162 @@ namespace AutopilotMonitor.Functions.Services.Vulnerability
             return results;
         }
 
+        /// <summary>
+        /// Side-row + counter write path (Plan §17). Reads the previous contributions row,
+        /// computes the (added, removed) delta against the new inventory, then applies
+        /// <c>+1</c> per added (with full metadata for first-insert) and <c>-1</c> per removed
+        /// (clamp ≥ 0).
+        /// <para>
+        /// Per Plan §17.3 "Side-row write before counter writes" + §17.8: individual counter
+        /// failures must NOT abort the rest of the loop, because the side-row already promises
+        /// the new key set — if we stopped on first failure, the next correlation run would see
+        /// an empty delta and never retry the under-counted keys. Each counter call is therefore
+        /// guarded individually; failures emit an <c>inventory_counter_write_failed</c> audit
+        /// entry (when the maintenance repo is wired) and continue processing. The wrapping
+        /// catch handles only the side-row CAS exhaustion + the parser path.
+        /// </para>
+        /// </summary>
+        internal async Task ApplyInventoryContributionsAsync(
+            string tenantId, string sessionId,
+            List<Dictionary<string, object>> inventoryItems,
+            Dictionary<string, string?>? cpeMappings)
+        {
+            InventoryContributionsDelta delta;
+            try
+            {
+                var typedItems = ParseSoftwareInventoryItems(inventoryItems, cpeMappings);
+                if (typedItems.Count == 0) return;
+                delta = await _vulnRepo.UpsertSessionInventoryContributionsAsync(tenantId, sessionId, typedItems);
+            }
+            catch (Exception ex)
+            {
+                _logger.LogWarning(ex,
+                    "Failed to upsert software inventory contributions side-row (non-fatal). tenant={TenantId} session={SessionId}",
+                    tenantId, sessionId);
+                return;
+            }
+
+            foreach (var added in delta.AddedItems)
+            {
+                try
+                {
+                    await _vulnRepo.IncrementSoftwareInventoryEntryAsync(tenantId, added, sessionId);
+                }
+                catch (Exception ex)
+                {
+                    _logger.LogWarning(ex,
+                        "SoftwareInventory increment failed (non-fatal — counter drift accepted). tenant={TenantId} session={SessionId} vendor={Vendor} name={Name} version={Version}",
+                        tenantId, sessionId, added.NormalizedVendor, added.NormalizedName, added.NormalizedVersion);
+                    await TryAuditInventoryCounterFailureAsync(
+                        tenantId, sessionId, "increment",
+                        added.NormalizedVendor, added.NormalizedName, added.NormalizedVersion, ex);
+                }
+            }
+
+            foreach (var removed in delta.RemovedKeys)
+            {
+                try
+                {
+                    await _vulnRepo.DecrementSoftwareInventoryEntryAsync(
+                        tenantId, removed.Vendor, removed.Name, removed.Version);
+                }
+                catch (Exception ex)
+                {
+                    _logger.LogWarning(ex,
+                        "SoftwareInventory decrement failed (non-fatal — counter drift accepted). tenant={TenantId} session={SessionId} vendor={Vendor} name={Name} version={Version}",
+                        tenantId, sessionId, removed.Vendor, removed.Name, removed.Version);
+                    await TryAuditInventoryCounterFailureAsync(
+                        tenantId, sessionId, "decrement",
+                        removed.Vendor, removed.Name, removed.Version, ex);
+                }
+            }
+        }
+
+        /// <summary>
+        /// Emits the <c>inventory_counter_write_failed</c> audit entry on a single counter
+        /// failure (Plan §17.8). Non-fatal — audit emission itself must not break the loop.
+        /// </summary>
+        private async Task TryAuditInventoryCounterFailureAsync(
+            string tenantId, string sessionId, string operation,
+            string vendor, string name, string version, Exception failureCause)
+        {
+            if (_maintenanceRepo == null) return;
+            try
+            {
+                await _maintenanceRepo.LogAuditEntryAsync(
+                    tenantId,
+                    action: "inventory_counter_write_failed",
+                    entityType: "Session",
+                    entityId: sessionId,
+                    performedBy: "system",
+                    details: new Dictionary<string, string>(StringComparer.Ordinal)
+                    {
+                        ["operation"] = operation,
+                        ["vendor"] = vendor,
+                        ["name"] = name,
+                        ["version"] = version,
+                        ["exceptionType"] = failureCause.GetType().Name,
+                        ["exceptionMessage"] = failureCause.Message,
+                    });
+            }
+            catch (Exception ex)
+            {
+                _logger.LogDebug(ex,
+                    "Failed to emit inventory_counter_write_failed audit entry (suppressed). tenant={TenantId} session={SessionId}",
+                    tenantId, sessionId);
+            }
+        }
+
+        /// <summary>
+        /// Materializes typed <see cref="SoftwareInventoryItem"/> entries from the loose
+        /// <c>inventoryItems</c> dictionary shape carried over the wire. Skips entries with no
+        /// usable normalization triple (matches the existing dedup behaviour in
+        /// <c>UpsertSoftwareInventoryAsync</c>). Attaches the CPE URI from <paramref name="cpeMappings"/>
+        /// when one is available for the normalized name — the increment helper then stamps it
+        /// on the SoftwareInventory row at first-insert.
+        /// </summary>
+        internal static List<SoftwareInventoryItem> ParseSoftwareInventoryItems(
+            List<Dictionary<string, object>> inventoryItems,
+            Dictionary<string, string?>? cpeMappings)
+        {
+            var items = new List<SoftwareInventoryItem>(inventoryItems.Count);
+            foreach (var raw in inventoryItems)
+            {
+                var normalizedName = raw.ContainsKey("normalizedName") ? raw["normalizedName"]?.ToString() ?? "" : "";
+                var normalizedVendor = raw.ContainsKey("normalizedPublisher") ? raw["normalizedPublisher"]?.ToString() ?? "" : "";
+                var normalizedVersion = raw.ContainsKey("normalizedVersion") ? raw["normalizedVersion"]?.ToString() ?? "" : "";
+
+                if (string.IsNullOrEmpty(normalizedVendor)
+                    && string.IsNullOrEmpty(normalizedName)
+                    && string.IsNullOrEmpty(normalizedVersion))
+                {
+                    continue;
+                }
+
+                string? cpeUri = null;
+                if (cpeMappings != null
+                    && !string.IsNullOrEmpty(normalizedName)
+                    && cpeMappings.TryGetValue(normalizedName, out var mapped)
+                    && !string.IsNullOrEmpty(mapped))
+                {
+                    cpeUri = mapped;
+                }
+
+                items.Add(new SoftwareInventoryItem
+                {
+                    NormalizedVendor = normalizedVendor,
+                    NormalizedName = normalizedName,
+                    NormalizedVersion = normalizedVersion,
+                    DisplayName = raw.ContainsKey("displayName") ? raw["displayName"]?.ToString() ?? "" : "",
+                    Publisher = raw.ContainsKey("publisher") ? raw["publisher"]?.ToString() ?? "" : "",
+                    RegistrySource = raw.ContainsKey("registrySource") ? raw["registrySource"]?.ToString() ?? "" : "",
+                    NormalizationConfidence = raw.ContainsKey("normalizationConfidence") ? raw["normalizationConfidence"]?.ToString() ?? "" : "",
+                    CpeUri = cpeUri,
+                });
+            }
+            return items;
+        }
+
         /// <summary>
         /// Replace characters not allowed in Azure Table Storage keys (/, \, #, ?).
         /// </summary>

src/Shared/AutopilotMonitor.Shared/DataAccess/IVulnerabilityRepository.cs

@@ -1,5 +1,7 @@
 using System.Collections.Generic;
+using System.Threading;
 using System.Threading.Tasks;
+using AutopilotMonitor.Shared.Models.Deletion;
 
 namespace AutopilotMonitor.Shared.DataAccess
 {
@@ -27,6 +29,48 @@ namespace AutopilotMonitor.Shared.DataAccess
         Task<List<Dictionary<string, object>>> GetAllSoftwareInventoryAsync();
         Task UpdateSoftwareInventoryCpeAsync(string normalizedName, string cpeUri);
 
+        // --- Software Inventory v2: contributions side-row + delta-driven counters (Plan §17) ---
+
+        /// <summary>
+        /// Reads the <c>SessionInventoryContributions</c> side-row for a session. Returns the
+        /// previously-counted normalized keys, or <c>null</c> when no side-row exists yet
+        /// (pre-side-row session OR first inventory correlation for this session).
+        /// </summary>
+        Task<IReadOnlyList<DeletionDecrementKey>?> GetSessionInventoryContributionsAsync(
+            string tenantId, string sessionId, CancellationToken ct = default);
+
+        /// <summary>
+        /// At-most-once-per-session, delta-update upsert of the side-row. Plan §17.3 ETag-CAS
+        /// loop: reads side-row, computes (added, removed), writes the new key set back with
+        /// ETag-Replace (or AddEntity on first call). Returns the delta the caller must apply
+        /// to the aggregate counter. Fail-loud on bounded-retry exhaustion (10 attempts or 60s
+        /// wall-clock per §12-Q10).
+        /// </summary>
+        Task<InventoryContributionsDelta> UpsertSessionInventoryContributionsAsync(
+            string tenantId, string sessionId,
+            IReadOnlyList<SoftwareInventoryItem> newItems,
+            CancellationToken ct = default);
+
+        /// <summary>Deletes the side-row (cascade tombstone step). 404 = treated as already gone.</summary>
+        Task DeleteSessionInventoryContributionsAsync(string tenantId, string sessionId, CancellationToken ct = default);
+
+        /// <summary>
+        /// ETag-CAS <c>SessionCount += 1</c> on the <c>SoftwareInventory</c> row for the given
+        /// normalized key. Creates the row with full metadata on first encounter. Bounded retry
+        /// (10 attempts / 60s wall-clock); throws on exhaustion.
+        /// </summary>
+        Task IncrementSoftwareInventoryEntryAsync(
+            string tenantId, SoftwareInventoryItem item, string sessionId, CancellationToken ct = default);
+
+        /// <summary>
+        /// ETag-CAS <c>SessionCount -= 1</c> with clamp-≥-0 on the matching <c>SoftwareInventory</c>
+        /// row. Deletes the row if the count hits zero. 404 = no-op (idempotent). Bounded retry
+        /// (10 attempts / 60s); throws on exhaustion.
+        /// </summary>
+        Task DecrementSoftwareInventoryEntryAsync(
+            string tenantId, string normalizedVendor, string normalizedName, string normalizedVersion,
+            CancellationToken ct = default);
+
         // --- CPE Mappings ---
         Task<int> ImportCpeMappingSeedFromJsonAsync(string json);
         Task<int> ImportCpeCommunityMappingsFromJsonAsync(string json);

src/Shared/AutopilotMonitor.Shared/Models/Deletion/SoftwareInventoryContributions.cs

@@ -0,0 +1,188 @@
+using System;
+using System.Collections.Generic;
+using System.IO;
+using System.IO.Compression;
+using System.Linq;
+using System.Text;
+using System.Text.Json;
+
+namespace AutopilotMonitor.Shared.Models.Deletion
+{
+    /// <summary>
+    /// Single software entry as it arrives from a <c>software_inventory_analysis</c> event.
+    /// PR2: the unit that <see cref="DeletionDecrementKey"/> identifies for cascade-time
+    /// decrement, plus the metadata needed to (re-)create a fresh <c>SoftwareInventory</c> row
+    /// when this is the first session that's seen the software.
+    /// </summary>
+    public class SoftwareInventoryItem
+    {
+        // Normalized fields used as the identity key. The triple (Vendor, Name, Version) is
+        // the side-row key and the SoftwareInventory RowKey component.
+        public string NormalizedVendor { get; set; } = string.Empty;
+        public string NormalizedName { get; set; } = string.Empty;
+        public string NormalizedVersion { get; set; } = string.Empty;
+
+        // Display metadata — only written on first-insert into SoftwareInventory; later
+        // correlations only bump SessionCount + LastSeenAt + LastSessionId.
+        public string DisplayName { get; set; } = string.Empty;
+        public string Publisher { get; set; } = string.Empty;
+        public string RegistrySource { get; set; } = string.Empty;
+        public string NormalizationConfidence { get; set; } = string.Empty;
+
+        /// <summary>
+        /// Optional CPE URI for this software (resolved by CpeMappingService at correlation
+        /// time). Carried so the increment helper can stamp it on the row at first-insert.
+        /// </summary>
+        public string? CpeUri { get; set; }
+
+        /// <summary>The canonical key used in the contributions side-row JSON and as the SoftwareInventory RowKey component.</summary>
+        public DeletionDecrementKey ToDecrementKey() => new DeletionDecrementKey
+        {
+            Vendor = NormalizedVendor,
+            Name = NormalizedName,
+            Version = NormalizedVersion,
+        };
+    }
+
+    /// <summary>
+    /// Result of a <c>UpsertSessionInventoryContributionsAsync</c> call: the items added vs the
+    /// keys removed compared to the previous side-row state. Caller applies <c>+1</c> per added
+    /// (with full metadata for first-insert) and <c>-1</c> per removed (key only, decrement
+    /// clamps at zero).
+    /// </summary>
+    public class InventoryContributionsDelta
+    {
+        public List<SoftwareInventoryItem> AddedItems { get; set; } = new List<SoftwareInventoryItem>();
+        public List<DeletionDecrementKey> RemovedKeys { get; set; } = new List<DeletionDecrementKey>();
+
+        /// <summary>True when this was the first time the side-row was created for the session.</summary>
+        public bool FirstTime { get; set; }
+    }
+
+    /// <summary>
+    /// JSON + gzip+Base64 codec for the <c>SoftwareKeysJson</c> column on the
+    /// <c>SessionInventoryContributions</c> side-row. Pure functions — easy to unit-test
+    /// independent of any Azure SDK plumbing.
+    /// <para>
+    /// Plan §17.7: raw JSON for small key sets, gzip+Base64 once the raw payload exceeds
+    /// <see cref="CompressionThresholdBytes"/>. <c>IsCompressed</c> on the row chooses the
+    /// decode path. <see cref="DeletionManifestBuilder"/> uses this same codec for the
+    /// cascade-time decrement-key extraction (PR1 already wired).
+    /// </para>
+    /// </summary>
+    public static class SoftwareKeysJsonCodec
+    {
+        public const int CompressionThresholdBytes = 30_000;
+
+        public static EncodedSoftwareKeys Encode(IEnumerable<DeletionDecrementKey> keys)
+        {
+            var ordered = keys
+                .OrderBy(k => k.Vendor, StringComparer.Ordinal)
+                .ThenBy(k => k.Name, StringComparer.Ordinal)
+                .ThenBy(k => k.Version, StringComparer.Ordinal)
+                .ToList();
+
+            var raw = JsonSerializer.Serialize(ordered, DeletionManifestJson.SerializerOptions);
+            if (raw.Length <= CompressionThresholdBytes)
+            {
+                return new EncodedSoftwareKeys { Encoded = raw, IsCompressed = false, KeyCount = ordered.Count };
+            }
+
+            using var output = new MemoryStream();
+            using (var gzip = new GZipStream(output, CompressionLevel.Optimal, leaveOpen: true))
+            {
+                var bytes = Encoding.UTF8.GetBytes(raw);
+                gzip.Write(bytes, 0, bytes.Length);
+            }
+            var compressed = Convert.ToBase64String(output.ToArray());
+            return new EncodedSoftwareKeys { Encoded = compressed, IsCompressed = true, KeyCount = ordered.Count };
+        }
+
+        public static List<DeletionDecrementKey> Decode(string encoded, bool isCompressed)
+        {
+            if (string.IsNullOrEmpty(encoded)) return new List<DeletionDecrementKey>();
+
+            string json;
+            if (isCompressed)
+            {
+                byte[] gz;
+                try
+                {
+                    gz = Convert.FromBase64String(encoded);
+                }
+                catch (FormatException ex)
+                {
+                    throw new InvalidDataException(
+                        "SoftwareKeysJson is marked IsCompressed=true but the payload is not valid Base64.", ex);
+                }
+                using var input = new MemoryStream(gz);
+                using var gunzip = new GZipStream(input, CompressionMode.Decompress);
+                using var output = new MemoryStream();
+                gunzip.CopyTo(output);
+                json = Encoding.UTF8.GetString(output.ToArray());
+            }
+            else
+            {
+                json = encoded;
+            }
+
+            try
+            {
+                return JsonSerializer.Deserialize<List<DeletionDecrementKey>>(json, DeletionManifestJson.SerializerOptions)
+                       ?? new List<DeletionDecrementKey>();
+            }
+            catch (JsonException ex)
+            {
+                throw new InvalidDataException(
+                    "SoftwareKeysJson is not valid JSON for List<DeletionDecrementKey>.", ex);
+            }
+        }
+
+        /// <summary>
+        /// Compares <paramref name="newKeys"/> against <paramref name="oldKeys"/> and returns the
+        /// (added, removed) delta. Comparison is case-insensitive on every component because
+        /// upstream normalization is best-effort across registry sources.
+        /// </summary>
+        public static (HashSet<string> Added, HashSet<string> Removed) ComputeDelta(
+            IEnumerable<DeletionDecrementKey> oldKeys,
+            IEnumerable<DeletionDecrementKey> newKeys)
+        {
+            var oldSet = ToCompositeSet(oldKeys);
+            var newSet = ToCompositeSet(newKeys);
+            var added = new HashSet<string>(newSet, StringComparer.OrdinalIgnoreCase);
+            added.ExceptWith(oldSet);
+            var removed = new HashSet<string>(oldSet, StringComparer.OrdinalIgnoreCase);
+            removed.ExceptWith(newSet);
+            return (added, removed);
+        }
+
+        /// <summary>
+        /// Composite-key string used as a lookup identity. Three-part null-safe join with a
+        /// separator character that cannot legitimately appear in any normalized component
+        /// (we use <c>'
'</c>, START OF HEADING — never present in publisher / name / version).
+        /// </summary>
+        public static string CompositeKey(string vendor, string name, string version)
+            => $"{vendor}
{name}
{version}";
+
+        public static string CompositeKey(DeletionDecrementKey key)
+            => CompositeKey(key.Vendor ?? string.Empty, key.Name ?? string.Empty, key.Version ?? string.Empty);
+
+        private static HashSet<string> ToCompositeSet(IEnumerable<DeletionDecrementKey> keys)
+        {
+            var set = new HashSet<string>(StringComparer.OrdinalIgnoreCase);
+            foreach (var k in keys)
+            {
+                set.Add(CompositeKey(k));
+            }
+            return set;
+        }
+    }
+
+    /// <summary>Encoded form of a software-keys JSON payload, tagged with the codec flag.</summary>
+    public class EncodedSoftwareKeys
+    {
+        public string Encoded { get; set; } = string.Empty;
+        public bool IsCompressed { get; set; }
+        public int KeyCount { get; set; }
+    }
+}