feat: Implement telemetry processing with schema validation and dead-letter queue support

- Added TelemetryQueueOptions for queue configuration including dead-letter settings.
- Integrated SchemaRegistry for validating telemetry messages against defined schemas.
- Enhanced TelemetryProcessor to handle schema validation and requeue logic.
- Introduced SchemaRegistryService to manage schema registration and validation.
- Created SchemaRegistrySeeder to seed initial schema definitions on startup.
- Added new models and migrations for database schema related to telemetry and schema registry.
- Updated appsettings.json to include new configuration options for telemetry and schema registry.

Author: JasonEran

.config/dotnet-tools.json

@@ -0,0 +1,13 @@
+{
+  "version": 1,
+  "isRoot": true,
+  "tools": {
+    "dotnet-ef": {
+      "version": "8.0.23",
+      "commands": [
+        "dotnet-ef"
+      ],
+      "rollForward": false
+    }
+  }
+}

.gitignore

@@ -23,6 +23,7 @@ CMakeCache.txt
 cmake_install.cmake
 Makefile
 build/
+build_nogrpc/
 src/services/agent-cpp/build*/
 src/services/agent-cpp/build_*/
 *.o
@@ -68,6 +69,11 @@ coverage/
 
 # Local runtime artifacts
 Data/
+!src/services/core-dotnet/AetherGuard.Core/Data/
+!src/services/core-dotnet/AetherGuard.Core/Data/Migrations/
+!src/services/core-dotnet/AetherGuard.Core/Data/Migrations/**
+src/services/core-dotnet/AetherGuard.Core/Data/Snapshots/
+src/services/core-dotnet/AetherGuard.Core/Data/market_signal.json
 dummy_snapshot.tar.gz
 .compose.override.temp.yml
 .act-logs/

README.md

@@ -18,7 +18,7 @@ v2.2 reference architecture with a concrete implementation guide.
 
 ## Project Status
 
-- Stage: v2.2 baseline delivered (Phase 0-4). Remaining productization gaps tracked below.
+- Stage: v2.2 baseline delivered (Phase 0-4). Remaining productization gaps tracked below (currently none).
 - License: MIT
 - Authors: Qi Junyi, Xiao Erdong (2026)
 
@@ -56,15 +56,13 @@ This project targets a product-grade release, not a demo. The following standard
 - Dashboard (Next.js): telemetry and command visibility with NextAuth credentials.
 - Storage: snapshots stored on local filesystem by default; optional S3/MinIO backend with retention sweeper and S3 lifecycle support.
 - Security: API key for command endpoints; SPIRE mTLS for agent/core; OpenTelemetry baseline across core/AI/dashboard.
-- Messaging: telemetry queue payloads are wrapped in a schema-versioned envelope (v1).
+- Messaging: telemetry queue payloads are wrapped in a schema-versioned envelope (v1) with DLQ routing and prefetch backpressure.
+- Schema registry: telemetry envelope and payload schemas are registered in Postgres and validated at ingest.
 - Supply chain: SBOM generation, cosign signing, and SLSA provenance in CI.
 
 ### Productization Gaps (v1.x)
 
-- No schema registry or formal compatibility policy for MQ events (only a lightweight schema-version envelope is in place).
-- No EF Core migrations or formal upgrade path (production requires schema versioning + migrations).
-
-Code scan note: no TODO/FIXME markers found in the repo; the remaining gaps are architectural items listed above.
+- None for the v2.2 baseline in this repo. Remaining enhancements are tracked via issues/roadmap.
 
 ## v2.2 Reference Architecture
 
@@ -297,6 +295,20 @@ Telemetry schema policy (core-service):
 - TelemetrySchema__MaxSupportedVersion=1
 - TelemetrySchema__OnUnsupported=drop
 
+Telemetry queue options (core-service):
+
+- TelemetryQueue__QueueName=telemetry_data
+- TelemetryQueue__EnableDeadLettering=true
+- TelemetryQueue__DeadLetterExchange=telemetry_dlx
+- TelemetryQueue__DeadLetterQueueName=telemetry_data.dlq
+- TelemetryQueue__DeadLetterRoutingKey=telemetry_data.dlq
+- TelemetryQueue__PrefetchCount=50
+
+Schema registry (core-service):
+
+- SchemaRegistry__Enabled=true
+- SchemaRegistry__Compatibility=BACKWARD
+
 Agent OpenTelemetry (agent-service):
 
 - AG_OTEL_ENABLED=true
@@ -412,10 +424,24 @@ TelemetryRecord persisted to PostgreSQL:
 - PredictedCpu
 - Timestamp (UTC)
 
-For production, add EF Core migrations and a formal upgrade process.
+EF Core migrations are included; use them as the basis for upgrade and rollback workflows.
 
 ## Development
 
+### Database Migrations
+
+Generate or update migrations:
+
+```bash
+dotnet tool run dotnet-ef migrations add InitialCreate \
+  --project src/services/core-dotnet/AetherGuard.Core/AetherGuard.Core.csproj \
+  --startup-project src/services/core-dotnet/AetherGuard.Core/AetherGuard.Core.csproj \
+  --context ApplicationDbContext \
+  --output-dir Data/Migrations
+```
+
+Use `AG_DB_CONNECTION` to override the connection string during migration generation.
+
 Dashboard:
 
 ```bash

src/services/core-dotnet/AetherGuard.Core/AetherGuard.Core.csproj

@@ -11,6 +11,7 @@
     <PackageReference Include="Google.Api.CommonProtos" Version="2.17.0" />
     <PackageReference Include="Google.Protobuf" Version="3.31.1" />
     <PackageReference Include="Grpc.AspNetCore" Version="2.62.0" />
+    <PackageReference Include="JsonSchema.Net" Version="8.0.5" />
     <PackageReference Include="Microsoft.AspNetCore.Grpc.JsonTranscoding" Version="8.0.23" />
     <PackageReference Include="Grpc.Tools" Version="2.62.0" PrivateAssets="All" />
     <PackageReference Include="Microsoft.AspNetCore.OpenApi" Version="8.0.23" />
@@ -32,15 +33,9 @@
   </ItemGroup>
 
   <ItemGroup>
-    <Protobuf Include="..\..\..\shared\protos\common.proto"
-              GrpcServices="None"
-              AdditionalImportDirs="..\..\..\shared\protos" />
-    <Protobuf Include="..\..\..\shared\protos\agent_service.proto"
-              GrpcServices="Server"
-              AdditionalImportDirs="..\..\..\shared\protos" />
-    <Protobuf Include="..\..\..\shared\protos\control_plane.proto"
-              GrpcServices="Server"
-              AdditionalImportDirs="..\..\..\shared\protos" />
+    <Protobuf Include="..\..\..\shared\protos\common.proto" GrpcServices="None" AdditionalImportDirs="..\..\..\shared\protos" />
+    <Protobuf Include="..\..\..\shared\protos\agent_service.proto" GrpcServices="Server" AdditionalImportDirs="..\..\..\shared\protos" />
+    <Protobuf Include="..\..\..\shared\protos\control_plane.proto" GrpcServices="Server" AdditionalImportDirs="..\..\..\shared\protos" />
   </ItemGroup>
 
 </Project>

src/services/core-dotnet/AetherGuard.Core/Data/ApplicationDbContext.cs

@@ -14,6 +14,7 @@ public ApplicationDbContext(DbContextOptions<ApplicationDbContext> options)
     public DbSet<AgentCommand> AgentCommands => Set<AgentCommand>();
     public DbSet<CommandAudit> CommandAudits => Set<CommandAudit>();
     public DbSet<TelemetryRecord> TelemetryRecords => Set<TelemetryRecord>();
+    public DbSet<SchemaRegistryEntry> SchemaRegistryEntries => Set<SchemaRegistryEntry>();
 
     protected override void OnModelCreating(ModelBuilder modelBuilder)
     {
@@ -74,5 +75,17 @@ protected override void OnModelCreating(ModelBuilder modelBuilder)
             entity.Property(e => e.Error).HasColumnName("error");
             entity.Property(e => e.CreatedAt).HasColumnName("created_at");
         });
+
+        modelBuilder.Entity<SchemaRegistryEntry>(entity =>
+        {
+            entity.ToTable("schema_registry");
+            entity.HasKey(e => e.Id);
+            entity.HasIndex(e => new { e.Subject, e.Version }).IsUnique();
+            entity.Property(e => e.Id).HasColumnName("id").ValueGeneratedOnAdd();
+            entity.Property(e => e.Subject).HasColumnName("subject");
+            entity.Property(e => e.Version).HasColumnName("version");
+            entity.Property(e => e.Schema).HasColumnName("schema");
+            entity.Property(e => e.CreatedAt).HasColumnName("created_at");
+        });
     }
 }

src/services/core-dotnet/AetherGuard.Core/Data/DesignTimeDbContextFactory.cs

@@ -0,0 +1,17 @@
+using Microsoft.EntityFrameworkCore;
+using Microsoft.EntityFrameworkCore.Design;
+
+namespace AetherGuard.Core.Data;
+
+public sealed class DesignTimeDbContextFactory : IDesignTimeDbContextFactory<ApplicationDbContext>
+{
+    public ApplicationDbContext CreateDbContext(string[] args)
+    {
+        var connectionString = Environment.GetEnvironmentVariable("AG_DB_CONNECTION")
+            ?? "Host=localhost;Database=AetherGuardDb;Username=postgres;Password=password";
+
+        var optionsBuilder = new DbContextOptionsBuilder<ApplicationDbContext>();
+        optionsBuilder.UseNpgsql(connectionString);
+        return new ApplicationDbContext(optionsBuilder.Options);
+    }
+}