feat(backend): M5.a primary tables — Signals + DecisionTransitions repositories

Groundwork for the V2 ingest path (/api/agent/telemetry). Adds the two primary Decision-Engine tables projected from the agent-side SignalLog and Journal so the Inspector + Reducer-Verifier can replay sessions. - Constants.TableNames: Signals, DecisionTransitions (+ All[] init). - Shared DTOs SignalRecord / DecisionTransitionRecord carry a flat projection plus the agent-serialised PayloadJson for full fidelity. DecisionTransitionRecord eagerly carries index discriminators (IsTerminal, ClassifierVerdictId, ClassifierHypothesisLevel) so the later index fan-out can read them straight off the primary row. - ISignalRepository / IDecisionTransitionRepository: lean interfaces (StoreBatchAsync only — queries land with the read endpoints in M5.c). - TableSignalRepository / TableDecisionTransitionRepository: idempotent UpsertReplace via entity-group-transactions, grouped per (tenant, session), chunked at the 100-op Azure limit. PayloadJson goes through TableStorageChunking so the 32K-char property ceiling can't bite on large Evidence + Payload blobs. - Registered in DataAccessServiceExtensions alongside the existing 13 Table repositories — Cosmos swap remains a single-file change. - 11 pure mapping tests cover key builders (PK tenant_session, RK D19(ordinal) / D10(stepIndex), lex-ordering invariant), full entity projection, chunked PayloadJson handling, dead-end paths, and null-payload tolerance. Index tables (SessionsByTerminal, SessionsByStage, DeadEndsByReason, ClassifierVerdictsByIdLevel, SignalsByKind) land with the queue-driven fan-out in M5.b; the primary schemas are stable. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
2026-10-24 20:53:06 +00:00 · 2026-04-21 09:19:08 +02:00
parent 941e03cf32
commit eeaa0643a2
10 changed files with 639 additions and 1 deletions
@@ -0,0 +1,127 @@
+using AutopilotMonitor.Functions.DataAccess.TableStorage;
+using AutopilotMonitor.Shared.Models;
+
+namespace AutopilotMonitor.Functions.Tests;
+
+/// <summary>
+/// Pure mapping tests for <see cref="TableDecisionTransitionRepository"/>. Sibling of
+/// <see cref="TableSignalRepositoryMappingTests"/> — same rationale.
+/// </summary>
+public class TableDecisionTransitionRepositoryMappingTests
+{
+    private const string TenantId  = "a1b2c3d4-e5f6-7890-abcd-ef1234567890";
+    private const string SessionId = "b2c3d4e5-f6a7-8901-bcde-f12345678901";
+
+    [Fact]
+    public void BuildPartitionKey_is_tenant_underscore_session()
+    {
+        var pk = TableDecisionTransitionRepository.BuildPartitionKey(TenantId, SessionId);
+        Assert.Equal($"{TenantId}_{SessionId}", pk);
+    }
+
+    [Fact]
+    public void BuildRowKey_pads_stepIndex_to_10_digits()
+    {
+        Assert.Equal("0000000000", TableDecisionTransitionRepository.BuildRowKey(0));
+        Assert.Equal("0000000042", TableDecisionTransitionRepository.BuildRowKey(42));
+        Assert.Equal("2147483647", TableDecisionTransitionRepository.BuildRowKey(int.MaxValue));
+    }
+
+    [Fact]
+    public void ToEntity_projects_all_typed_columns()
+    {
+        var record = new DecisionTransitionRecord
+        {
+            TenantId                    = TenantId,
+            SessionId                   = SessionId,
+            StepIndex                   = 7,
+            SessionTraceOrdinal         = 117,
+            SignalOrdinalRef            = 42,
+            OccurredAtUtc               = new DateTime(2026, 4, 21, 10, 0, 0, DateTimeKind.Utc),
+            Trigger                     = "EspExiting",
+            FromStage                   = "EspInProgress",
+            ToStage                     = "WhiteGloveSealed",
+            Taken                       = true,
+            DeadEndReason               = null,
+            ReducerVersion              = "1.0.0",
+            IsTerminal                  = true,
+            ClassifierVerdictId         = "whiteglove-sealing",
+            ClassifierHypothesisLevel   = "Strong",
+            PayloadJson                 = "{\"foo\":\"bar\"}",
+        };
+
+        var entity = TableDecisionTransitionRepository.ToEntity(record);
+
+        Assert.Equal($"{TenantId}_{SessionId}", entity.PartitionKey);
+        Assert.Equal("0000000007", entity.RowKey);
+        Assert.Equal(TenantId, entity.GetString("TenantId"));
+        Assert.Equal(SessionId, entity.GetString("SessionId"));
+        Assert.Equal(7, entity.GetInt32("StepIndex"));
+        Assert.Equal(117L, entity.GetInt64("SessionTraceOrdinal"));
+        Assert.Equal(42L, entity.GetInt64("SignalOrdinalRef"));
+        Assert.Equal(record.OccurredAtUtc, entity.GetDateTime("OccurredAtUtc"));
+        Assert.Equal("EspExiting", entity.GetString("Trigger"));
+        Assert.Equal("EspInProgress", entity.GetString("FromStage"));
+        Assert.Equal("WhiteGloveSealed", entity.GetString("ToStage"));
+        Assert.Equal(true, entity.GetBoolean("Taken"));
+        Assert.Null(entity.GetString("DeadEndReason"));
+        Assert.Equal("1.0.0", entity.GetString("ReducerVersion"));
+        Assert.Equal(true, entity.GetBoolean("IsTerminal"));
+        Assert.Equal("whiteglove-sealing", entity.GetString("ClassifierVerdictId"));
+        Assert.Equal("Strong", entity.GetString("ClassifierHypothesisLevel"));
+        Assert.Equal("{\"foo\":\"bar\"}", entity.GetString("PayloadJson"));
+    }
+
+    [Fact]
+    public void ToEntity_preserves_dead_end_reason_for_blocked_transitions()
+    {
+        var record = new DecisionTransitionRecord
+        {
+            TenantId            = TenantId,
+            SessionId           = SessionId,
+            StepIndex           = 3,
+            SessionTraceOrdinal = 5,
+            SignalOrdinalRef    = 3,
+            OccurredAtUtc       = DateTime.UtcNow,
+            Trigger             = "EspExiting",
+            FromStage           = "EspInProgress",
+            ToStage             = "EspInProgress", // no transition — dead end
+            Taken               = false,
+            DeadEndReason       = "hybrid_reboot_gate_blocking",
+            ReducerVersion      = "1.0.0",
+            IsTerminal          = false,
+            PayloadJson         = "{}",
+        };
+
+        var entity = TableDecisionTransitionRepository.ToEntity(record);
+
+        Assert.Equal(false, entity.GetBoolean("Taken"));
+        Assert.Equal("hybrid_reboot_gate_blocking", entity.GetString("DeadEndReason"));
+    }
+
+    [Fact]
+    public void ToEntity_chunks_oversized_PayloadJson()
+    {
+        var oversized = new string('x', 75_000);
+        var record = new DecisionTransitionRecord
+        {
+            TenantId            = TenantId,
+            SessionId           = SessionId,
+            StepIndex           = 1,
+            SessionTraceOrdinal = 1,
+            SignalOrdinalRef    = 1,
+            OccurredAtUtc       = DateTime.UtcNow,
+            Trigger             = "t",
+            FromStage           = "F",
+            ToStage             = "T",
+            Taken               = true,
+            ReducerVersion      = "1.0.0",
+            PayloadJson         = oversized,
+        };
+
+        var entity = TableDecisionTransitionRepository.ToEntity(record);
+
+        Assert.Equal("3", entity.GetString("PayloadJson_ChunkCount"));
+        Assert.False(entity.ContainsKey("PayloadJson"));
+    }
+}
@@ -0,0 +1,117 @@
+using AutopilotMonitor.Functions.DataAccess.TableStorage;
+using AutopilotMonitor.Shared.Models;
+
+namespace AutopilotMonitor.Functions.Tests;
+
+/// <summary>
+/// Pure mapping tests for <see cref="TableSignalRepository"/>. The transaction pipeline itself
+/// needs a live Azurite harness (out of scope for unit tests); the deterministic bit — how the
+/// record projects onto the Azure Table entity — is covered here.
+/// </summary>
+public class TableSignalRepositoryMappingTests
+{
+    private const string TenantId  = "a1b2c3d4-e5f6-7890-abcd-ef1234567890";
+    private const string SessionId = "b2c3d4e5-f6a7-8901-bcde-f12345678901";
+
+    [Fact]
+    public void BuildPartitionKey_is_tenant_underscore_session()
+    {
+        var pk = TableSignalRepository.BuildPartitionKey(TenantId, SessionId);
+        Assert.Equal($"{TenantId}_{SessionId}", pk);
+    }
+
+    [Fact]
+    public void BuildRowKey_pads_ordinal_to_19_digits_for_lex_ordering()
+    {
+        Assert.Equal("0000000000000000000", TableSignalRepository.BuildRowKey(0));
+        Assert.Equal("0000000000000000042", TableSignalRepository.BuildRowKey(42));
+        Assert.Equal("9223372036854775807", TableSignalRepository.BuildRowKey(long.MaxValue));
+    }
+
+    [Fact]
+    public void BuildRowKey_orders_lexicographically_matching_numeric_order()
+    {
+        var a = TableSignalRepository.BuildRowKey(9);
+        var b = TableSignalRepository.BuildRowKey(10);
+        // 19-digit padding means string comparison matches numeric order (the property of the key).
+        Assert.True(string.CompareOrdinal(a, b) < 0);
+    }
+
+    [Fact]
+    public void ToEntity_projects_all_typed_columns()
+    {
+        var record = new SignalRecord
+        {
+            TenantId             = TenantId,
+            SessionId            = SessionId,
+            SessionSignalOrdinal = 17,
+            SessionTraceOrdinal  = 117,
+            Kind                 = "EspPhaseChanged",
+            KindSchemaVersion    = 2,
+            OccurredAtUtc        = new DateTime(2026, 4, 21, 10, 0, 0, DateTimeKind.Utc),
+            SourceOrigin         = "EspAndHelloTrackerAdapter",
+            PayloadJson          = "{\"foo\":\"bar\"}",
+        };
+
+        var entity = TableSignalRepository.ToEntity(record);
+
+        Assert.Equal($"{TenantId}_{SessionId}", entity.PartitionKey);
+        Assert.Equal("0000000000000000017", entity.RowKey);
+        Assert.Equal(TenantId, entity.GetString("TenantId"));
+        Assert.Equal(SessionId, entity.GetString("SessionId"));
+        Assert.Equal(17L, entity.GetInt64("SessionSignalOrdinal"));
+        Assert.Equal(117L, entity.GetInt64("SessionTraceOrdinal"));
+        Assert.Equal("EspPhaseChanged", entity.GetString("Kind"));
+        Assert.Equal(2, entity.GetInt32("KindSchemaVersion"));
+        Assert.Equal(record.OccurredAtUtc, entity.GetDateTime("OccurredAtUtc"));
+        Assert.Equal("EspAndHelloTrackerAdapter", entity.GetString("SourceOrigin"));
+        Assert.Equal("{\"foo\":\"bar\"}", entity.GetString("PayloadJson"));
+    }
+
+    [Fact]
+    public void ToEntity_chunks_oversized_PayloadJson_to_survive_the_32K_char_per_property_limit()
+    {
+        var oversized = new string('x', 75_000);
+        var record = new SignalRecord
+        {
+            TenantId             = TenantId,
+            SessionId            = SessionId,
+            SessionSignalOrdinal = 1,
+            SessionTraceOrdinal  = 1,
+            Kind                 = "EspPhaseChanged",
+            KindSchemaVersion    = 1,
+            OccurredAtUtc        = DateTime.UtcNow,
+            SourceOrigin         = "test",
+            PayloadJson          = oversized,
+        };
+
+        var entity = TableSignalRepository.ToEntity(record);
+
+        Assert.Equal("3", entity.GetString("PayloadJson_ChunkCount"));
+        Assert.Equal(30_000, entity.GetString("PayloadJson_0").Length);
+        Assert.Equal(30_000, entity.GetString("PayloadJson_1").Length);
+        Assert.Equal(15_000, entity.GetString("PayloadJson_2").Length);
+        Assert.False(entity.ContainsKey("PayloadJson"));
+    }
+
+    [Fact]
+    public void ToEntity_tolerates_null_PayloadJson_without_throwing()
+    {
+        var record = new SignalRecord
+        {
+            TenantId             = TenantId,
+            SessionId            = SessionId,
+            SessionSignalOrdinal = 1,
+            SessionTraceOrdinal  = 1,
+            Kind                 = "EspPhaseChanged",
+            KindSchemaVersion    = 1,
+            OccurredAtUtc        = DateTime.UtcNow,
+            SourceOrigin         = "test",
+            PayloadJson          = null!,
+        };
+
+        var entity = TableSignalRepository.ToEntity(record);
+
+        Assert.Equal(string.Empty, entity.GetString("PayloadJson"));
+    }
+}
@@ -36,6 +36,10 @@ namespace AutopilotMonitor.Functions.DataAccess
            services.AddSingleton<IUserUsageRepository, TableUserUsageRepository>();
            services.AddSingleton<IDistressReportRepository, TableDistressReportRepository>();
            services.AddSingleton<IOpsEventRepository, TableOpsEventRepository>();
