diff --git a/src/Backend/AutopilotMonitor.Functions.Tests/TableBatchDedupTests.cs b/src/Backend/AutopilotMonitor.Functions.Tests/TableBatchDedupTests.cs
new file mode 100644
index 00000000..11d065f7
--- /dev/null
+++ b/src/Backend/AutopilotMonitor.Functions.Tests/TableBatchDedupTests.cs
@@ -0,0 +1,101 @@
+using System;
+using System.Linq;
+using AutopilotMonitor.Functions.DataAccess.TableStorage;
+using AutopilotMonitor.Shared.Models;
+using Azure.Data.Tables;
+
+namespace AutopilotMonitor.Functions.Tests;
+
+///
+/// Covers — the guard that stops an agent replaying an
+/// overlapping ordinal/step from failing an entire ingest entity-group-transaction with Azure
+/// Tables' HTTP 400 / InvalidDuplicateRow. Exercised through the three real ingest
+/// projections (Signals, DecisionTransitions, Index) so the (PK, RowKey) contract is asserted
+/// against the keys those repos actually emit.
+///
+public class TableBatchDedupTests
+{
+ private const string TenantId = "a1b2c3d4-e5f6-7890-abcd-ef1234567890";
+ private const string SessionId = "b2c3d4e5-f6a7-8901-bcde-f12345678901";
+
+ private static SignalRecord Signal(long ordinal, string kind) => new()
+ {
+ TenantId = TenantId,
+ SessionId = SessionId,
+ SessionSignalOrdinal = ordinal,
+ SessionTraceOrdinal = ordinal,
+ Kind = kind,
+ KindSchemaVersion = 1,
+ OccurredAtUtc = new DateTime(2026, 4, 21, 10, 0, 0, DateTimeKind.Utc),
+ SourceOrigin = "test",
+ PayloadJson = "{}",
+ };
+
+ [Fact]
+ public void ByRowKey_collapses_duplicate_signal_ordinal_keeping_last_write()
+ {
+ var first = TableSignalRepository.ToEntity(Signal(6, "first"));
+ var second = TableSignalRepository.ToEntity(Signal(6, "second")); // same RowKey as first
+ var other = TableSignalRepository.ToEntity(Signal(7, "other"));
+
+ var (deduped, dropped) = TableBatchDedup.ByRowKey(new[] { first, second, other });
+
+ Assert.Equal(1, dropped);
+ Assert.Equal(2, deduped.Count);
+
+ var collapsed = deduped.Single(e => e.RowKey == TableSignalRepository.BuildRowKey(6));
+ Assert.Equal("second", collapsed.GetString("Kind")); // last-wins
+ }
+
+ [Fact]
+ public void ByRowKey_collapses_duplicate_transition_stepIndex()
+ {
+ DecisionTransitionRecord Transition(int step, string toStage) => new()
+ {
+ TenantId = TenantId,
+ SessionId = SessionId,
+ StepIndex = step,
+ SessionTraceOrdinal = step,
+ SignalOrdinalRef = step,
+ OccurredAtUtc = new DateTime(2026, 4, 21, 10, 0, 0, DateTimeKind.Utc),
+ Trigger = "t",
+ FromStage = "A",
+ ToStage = toStage,
+ ReducerVersion = "1",
+ PayloadJson = "{}",
+ };
+
+ var first = TableDecisionTransitionRepository.ToEntity(Transition(3, "Stage1"));
+ var second = TableDecisionTransitionRepository.ToEntity(Transition(3, "Stage2"));
+
+ var (deduped, dropped) = TableBatchDedup.ByRowKey(new[] { first, second });
+
+ Assert.Equal(1, dropped);
+ var collapsed = Assert.Single(deduped);
+ Assert.Equal("Stage2", collapsed.GetString("ToStage")); // last-wins
+ }
+
+ [Fact]
+ public void ByRowKey_is_a_passthrough_when_no_duplicates_and_orders_by_rowkey()
+ {
+ var ten = TableSignalRepository.ToEntity(Signal(10, "ten"));
+ var nine = TableSignalRepository.ToEntity(Signal(9, "nine"));
+
+ var (deduped, dropped) = TableBatchDedup.ByRowKey(new[] { ten, nine });
+
+ Assert.Equal(0, dropped);
+ Assert.Equal(2, deduped.Count);
+ // D19 padding → RowKey lex order matches numeric order: 9 before 10.
+ Assert.Equal(TableSignalRepository.BuildRowKey(9), deduped[0].RowKey);
+ Assert.Equal(TableSignalRepository.BuildRowKey(10), deduped[1].RowKey);
+ }
+
+ [Fact]
+ public void ByRowKey_handles_empty_input()
+ {
+ var (deduped, dropped) = TableBatchDedup.ByRowKey(Array.Empty());
+
+ Assert.Empty(deduped);
+ Assert.Equal(0, dropped);
+ }
+}
diff --git a/src/Backend/AutopilotMonitor.Functions/DataAccess/TableStorage/TableBatchDedup.cs b/src/Backend/AutopilotMonitor.Functions/DataAccess/TableStorage/TableBatchDedup.cs
new file mode 100644
index 00000000..1f1bc38d
--- /dev/null
+++ b/src/Backend/AutopilotMonitor.Functions/DataAccess/TableStorage/TableBatchDedup.cs
@@ -0,0 +1,43 @@
+using System;
+using System.Collections.Generic;
+using System.Linq;
+using Azure.Data.Tables;
+
+namespace AutopilotMonitor.Functions.DataAccess.TableStorage
+{
+ ///
+ /// Collapses entities sharing a (PartitionKey, RowKey) within a single batch down to the
+ /// last occurrence (idempotent →
+ /// last-write-wins). Azure Table Storage rejects the entire entity-group-transaction
+ /// with HTTP 400 / InvalidDuplicateRow if any two actions target the same RowKey, so a
+ /// duplicate-keyed record (e.g. an agent replaying an overlapping signal ordinal) would
+ /// otherwise fail the whole ingest batch. Callers must group by PartitionKey before submitting;
+ /// the (PK, RK) key here keeps the helper correct even if that precondition is ever relaxed.
+ ///
+ internal static class TableBatchDedup
+ {
+ ///
+ /// Returns the de-duplicated entities (unique by (PartitionKey, RowKey), last-wins) plus the
+ /// number of rows dropped. Output is PK/RowKey-ordered for deterministic chunk boundaries.
+ ///
+ internal static (List Deduped, int Dropped) ByRowKey(IEnumerable entities)
+ {
+ var byKey = new Dictionary<(string Pk, string Rk), TableEntity>();
+ var dropped = 0;
+
+ foreach (var e in entities)
+ {
+ var key = (e.PartitionKey, e.RowKey);
+ if (byKey.ContainsKey(key)) dropped++;
+ byKey[key] = e; // last-wins
+ }
+
+ var deduped = byKey.Values
+ .OrderBy(e => e.PartitionKey, StringComparer.Ordinal)
+ .ThenBy(e => e.RowKey, StringComparer.Ordinal)
+ .ToList();
+
+ return (deduped, dropped);
+ }
+ }
+}
diff --git a/src/Backend/AutopilotMonitor.Functions/DataAccess/TableStorage/TableDecisionTransitionRepository.cs b/src/Backend/AutopilotMonitor.Functions/DataAccess/TableStorage/TableDecisionTransitionRepository.cs
index 44c091c0..a7501f12 100644
--- a/src/Backend/AutopilotMonitor.Functions/DataAccess/TableStorage/TableDecisionTransitionRepository.cs
+++ b/src/Backend/AutopilotMonitor.Functions/DataAccess/TableStorage/TableDecisionTransitionRepository.cs
@@ -51,17 +51,23 @@ namespace AutopilotMonitor.Functions.DataAccess.TableStorage
foreach (var group in records.GroupBy(r => (r.TenantId, r.SessionId)))
{
- var ordered = group.OrderBy(r => r.StepIndex).ToList();
+ // Collapse duplicate RowKeys (D10(StepIndex)) before building the transaction —
+ // Azure Tables rejects the whole batch with InvalidDuplicateRow otherwise. Dedup
+ // also yields RowKey ordering, so the previous OrderBy(StepIndex) is redundant.
+ var (deduped, dropped) = TableBatchDedup.ByRowKey(group.Select(ToEntity));
+ if (dropped > 0)
+ _logger.LogWarning(
+ "DecisionTransitions: dropped {Dropped} duplicate-RowKey row(s) (last-wins) for {Tenant}_{Session} — agent likely replayed overlapping StepIndex",
+ dropped, group.Key.TenantId, group.Key.SessionId);
- for (var offset = 0; offset < ordered.Count; offset += TransactionChunkSize)
+ for (var offset = 0; offset < deduped.Count; offset += TransactionChunkSize)
{
cancellationToken.ThrowIfCancellationRequested();
- var chunk = ordered.Skip(offset).Take(TransactionChunkSize).ToList();
+ var chunk = deduped.Skip(offset).Take(TransactionChunkSize).ToList();
var actions = chunk
- .Select(r => new TableTransactionAction(
- TableTransactionActionType.UpsertReplace,
- ToEntity(r)))
+ .Select(e => new TableTransactionAction(
+ TableTransactionActionType.UpsertReplace, e))
.ToList();
await table.SubmitTransactionAsync(actions, cancellationToken).ConfigureAwait(false);
@@ -70,7 +76,7 @@ namespace AutopilotMonitor.Functions.DataAccess.TableStorage
_logger.LogDebug(
"DecisionTransitions: committed {Count} rows for {Tenant}_{Session}",
- ordered.Count, group.Key.TenantId, group.Key.SessionId);
+ deduped.Count, group.Key.TenantId, group.Key.SessionId);
}
return committed;
diff --git a/src/Backend/AutopilotMonitor.Functions/DataAccess/TableStorage/TableIndexRepository.cs b/src/Backend/AutopilotMonitor.Functions/DataAccess/TableStorage/TableIndexRepository.cs
index 44474624..bab7303c 100644
--- a/src/Backend/AutopilotMonitor.Functions/DataAccess/TableStorage/TableIndexRepository.cs
+++ b/src/Backend/AutopilotMonitor.Functions/DataAccess/TableStorage/TableIndexRepository.cs
@@ -84,13 +84,20 @@ namespace AutopilotMonitor.Functions.DataAccess.TableStorage
foreach (var group in entities.GroupBy(e => e.PartitionKey))
{
- var ordered = group.OrderBy(e => e.RowKey, StringComparer.Ordinal).ToList();
+ // Collapse duplicate RowKeys before building the transaction — Azure Tables
+ // rejects the whole batch with InvalidDuplicateRow otherwise. Dedup also yields
+ // RowKey ordering, replacing the previous explicit OrderBy.
+ var (deduped, dropped) = TableBatchDedup.ByRowKey(group);
+ if (dropped > 0)
+ _logger.LogWarning(
+ "{Table}: dropped {Dropped} duplicate-RowKey index row(s) (last-wins) for partition {Pk}",
+ tableName, dropped, group.Key);
- for (var offset = 0; offset < ordered.Count; offset += TransactionChunkSize)
+ for (var offset = 0; offset < deduped.Count; offset += TransactionChunkSize)
{
cancellationToken.ThrowIfCancellationRequested();
- var chunk = ordered.Skip(offset).Take(TransactionChunkSize).ToList();
+ var chunk = deduped.Skip(offset).Take(TransactionChunkSize).ToList();
var actions = chunk
.Select(e => new TableTransactionAction(TableTransactionActionType.UpsertReplace, e))
.ToList();
diff --git a/src/Backend/AutopilotMonitor.Functions/DataAccess/TableStorage/TableSignalRepository.cs b/src/Backend/AutopilotMonitor.Functions/DataAccess/TableStorage/TableSignalRepository.cs
index c35ffc60..b59f1001 100644
--- a/src/Backend/AutopilotMonitor.Functions/DataAccess/TableStorage/TableSignalRepository.cs
+++ b/src/Backend/AutopilotMonitor.Functions/DataAccess/TableStorage/TableSignalRepository.cs
@@ -49,17 +49,24 @@ namespace AutopilotMonitor.Functions.DataAccess.TableStorage
foreach (var group in records.GroupBy(r => (r.TenantId, r.SessionId)))
{
- var ordered = group.OrderBy(r => r.SessionSignalOrdinal).ToList();
+ // Collapse duplicate RowKeys (D19(SessionSignalOrdinal)) before building the
+ // transaction — Azure Tables rejects the whole batch with InvalidDuplicateRow
+ // otherwise (e.g. an agent replaying an overlapping ordinal). Dedup also yields
+ // RowKey ordering, so the previous OrderBy(ordinal) is no longer needed.
+ var (deduped, dropped) = TableBatchDedup.ByRowKey(group.Select(ToEntity));
+ if (dropped > 0)
+ _logger.LogWarning(
+ "Signals: dropped {Dropped} duplicate-RowKey row(s) (last-wins) for {Tenant}_{Session} — agent likely replayed overlapping SessionSignalOrdinal",
+ dropped, group.Key.TenantId, group.Key.SessionId);
- for (var offset = 0; offset < ordered.Count; offset += TransactionChunkSize)
+ for (var offset = 0; offset < deduped.Count; offset += TransactionChunkSize)
{
cancellationToken.ThrowIfCancellationRequested();
- var chunk = ordered.Skip(offset).Take(TransactionChunkSize).ToList();
+ var chunk = deduped.Skip(offset).Take(TransactionChunkSize).ToList();
var actions = chunk
- .Select(r => new TableTransactionAction(
- TableTransactionActionType.UpsertReplace,
- ToEntity(r)))
+ .Select(e => new TableTransactionAction(
+ TableTransactionActionType.UpsertReplace, e))
.ToList();
await table.SubmitTransactionAsync(actions, cancellationToken).ConfigureAwait(false);
@@ -68,7 +75,7 @@ namespace AutopilotMonitor.Functions.DataAccess.TableStorage
_logger.LogDebug(
"Signals: committed {Count} rows for {Tenant}_{Session}",
- ordered.Count, group.Key.TenantId, group.Key.SessionId);
+ deduped.Count, group.Key.TenantId, group.Key.SessionId);
}
return committed;