+
+            // V2 Decision Engine repositories (Plan §M5).
+            services.AddSingleton<ISignalRepository, TableSignalRepository>();
+            services.AddSingleton<IDecisionTransitionRepository, TableDecisionTransitionRepository>();
            return services;
        }

@@ -0,0 +1,120 @@
+using System.Collections.Generic;
+using System.Linq;
+using System.Threading;
+using System.Threading.Tasks;
+using Azure.Data.Tables;
+using AutopilotMonitor.Functions.Security;
+using AutopilotMonitor.Functions.Services;
+using AutopilotMonitor.Shared;
+using AutopilotMonitor.Shared.DataAccess;
+using AutopilotMonitor.Shared.Models;
+using Microsoft.Extensions.Logging;
+
+namespace AutopilotMonitor.Functions.DataAccess.TableStorage
+{
+    /// <summary>
+    /// Azure Tables implementation of <see cref="IDecisionTransitionRepository"/>. Upserts to
+    /// the <c>DecisionTransitions</c> table via entity-group-transactions (chunked to the 100-op
+    /// limit).
+    /// </summary>
+    public sealed class TableDecisionTransitionRepository : IDecisionTransitionRepository
+    {
+        /// <summary>Azure Table Storage limit per entity-group-transaction.</summary>
+        internal const int TransactionChunkSize = 100;
+
+        private readonly TableStorageService _storage;
+        private readonly ILogger<TableDecisionTransitionRepository> _logger;
+
+        public TableDecisionTransitionRepository(
+            TableStorageService storage,
+            ILogger<TableDecisionTransitionRepository> logger)
+        {
+            _storage = storage;
+            _logger = logger;
+        }
+
+        public async Task<int> StoreBatchAsync(
+            IReadOnlyList<DecisionTransitionRecord> records,
+            CancellationToken cancellationToken = default)
+        {
+            if (records == null || records.Count == 0) return 0;
+
+            foreach (var r in records)
+            {
+                SecurityValidator.EnsureValidGuid(r.TenantId, "TenantId");
+                SecurityValidator.EnsureValidGuid(r.SessionId, "SessionId");
+            }
+
+            var table = _storage.GetTableClient(Constants.TableNames.DecisionTransitions);
+            var committed = 0;
+
+            foreach (var group in records.GroupBy(r => (r.TenantId, r.SessionId)))
+            {
+                var ordered = group.OrderBy(r => r.StepIndex).ToList();
+
+                for (var offset = 0; offset < ordered.Count; offset += TransactionChunkSize)
+                {
+                    cancellationToken.ThrowIfCancellationRequested();
+
+                    var chunk = ordered.Skip(offset).Take(TransactionChunkSize).ToList();
+                    var actions = chunk
+                        .Select(r => new TableTransactionAction(
+                            TableTransactionActionType.UpsertReplace,
+                            ToEntity(r)))
+                        .ToList();
+
+                    await table.SubmitTransactionAsync(actions, cancellationToken).ConfigureAwait(false);
+                    committed += chunk.Count;
+                }
+
+                _logger.LogDebug(
+                    "DecisionTransitions: committed {Count} rows for {Tenant}_{Session}",
+                    ordered.Count, group.Key.TenantId, group.Key.SessionId);
+            }
+
+            return committed;
+        }
+
+        /// <summary>
+        /// Projects a <see cref="DecisionTransitionRecord"/> onto its Azure <see cref="TableEntity"/>
+        /// shape. Keys: PK = <c>{TenantId}_{SessionId}</c>, RK = <c>{StepIndex:D10}</c>.
+        /// </summary>
+        internal static TableEntity ToEntity(DecisionTransitionRecord r)
+        {
+            var pk = BuildPartitionKey(r.TenantId, r.SessionId);
+            var rk = BuildRowKey(r.StepIndex);
+
+            var entity = new TableEntity(pk, rk)
+            {
+                ["TenantId"] = r.TenantId,
+                ["SessionId"] = r.SessionId,
+                ["StepIndex"] = r.StepIndex,
+                ["SessionTraceOrdinal"] = r.SessionTraceOrdinal,
+                ["SignalOrdinalRef"] = r.SignalOrdinalRef,
+                ["OccurredAtUtc"] = r.OccurredAtUtc,
+                ["Trigger"] = r.Trigger ?? string.Empty,
+                ["FromStage"] = r.FromStage ?? string.Empty,
+                ["ToStage"] = r.ToStage ?? string.Empty,
+                ["Taken"] = r.Taken,
+                ["DeadEndReason"] = r.DeadEndReason,
+                ["ReducerVersion"] = r.ReducerVersion ?? string.Empty,
+                ["IsTerminal"] = r.IsTerminal,
+                ["ClassifierVerdictId"] = r.ClassifierVerdictId,
+                ["ClassifierHypothesisLevel"] = r.ClassifierHypothesisLevel,
+            };
+
+            foreach (var kv in TableStorageChunking.ChunkProperty("PayloadJson", r.PayloadJson ?? string.Empty))
+            {
+                entity[kv.Key] = kv.Value;
+            }
+
+            return entity;
+        }
+
+        internal static string BuildPartitionKey(string tenantId, string sessionId)
+            => $"{tenantId}_{sessionId}";
+
+        internal static string BuildRowKey(int stepIndex)
+            => stepIndex.ToString("D10");
+    }
+}
@@ -0,0 +1,112 @@
+using System.Collections.Generic;
+using System.Linq;
+using System.Threading;
+using System.Threading.Tasks;
+using Azure.Data.Tables;
+using AutopilotMonitor.Functions.Security;
+using AutopilotMonitor.Functions.Services;
+using AutopilotMonitor.Shared;
+using AutopilotMonitor.Shared.DataAccess;
+using AutopilotMonitor.Shared.Models;
+using Microsoft.Extensions.Logging;
+
+namespace AutopilotMonitor.Functions.DataAccess.TableStorage
+{
+    /// <summary>
+    /// Azure Tables implementation of <see cref="ISignalRepository"/>. Upserts to the
+    /// <c>Signals</c> table via entity-group-transactions (chunked to the 100-op limit).
+    /// </summary>
+    public sealed class TableSignalRepository : ISignalRepository
+    {
+        /// <summary>Azure Table Storage limit per entity-group-transaction.</summary>
+        internal const int TransactionChunkSize = 100;
+
+        private readonly TableStorageService _storage;
+        private readonly ILogger<TableSignalRepository> _logger;
+
+        public TableSignalRepository(TableStorageService storage, ILogger<TableSignalRepository> logger)
+        {
+            _storage = storage;
+            _logger = logger;
+        }
+
+        public async Task<int> StoreBatchAsync(
+            IReadOnlyList<SignalRecord> records,
+            CancellationToken cancellationToken = default)
+        {
+            if (records == null || records.Count == 0) return 0;
+
+            foreach (var r in records)
+            {
+                SecurityValidator.EnsureValidGuid(r.TenantId, "TenantId");
+                SecurityValidator.EnsureValidGuid(r.SessionId, "SessionId");
+            }
+
+            var table = _storage.GetTableClient(Constants.TableNames.Signals);
+            var committed = 0;
+
+            foreach (var group in records.GroupBy(r => (r.TenantId, r.SessionId)))
+            {
+                var ordered = group.OrderBy(r => r.SessionSignalOrdinal).ToList();
+
+                for (var offset = 0; offset < ordered.Count; offset += TransactionChunkSize)
+                {
+                    cancellationToken.ThrowIfCancellationRequested();
+
+                    var chunk = ordered.Skip(offset).Take(TransactionChunkSize).ToList();
+                    var actions = chunk
+                        .Select(r => new TableTransactionAction(
+                            TableTransactionActionType.UpsertReplace,
+                            ToEntity(r)))
+                        .ToList();
+
+                    await table.SubmitTransactionAsync(actions, cancellationToken).ConfigureAwait(false);
+                    committed += chunk.Count;
+                }
+
+                _logger.LogDebug(
+                    "Signals: committed {Count} rows for {Tenant}_{Session}",
+                    ordered.Count, group.Key.TenantId, group.Key.SessionId);
+            }
+
+            return committed;
+        }
+
+        /// <summary>
+        /// Projects a <see cref="SignalRecord"/> onto its Azure <see cref="TableEntity"/> shape.
+        /// Keys: PK = <c>{TenantId}_{SessionId}</c>, RK = <c>{SessionSignalOrdinal:D19}</c>
+        /// (19 digits covers long.MaxValue for lexicographic ordering).
+        /// </summary>
+        internal static TableEntity ToEntity(SignalRecord r)
+        {
+            var pk = BuildPartitionKey(r.TenantId, r.SessionId);
+            var rk = BuildRowKey(r.SessionSignalOrdinal);
+
+            var entity = new TableEntity(pk, rk)
+            {
+                ["TenantId"] = r.TenantId,
+                ["SessionId"] = r.SessionId,
+                ["SessionSignalOrdinal"] = r.SessionSignalOrdinal,
+                ["SessionTraceOrdinal"] = r.SessionTraceOrdinal,
+                ["Kind"] = r.Kind ?? string.Empty,
+                ["KindSchemaVersion"] = r.KindSchemaVersion,
+                ["OccurredAtUtc"] = r.OccurredAtUtc,
+                ["SourceOrigin"] = r.SourceOrigin ?? string.Empty,
+            };
+
+            // Guard the 32 K-char per-property limit — DecisionSignal Evidence+Payload can cross it.
+            foreach (var kv in TableStorageChunking.ChunkProperty("PayloadJson", r.PayloadJson ?? string.Empty))
+            {
+                entity[kv.Key] = kv.Value;
+            }
+
+            return entity;
+        }
+
+        internal static string BuildPartitionKey(string tenantId, string sessionId)
+            => $"{tenantId}_{sessionId}";
+
+        internal static string BuildRowKey(long sessionSignalOrdinal)
+            => sessionSignalOrdinal.ToString("D19");
+    }
+}
@@ -343,6 +343,12 @@ namespace AutopilotMonitor.Shared
            // Lightweight index of sessions that have events (for orphan detection)
            public const string EventSessionIndex = "EventSessionIndex";

+            // V2 Decision Engine primary tables (Plan §M5).
+            // SignalLog (input-truth) and Journal (decision-truth) projected to the backend for
+            // the Inspector + Reducer-Verifier. Both partitioned by {TenantId}_{SessionId}.
+            public const string Signals             = "Signals";
+            public const string DecisionTransitions = "DecisionTransitions";
+
            /// <summary>
            /// Returns all table names for initialization
            /// </summary>
@@ -384,7 +390,9 @@ namespace AutopilotMonitor.Shared
                OpsEvents,
                ImeVersionHistory,
                RuleStats,
-                EventSessionIndex
+                EventSessionIndex,
+                Signals,
+                DecisionTransitions
            };
        }
    }
@@ -0,0 +1,23 @@
+using System.Collections.Generic;
+using System.Threading;
+using System.Threading.Tasks;
+using AutopilotMonitor.Shared.Models;
+
+namespace AutopilotMonitor.Shared.DataAccess
+{
+    /// <summary>
+    /// Repository for the <c>DecisionTransitions</c> primary table (Plan §M5). Writes projected
+    /// <see cref="DecisionTransitionRecord"/>s derived from agent Journal entries.
+    /// <para>
+    /// Writes are idempotent via <c>(PartitionKey, RowKey)</c>: a retried batch replays without
+    /// duplicating rows. Records within a single batch must share the same (TenantId, SessionId).
+    /// </para>
+    /// </summary>
+    public interface IDecisionTransitionRepository
+    {
+        /// <summary>
+        /// Upserts a batch of transition records. Returns the number of rows committed.
+        /// </summary>
+        Task<int> StoreBatchAsync(IReadOnlyList<DecisionTransitionRecord> records, CancellationToken cancellationToken = default);
+    }
+}
@@ -0,0 +1,25 @@
+using System.Collections.Generic;
+using System.Threading;
+using System.Threading.Tasks;
+using AutopilotMonitor.Shared.Models;
+
+namespace AutopilotMonitor.Shared.DataAccess
+{
+    /// <summary>
+    /// Repository for the <c>Signals</c> primary table (Plan §M5). Writes projected
+    /// <see cref="SignalRecord"/>s derived from agent-emitted DecisionSignals.
+    /// <para>
+    /// Writes are idempotent via <c>(PartitionKey, RowKey)</c>: a retried batch replays
+    /// without duplicating rows. Records within a single batch must share the same
+    /// (TenantId, SessionId) — the repository groups/chunks per the Azure Tables 100-op
+    /// transaction limit.
+    /// </para>
+    /// </summary>
+    public interface ISignalRepository
+    {
+        /// <summary>
+        /// Upserts a batch of signal records. Returns the number of rows committed.
+        /// </summary>
+        Task<int> StoreBatchAsync(IReadOnlyList<SignalRecord> records, CancellationToken cancellationToken = default);
+    }
+}
@@ -0,0 +1,59 @@
+using System;
+
+namespace AutopilotMonitor.Shared.Models
+{
+    /// <summary>
+    /// Backend storage record for a single DecisionTransition journal step (Plan §M5).
+    /// Flat shape projected from the agent's DecisionTransition — see <see cref="SignalRecord"/>
+    /// for the overall projection pattern.
+    /// <para>
+    /// PK = <c>{TenantId}_{SessionId}</c>, RK = <c>{StepIndex:D10}</c>.
+    /// </para>
+    /// <para>
+    /// The index-discriminator columns (<see cref="IsTerminal"/>, <see cref="ClassifierVerdictId"/>,
+    /// <see cref="ClassifierHypothesisLevel"/>) are projected eagerly so index-table fan-out
+    /// (SessionsByTerminal / ClassifierVerdictsByIdLevel / …) can happen without re-parsing
+    /// <see cref="PayloadJson"/>. The primary write stores them inline; the queue-driven
+    /// index writer reads them straight off the primary row.
+    /// </para>
+    /// </summary>
+    public sealed class DecisionTransitionRecord
+    {
+        public string TenantId { get; set; } = string.Empty;
+        public string SessionId { get; set; } = string.Empty;
+
+        /// <summary>Monotonic per Journal. Drives table RowKey ordering.</summary>
+        public int StepIndex { get; set; }
+
+        /// <summary>Session-wide monotonic across Event + Signal + Transition.</summary>
+        public long SessionTraceOrdinal { get; set; }
+
+        /// <summary>Plan §2.2 cross-reference — which signal caused this transition.</summary>
+        public long SignalOrdinalRef { get; set; }
+
+        public DateTime OccurredAtUtc { get; set; }
+
+        public string Trigger { get; set; } = string.Empty;
+
+        /// <summary>SessionStage enum name.</summary>
+        public string FromStage { get; set; } = string.Empty;
+        public string ToStage { get; set; } = string.Empty;
+
+        /// <summary>false = dead-end (blocked path, <see cref="DeadEndReason"/> non-null).</summary>
+        public bool Taken { get; set; }
+
+        public string? DeadEndReason { get; set; }
+
+        public string ReducerVersion { get; set; } = string.Empty;
+
+        /// <summary>Projected index discriminator — true when <see cref="ToStage"/> is a terminal stage.</summary>
+        public bool IsTerminal { get; set; }
+
+        /// <summary>Projected index discriminator — non-null when this step embeds a classifier verdict.</summary>
+        public string? ClassifierVerdictId { get; set; }
+        public string? ClassifierHypothesisLevel { get; set; }
+
+        /// <summary>Agent-serialized DecisionTransition JSON (Guards + EmittedEventSequences + ClassifierVerdict included).</summary>
+        public string PayloadJson { get; set; } = string.Empty;
+    }
+}
@@ -0,0 +1,43 @@
+using System;
+
+namespace AutopilotMonitor.Shared.Models
+{
+    /// <summary>
+    /// Backend storage record for a single DecisionSignal (Plan §M5). Flat shape projected
+    /// from the agent's DecisionSignal so the Backend doesn't need to reference DecisionCore
+    /// just to persist and serve it back out.
+    /// <para>
+    /// <b>Keys</b> (authoritative, supplied by the agent for idempotent upsert via
+    /// <c>(PartitionKey, RowKey)</c>):
+    /// PK = <c>{TenantId}_{SessionId}</c>, RK = <c>{SessionSignalOrdinal:D19}</c>.
+    /// </para>
+    /// <para>
+    /// <b>Fidelity</b>: <see cref="PayloadJson"/> carries the complete agent-serialized
+    /// DecisionSignal (including Evidence + Payload dictionary). Typed columns exist only
+    /// for query/projection; replay is driven off the JSON blob.
+    /// </para>
+    /// </summary>
+    public sealed class SignalRecord
+    {
+        public string TenantId { get; set; } = string.Empty;
+        public string SessionId { get; set; } = string.Empty;
+
+        /// <summary>Monotonic per SignalLog. Drives table RowKey ordering.</summary>
+        public long SessionSignalOrdinal { get; set; }
+
+        /// <summary>Session-wide monotonic across Event + Signal + Transition. Inspector correlation only.</summary>
+        public long SessionTraceOrdinal { get; set; }
+
+        /// <summary>DecisionSignalKind enum name — stored as string for forward-compat.</summary>
+        public string Kind { get; set; } = string.Empty;
+
+        public int KindSchemaVersion { get; set; }
+
+        public DateTime OccurredAtUtc { get; set; }
+
+        public string SourceOrigin { get; set; } = string.Empty;
+
+        /// <summary>Agent-serialized DecisionSignal JSON (Evidence + Payload included).</summary>
+        public string PayloadJson { get; set; } = string.Empty;
+    }
+}