From f0fc3696f62a7a680dc18982bcbf2f9b599c3e7a Mon Sep 17 00:00:00 2001 From: Oliver Kieselbach Date: Tue, 21 Apr 2026 09:33:36 +0200 Subject: [PATCH] =?UTF-8?q?feat(backend):=20M5.b=20IngestTelemetry=20endpo?= =?UTF-8?q?int=20=E2=80=94=20POST=20/api/agent/telemetry?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit V2 ingest endpoint consuming a heterogeneous batch of TelemetryItemDtos (Event + Signal + DecisionTransition in one JSON array). gzip request bodies are decoded transparently by the existing UseRequestDecompression middleware — new code only has to read a plain stream. - TelemetryItemDto (Shared): wire-format DTO mirroring the agent's TelemetryItem JSON shape so the backend can deserialise without linking agent assemblies. - IngestTelemetryFunction: security check (cert / rate / hardware whitelist) → device & version kill-switch short-circuits with DeviceBlocked=true body → body parse → tenant-header/body match → partition by Kind → persist via the appropriate repository → 200 with AdminAction + Actions control-signal fields populated from ISessionRepository (matching the /api/agent/ingest parity needed by the M4.6.ε agent-side UploadResult parser). - TelemetryPayloadParser: pure defensive parsing layer (wire → storage record). Malformed PayloadJson on a single item returns null; one poisonous item can't poison the batch. Projects index discriminators (IsTerminal, ClassifierVerdictId, ClassifierHypothesisLevel) off the DecisionTransition payload so the later queue-driven index fan-out reads them straight off the primary row. - EndpointAccessPolicyCatalog: POST agent/telemetry registered under DeviceOrBootstrapAuth (same policy as agent/ingest). Without this the fail-closed catalog rejects the route with 403. Scope boundary — Events get stored via StoreEventsBatchAsync only; the rule engine, app-install aggregation, SignalR notification, and vulnerability correlation remain on the legacy /api/agent/ingest path. Sharing that pipeline is tracked as M5.b.2 (extract EventIngestProcessor so both functions run the same downstream steps) — M7 rollout gates on completing it. Documented inline in PersistItemsAsync. 28 new tests: defensive null-handling per kind, typed-column extraction for Signal (with envelope fallback for missing SessionTraceOrdinal), DecisionTransition dead-end + classifier-verdict nesting, IsTerminal authoritative-list guard (Completed / Failed / WhiteGloveSealed / WhiteGloveCompletedPart2 — mirrors DecisionCore SessionStageExtensions), PartitionKey parser including malformed-shape rejection. The EndpointPolicyCatalogCompletenessTests guard the route registration. Co-Authored-By: Claude Opus 4.7 (1M context) --- .../IngestTelemetryFunctionTests.cs | 42 +++ .../TelemetryPayloadParserTests.cs | 207 +++++++++++ .../Ingest/IngestTelemetryFunction.cs | 345 ++++++++++++++++++ .../Ingest/TelemetryPayloadParser.cs | 146 ++++++++ .../Security/EndpointAccessPolicyCatalog.cs | 1 + .../Models/TelemetryItemDto.cs | 44 +++ 6 files changed, 785 insertions(+) create mode 100644 src/Backend/AutopilotMonitor.Functions.Tests/IngestTelemetryFunctionTests.cs create mode 100644 src/Backend/AutopilotMonitor.Functions.Tests/TelemetryPayloadParserTests.cs create mode 100644 src/Backend/AutopilotMonitor.Functions/Functions/Ingest/IngestTelemetryFunction.cs create mode 100644 src/Backend/AutopilotMonitor.Functions/Functions/Ingest/TelemetryPayloadParser.cs create mode 100644 src/Shared/AutopilotMonitor.Shared/Models/TelemetryItemDto.cs diff --git a/src/Backend/AutopilotMonitor.Functions.Tests/IngestTelemetryFunctionTests.cs b/src/Backend/AutopilotMonitor.Functions.Tests/IngestTelemetryFunctionTests.cs new file mode 100644 index 00000000..26908e3b --- /dev/null +++ b/src/Backend/AutopilotMonitor.Functions.Tests/IngestTelemetryFunctionTests.cs @@ -0,0 +1,42 @@ +using AutopilotMonitor.Functions.Functions.Ingest; + +namespace AutopilotMonitor.Functions.Tests; + +/// +/// Tests for the pure helpers on . The HTTP-trigger end +/// of the function needs a live runtime harness to exercise; this file covers the deterministic +/// bits (PartitionKey parsing, body/header TenantId reconciliation shape). +/// +public class IngestTelemetryFunctionTests +{ + private const string TenantId = "a1b2c3d4-e5f6-7890-abcd-ef1234567890"; + private const string SessionId = "b2c3d4e5-f6a7-8901-bcde-f12345678901"; + + [Fact] + public void TryParsePartitionKey_splits_tenant_and_session_on_single_underscore() + { + var ok = IngestTelemetryFunction.TryParsePartitionKey( + $"{TenantId}_{SessionId}", out var tenant, out var session); + + Assert.True(ok); + Assert.Equal(TenantId, tenant); + Assert.Equal(SessionId, session); + } + + [Theory] + [InlineData("")] + [InlineData(null)] + [InlineData("no-underscore-here")] + [InlineData("_missing-tenant")] + [InlineData("missing-session_")] + [InlineData("too_many_parts_here")] + public void TryParsePartitionKey_rejects_malformed_shapes(string? input) + { + var ok = IngestTelemetryFunction.TryParsePartitionKey( + input!, out var tenant, out var session); + + Assert.False(ok); + Assert.Equal(string.Empty, tenant); + Assert.Equal(string.Empty, session); + } +} diff --git a/src/Backend/AutopilotMonitor.Functions.Tests/TelemetryPayloadParserTests.cs b/src/Backend/AutopilotMonitor.Functions.Tests/TelemetryPayloadParserTests.cs new file mode 100644 index 00000000..de7e1eb0 --- /dev/null +++ b/src/Backend/AutopilotMonitor.Functions.Tests/TelemetryPayloadParserTests.cs @@ -0,0 +1,207 @@ +using AutopilotMonitor.Functions.Functions.Ingest; +using AutopilotMonitor.Shared.Models; + +namespace AutopilotMonitor.Functions.Tests; + +/// +/// Pure parsing tests for . Verifies the wire → storage +/// mapping the relies on: defensive null handling, typed +/// column extraction, and the index-discriminator projection (IsTerminal, ClassifierVerdictId, +/// ClassifierHypothesisLevel). +/// +public class TelemetryPayloadParserTests +{ + private const string TenantId = "a1b2c3d4-e5f6-7890-abcd-ef1234567890"; + private const string SessionId = "b2c3d4e5-f6a7-8901-bcde-f12345678901"; + + private static TelemetryItemDto Dto(string kind, string payloadJson, long? traceOrdinal = null) + => new TelemetryItemDto + { + Kind = kind, + PartitionKey = $"{TenantId}_{SessionId}", + RowKey = "0000000001", + TelemetryItemId = 1, + SessionTraceOrdinal = traceOrdinal, + PayloadJson = payloadJson, + RequiresImmediateFlush = false, + EnqueuedAtUtc = new DateTime(2026, 4, 21, 10, 0, 0, DateTimeKind.Utc), + RetryCount = 0, + }; + + // ============================================================ Defensive parsing + + [Theory] + [InlineData("")] + [InlineData("{ this is not json }")] + [InlineData("null")] + public void ParseEvent_returns_null_for_missing_or_malformed_payload(string payload) + { + var result = TelemetryPayloadParser.ParseEvent(Dto("Event", payload), TenantId, SessionId); + Assert.Null(result); + } + + [Theory] + [InlineData("")] + [InlineData("{ this is not json }")] + public void ParseSignal_returns_null_for_missing_or_malformed_payload(string payload) + { + var result = TelemetryPayloadParser.ParseSignal(Dto("Signal", payload), TenantId, SessionId); + Assert.Null(result); + } + + [Theory] + [InlineData("")] + [InlineData("{ this is not json }")] + public void ParseTransition_returns_null_for_missing_or_malformed_payload(string payload) + { + var result = TelemetryPayloadParser.ParseTransition(Dto("DecisionTransition", payload), TenantId, SessionId); + Assert.Null(result); + } + + // ============================================================ Event + + [Fact] + public void ParseEvent_deserialises_enrollment_event_shape() + { + var payload = + "{" + + "\"EventId\":\"evt-123\"," + + "\"EventType\":\"test_event\"," + + "\"Severity\":1," + + "\"Source\":\"test\"," + + "\"Phase\":2," + + "\"Message\":\"hello\"," + + "\"Sequence\":42," + + "\"Timestamp\":\"2026-04-21T10:00:00Z\"," + + "\"ReceivedAt\":\"2026-04-21T10:00:01Z\"" + + "}"; + + var evt = TelemetryPayloadParser.ParseEvent(Dto("Event", payload), TenantId, SessionId); + + Assert.NotNull(evt); + Assert.Equal("evt-123", evt!.EventId); + Assert.Equal("test_event", evt.EventType); + Assert.Equal(42, evt.Sequence); + } + + // ============================================================ Signal + + [Fact] + public void ParseSignal_extracts_typed_columns_and_preserves_full_payload() + { + var payload = + "{" + + "\"SessionSignalOrdinal\":17," + + "\"SessionTraceOrdinal\":117," + + "\"Kind\":\"EspPhaseChanged\"," + + "\"KindSchemaVersion\":2," + + "\"OccurredAtUtc\":\"2026-04-21T10:00:00Z\"," + + "\"SourceOrigin\":\"EspAndHelloTrackerAdapter\"," + + "\"Evidence\":{\"Kind\":\"Raw\"}" + + "}"; + + var rec = TelemetryPayloadParser.ParseSignal(Dto("Signal", payload), TenantId, SessionId); + + Assert.NotNull(rec); + Assert.Equal(TenantId, rec!.TenantId); + Assert.Equal(SessionId, rec.SessionId); + Assert.Equal(17L, rec.SessionSignalOrdinal); + Assert.Equal(117L, rec.SessionTraceOrdinal); + Assert.Equal("EspPhaseChanged", rec.Kind); + Assert.Equal(2, rec.KindSchemaVersion); + Assert.Equal(new DateTime(2026, 4, 21, 10, 0, 0, DateTimeKind.Utc), rec.OccurredAtUtc.ToUniversalTime()); + Assert.Equal("EspAndHelloTrackerAdapter", rec.SourceOrigin); + Assert.Equal(payload, rec.PayloadJson); + } + + [Fact] + public void ParseSignal_falls_back_to_envelope_SessionTraceOrdinal_when_payload_missing_field() + { + // Payload intentionally omits SessionTraceOrdinal; the envelope carries it. + var payload = "{\"SessionSignalOrdinal\":1,\"Kind\":\"SessionStarted\"}"; + + var rec = TelemetryPayloadParser.ParseSignal(Dto("Signal", payload, traceOrdinal: 99), TenantId, SessionId); + + Assert.NotNull(rec); + Assert.Equal(99L, rec!.SessionTraceOrdinal); + } + + // ============================================================ DecisionTransition + + [Fact] + public void ParseTransition_extracts_core_columns_and_flags_taken_false_when_dead_end() + { + var payload = + "{" + + "\"StepIndex\":3," + + "\"SessionTraceOrdinal\":10," + + "\"SignalOrdinalRef\":9," + + "\"OccurredAtUtc\":\"2026-04-21T10:00:00Z\"," + + "\"Trigger\":\"EspExiting\"," + + "\"FromStage\":\"EspInProgress\"," + + "\"ToStage\":\"EspInProgress\"," + + "\"Taken\":false," + + "\"DeadEndReason\":\"hybrid_reboot_gate_blocking\"," + + "\"ReducerVersion\":\"1.0.0\"" + + "}"; + + var rec = TelemetryPayloadParser.ParseTransition(Dto("DecisionTransition", payload), TenantId, SessionId); + + Assert.NotNull(rec); + Assert.False(rec!.Taken); + Assert.Equal("hybrid_reboot_gate_blocking", rec.DeadEndReason); + Assert.False(rec.IsTerminal); + Assert.Null(rec.ClassifierVerdictId); + } + + [Fact] + public void ParseTransition_projects_IsTerminal_true_for_terminal_stages() + { + var terminalStages = new[] { "Completed", "Failed", "WhiteGloveSealed", "WhiteGloveCompletedPart2" }; + + foreach (var stage in terminalStages) + { + var payload = + "{\"StepIndex\":1,\"SessionTraceOrdinal\":1,\"SignalOrdinalRef\":1," + + $"\"Trigger\":\"t\",\"FromStage\":\"X\",\"ToStage\":\"{stage}\",\"Taken\":true," + + "\"ReducerVersion\":\"1.0.0\",\"OccurredAtUtc\":\"2026-04-21T10:00:00Z\"}"; + + var rec = TelemetryPayloadParser.ParseTransition(Dto("DecisionTransition", payload), TenantId, SessionId); + + Assert.NotNull(rec); + Assert.True(rec!.IsTerminal, $"Expected {stage} to be terminal"); + } + } + + [Fact] + public void ParseTransition_extracts_classifier_verdict_nested_fields() + { + var payload = + "{\"StepIndex\":1,\"SessionTraceOrdinal\":1,\"SignalOrdinalRef\":1," + + "\"Trigger\":\"t\",\"FromStage\":\"X\",\"ToStage\":\"WhiteGloveSealed\",\"Taken\":true," + + "\"ReducerVersion\":\"1.0.0\",\"OccurredAtUtc\":\"2026-04-21T10:00:00Z\"," + + "\"ClassifierVerdict\":{\"ClassifierId\":\"whiteglove-sealing\",\"HypothesisLevel\":\"Strong\"}}"; + + var rec = TelemetryPayloadParser.ParseTransition(Dto("DecisionTransition", payload), TenantId, SessionId); + + Assert.NotNull(rec); + Assert.Equal("whiteglove-sealing", rec!.ClassifierVerdictId); + Assert.Equal("Strong", rec.ClassifierHypothesisLevel); + } + + // ============================================================ IsTerminalStage + + [Theory] + [InlineData("Completed", true)] + [InlineData("Failed", true)] + [InlineData("WhiteGloveSealed", true)] + [InlineData("WhiteGloveCompletedPart2", true)] + [InlineData("EspInProgress", false)] + [InlineData("AccountSetup", false)] + [InlineData("", false)] + [InlineData("completed", false)] // case-sensitive — enum names are PascalCase + public void IsTerminalStage_matches_DecisionCore_authoritative_list(string stage, bool expected) + { + Assert.Equal(expected, TelemetryPayloadParser.IsTerminalStage(stage)); + } +} diff --git a/src/Backend/AutopilotMonitor.Functions/Functions/Ingest/IngestTelemetryFunction.cs b/src/Backend/AutopilotMonitor.Functions/Functions/Ingest/IngestTelemetryFunction.cs new file mode 100644 index 00000000..0b764e60 --- /dev/null +++ b/src/Backend/AutopilotMonitor.Functions/Functions/Ingest/IngestTelemetryFunction.cs @@ -0,0 +1,345 @@ +using System.IO; +using System.Net; +using AutopilotMonitor.Functions.Security; +using AutopilotMonitor.Functions.Services; +using AutopilotMonitor.Shared.DataAccess; +using AutopilotMonitor.Shared.Models; +using Microsoft.Azure.Functions.Worker; +using Microsoft.Azure.Functions.Worker.Http; +using Microsoft.Extensions.Logging; +using Newtonsoft.Json; + +namespace AutopilotMonitor.Functions.Functions.Ingest +{ + /// + /// V2 ingest endpoint. Consumes a heterogeneous batch of s + /// (Events + Signals + DecisionTransitions in a single JSON array, gzip-compressed on + /// the wire — the UseRequestDecompression middleware decompresses before we + /// parse). Plan §2.7a / §M5 / M4.6.ε. + /// + /// Routing by : + /// + /// Event → (storage only in this build — full event pipeline sharing with /api/agent/ingest is tracked as M5.b.2). + /// Signal → + /// DecisionTransition → + /// + /// + /// + /// Response (M4.6.ε): the agent parses DeviceBlocked/UnblockAt/ + /// DeviceKillSignal/AdminAction/Actions from the 2xx body and routes + /// kill-switches through its ServerActionDispatcher. Populated from the same + /// services the legacy /api/agent/ingest endpoint uses so behaviour is at parity. + /// + /// + public sealed class IngestTelemetryFunction + { + private readonly ILogger _logger; + private readonly ISessionRepository _sessionRepo; + private readonly ISignalRepository _signalRepo; + private readonly IDecisionTransitionRepository _transitionRepo; + private readonly TenantConfigurationService _configService; + private readonly RateLimitService _rateLimitService; + private readonly AutopilotDeviceValidator _autopilotDeviceValidator; + private readonly CorporateIdentifierValidator _corporateIdentifierValidator; + private readonly DeviceAssociationValidator _deviceAssociationValidator; + private readonly BootstrapSessionService _bootstrapSessionService; + private readonly BlockedDeviceService _blockedDeviceService; + private readonly BlockedVersionService _blockedVersionService; + + public IngestTelemetryFunction( + ILogger logger, + ISessionRepository sessionRepo, + ISignalRepository signalRepo, + IDecisionTransitionRepository transitionRepo, + TenantConfigurationService configService, + RateLimitService rateLimitService, + AutopilotDeviceValidator autopilotDeviceValidator, + CorporateIdentifierValidator corporateIdentifierValidator, + DeviceAssociationValidator deviceAssociationValidator, + BootstrapSessionService bootstrapSessionService, + BlockedDeviceService blockedDeviceService, + BlockedVersionService blockedVersionService) + { + _logger = logger; + _sessionRepo = sessionRepo; + _signalRepo = signalRepo; + _transitionRepo = transitionRepo; + _configService = configService; + _rateLimitService = rateLimitService; + _autopilotDeviceValidator = autopilotDeviceValidator; + _corporateIdentifierValidator = corporateIdentifierValidator; + _deviceAssociationValidator = deviceAssociationValidator; + _bootstrapSessionService = bootstrapSessionService; + _blockedDeviceService = blockedDeviceService; + _blockedVersionService = blockedVersionService; + } + + [Function("IngestTelemetry")] + public async Task Run( + [HttpTrigger(AuthorizationLevel.Anonymous, "post", Route = "agent/telemetry")] HttpRequestData req) + { + try + { + var tenantIdHeader = req.Headers.Contains("X-Tenant-Id") + ? req.Headers.GetValues("X-Tenant-Id").FirstOrDefault() + : null; + + if (string.IsNullOrEmpty(tenantIdHeader)) + { + return await WriteErrorAsync(req, HttpStatusCode.BadRequest, "X-Tenant-Id header is required"); + } + + var (validation, errorResponse) = await req.ValidateSecurityAsync( + tenantIdHeader, + _configService, + _rateLimitService, + _autopilotDeviceValidator, + _corporateIdentifierValidator, + _logger, + bootstrapSessionService: _bootstrapSessionService, + deviceAssociationValidator: _deviceAssociationValidator); + + if (errorResponse != null) return errorResponse; + + // Device + version kill-switches: short-circuit before parsing the body. + var serialNumberHeader = req.Headers.Contains("X-Device-SerialNumber") + ? req.Headers.GetValues("X-Device-SerialNumber").FirstOrDefault() + : null; + var agentVersionHeader = req.Headers.Contains("X-Agent-Version") + ? req.Headers.GetValues("X-Agent-Version").FirstOrDefault() + : null; + + var killResponse = await CheckKillSwitchesAsync( + req, tenantIdHeader, serialNumberHeader, agentVersionHeader); + if (killResponse != null) return killResponse; + + // Parse request body (already gzip-decompressed by middleware if Content-Encoding: gzip). + List? items; + try + { + using var reader = new StreamReader(req.Body); + var body = await reader.ReadToEndAsync(); + items = JsonConvert.DeserializeObject>(body); + } + catch (JsonException ex) + { + _logger.LogWarning(ex, "IngestTelemetry: malformed JSON body"); + return await WriteErrorAsync(req, HttpStatusCode.BadRequest, "Malformed JSON body"); + } + + if (items == null || items.Count == 0) + { + return await WriteErrorAsync(req, HttpStatusCode.BadRequest, "No telemetry items provided"); + } + + // Extract tenant+session from the first item's PartitionKey for body-vs-header tenant check + // and for AdminAction / ServerAction lookups. All items in a batch belong to one session. + if (!TryParsePartitionKey(items[0].PartitionKey, out var bodyTenantId, out var sessionId)) + { + return await WriteErrorAsync(req, HttpStatusCode.BadRequest, "Malformed PartitionKey"); + } + + if (!string.Equals(bodyTenantId, tenantIdHeader, StringComparison.OrdinalIgnoreCase)) + { + _logger.LogWarning( + "IngestTelemetry: TenantId mismatch — header={Header}, body={Body}", + tenantIdHeader, bodyTenantId); + return await WriteErrorAsync(req, HttpStatusCode.Forbidden, "TenantId mismatch between header and payload"); + } + + // Partition by Kind and persist. + var (eventCount, signalCount, transitionCount, unknownCount) = + await PersistItemsAsync(items, bodyTenantId, sessionId); + + _logger.LogInformation( + "IngestTelemetry: tenant={Tenant} session={Session} events={E} signals={S} transitions={T} unknown={U}", + bodyTenantId, sessionId, eventCount, signalCount, transitionCount, unknownCount); + + // Populate server→agent control signals (same services as /api/agent/ingest for parity). + var adminAction = await TryReadAdminActionAsync(bodyTenantId, sessionId); + var pendingActions = await _sessionRepo.FetchAndClearPendingActionsAsync(bodyTenantId, sessionId); + + var response = req.CreateResponse(HttpStatusCode.OK); + await response.WriteAsJsonAsync(new IngestEventsResponse + { + Success = true, + EventsReceived = items.Count, + EventsProcessed = eventCount + signalCount + transitionCount, + Message = $"Stored {eventCount} events, {signalCount} signals, {transitionCount} transitions", + ProcessedAt = DateTime.UtcNow, + AdminAction = adminAction, + Actions = pendingActions.Count > 0 ? pendingActions : null, + }); + return response; + } + catch (Exception ex) + { + _logger.LogError(ex, "IngestTelemetry: unhandled exception"); + return await WriteErrorAsync(req, HttpStatusCode.InternalServerError, "Internal server error"); + } + } + + /// + /// Runs device-serial and agent-version kill-switch checks. Returns a 200 response with + /// DeviceBlocked=true (and optional DeviceKillSignal=true) if the caller + /// should stop — or null if the request may proceed. + /// + private async Task CheckKillSwitchesAsync( + HttpRequestData req, string tenantId, string? serialNumber, string? agentVersion) + { + if (!string.IsNullOrEmpty(serialNumber)) + { + // Session-aware block: without body-parse we can't discriminate on SessionId, + // so we use the tenant/serial blanket check. Session-scoped blocks still require + // the agent to upload; the response just carries DeviceBlocked=true regardless of + // session — tighter scoping lands once the body is parsed (M5.b.2 pipeline share). + var (isBlocked, unblockAt, blockAction, _) = + await _blockedDeviceService.IsBlockedAsync(tenantId, serialNumber); + if (isBlocked) + { + var isKill = string.Equals(blockAction, "Kill", StringComparison.OrdinalIgnoreCase); + _logger.LogWarning( + "IngestTelemetry: {Action} device tenant={Tenant} serial={Serial} unblockAt={UnblockAt}", + isKill ? "KILL" : "Block", tenantId, serialNumber, unblockAt); + return await WriteDeviceBlockedAsync(req, isKill, unblockAt, + isKill ? "Device has been issued a remote kill signal." + : "Device is temporarily blocked by an administrator."); + } + } + + if (!string.IsNullOrEmpty(agentVersion)) + { + var (isVersionBlocked, versionAction, matchedPattern) = + await _blockedVersionService.IsVersionBlockedAsync(agentVersion); + if (isVersionBlocked) + { + var isKill = string.Equals(versionAction, "Kill", StringComparison.OrdinalIgnoreCase); + _logger.LogWarning( + "IngestTelemetry: version {Action} tenant={Tenant} agentVersion={AgentVersion} pattern={Pattern}", + isKill ? "KILL" : "block", tenantId, agentVersion, matchedPattern); + return await WriteDeviceBlockedAsync(req, isKill, null, + isKill ? $"Agent version {agentVersion} has been issued a remote kill signal (pattern: {matchedPattern})." + : $"Agent version {agentVersion} is blocked by administrator (pattern: {matchedPattern})."); + } + } + + return null; + } + + private async Task WriteDeviceBlockedAsync( + HttpRequestData req, bool isKill, DateTime? unblockAt, string message) + { + var response = req.CreateResponse(HttpStatusCode.OK); + await response.WriteAsJsonAsync(new IngestEventsResponse + { + Success = false, + DeviceBlocked = true, + DeviceKillSignal = isKill, + UnblockAt = unblockAt, + Message = message, + ProcessedAt = DateTime.UtcNow, + }); + return response; + } + + private async Task TryReadAdminActionAsync(string tenantId, string sessionId) + { + var session = await _sessionRepo.GetSessionAsync(tenantId, sessionId); + if (session == null) return null; + if (session.Status == SessionStatus.Succeeded) return "Succeeded"; + if (session.Status == SessionStatus.Failed) return "Failed"; + return null; + } + + private async Task<(int eventCount, int signalCount, int transitionCount, int unknownCount)> + PersistItemsAsync(IReadOnlyList items, string tenantId, string sessionId) + { + var events = new List(); + var signals = new List(); + var transitions = new List(); + var unknown = 0; + + foreach (var item in items) + { + switch (item.Kind) + { + case "Event": + var evt = TelemetryPayloadParser.ParseEvent(item, tenantId, sessionId); + if (evt != null) events.Add(evt); + break; + case "Signal": + var sig = TelemetryPayloadParser.ParseSignal(item, tenantId, sessionId); + if (sig != null) signals.Add(sig); + break; + case "DecisionTransition": + var tr = TelemetryPayloadParser.ParseTransition(item, tenantId, sessionId); + if (tr != null) transitions.Add(tr); + break; + default: + unknown++; + _logger.LogWarning("IngestTelemetry: unknown Kind '{Kind}' (TelemetryItemId={Id})", item.Kind, item.TelemetryItemId); + break; + } + } + + // Stamp authoritative TenantId/SessionId + ReceivedAt on events (same policy as legacy — never trust per-event agent values). + var receivedAt = DateTime.UtcNow; + foreach (var e in events) + { + e.TenantId = tenantId; + e.SessionId = sessionId; + e.ReceivedAt = receivedAt; + } + + var eventCount = 0; + if (events.Count > 0) + { + var stored = await _sessionRepo.StoreEventsBatchAsync(events); + eventCount = stored.Count; + // NOTE: rule engine, app-install aggregation, SignalR notification, vulnerability + // correlation NOT invoked here. Legacy /api/agent/ingest still owns that pipeline — + // sharing lands in M5.b.2 via an extracted EventIngestProcessor service. + } + + var signalCount = await _signalRepo.StoreBatchAsync(signals); + var transitionCount = await _transitionRepo.StoreBatchAsync(transitions); + + return (eventCount, signalCount, transitionCount, unknown); + } + + /// + /// PartitionKey convention is {tenantId}_{sessionId} — both GUIDs with dashes, so + /// splitting on the single underscore between them is unambiguous. Returns false for any + /// shape that doesn't match (malformed, extra parts, empty halves). + /// + internal static bool TryParsePartitionKey(string partitionKey, out string tenantId, out string sessionId) + { + tenantId = string.Empty; + sessionId = string.Empty; + if (string.IsNullOrEmpty(partitionKey)) return false; + + var parts = partitionKey.Split('_'); + if (parts.Length != 2) return false; + if (string.IsNullOrEmpty(parts[0]) || string.IsNullOrEmpty(parts[1])) return false; + + tenantId = parts[0]; + sessionId = parts[1]; + return true; + } + + private static async Task WriteErrorAsync( + HttpRequestData req, HttpStatusCode status, string message) + { + var response = req.CreateResponse(status); + await response.WriteAsJsonAsync(new IngestEventsResponse + { + Success = false, + EventsReceived = 0, + EventsProcessed = 0, + Message = message, + ProcessedAt = DateTime.UtcNow, + }); + return response; + } + } +} diff --git a/src/Backend/AutopilotMonitor.Functions/Functions/Ingest/TelemetryPayloadParser.cs b/src/Backend/AutopilotMonitor.Functions/Functions/Ingest/TelemetryPayloadParser.cs new file mode 100644 index 00000000..9329ea69 --- /dev/null +++ b/src/Backend/AutopilotMonitor.Functions/Functions/Ingest/TelemetryPayloadParser.cs @@ -0,0 +1,146 @@ +using AutopilotMonitor.Shared.Models; +using Newtonsoft.Json; +using Newtonsoft.Json.Linq; + +namespace AutopilotMonitor.Functions.Functions.Ingest +{ + /// + /// Pure mapping layer from wire format to backend storage + /// records. All inputs are defensive — malformed + /// returns null rather than throwing, so a single poisonous item can't take down + /// the whole batch. Unit-tested against the canonical JSON shapes the agent produces. + /// + internal static class TelemetryPayloadParser + { + /// + /// Parses an Event item's payload into an . + /// TenantId/SessionId on the event are authoritative from the caller; any agent-supplied + /// values are overwritten by . + /// + public static EnrollmentEvent? ParseEvent(TelemetryItemDto item, string tenantId, string sessionId) + { + if (string.IsNullOrEmpty(item?.PayloadJson)) return null; + try + { + var evt = JsonConvert.DeserializeObject(item.PayloadJson!); + if (evt == null) return null; + // Tenant/Session stamped by the function after return; here we only deserialise. + return evt; + } + catch (JsonException) + { + return null; + } + } + + /// + /// Parses a Signal item's payload (agent-serialised DecisionSignal) into the + /// flat storage shape. Extracts typed columns used for queries + /// and indexes while preserving the full JSON in + /// for replay fidelity. + /// + public static SignalRecord? ParseSignal(TelemetryItemDto item, string tenantId, string sessionId) + { + if (string.IsNullOrEmpty(item?.PayloadJson)) return null; + JObject root; + try { root = JObject.Parse(item.PayloadJson!); } + catch (JsonException) { return null; } + + var ordinal = (long?)root["SessionSignalOrdinal"] ?? 0L; + var traceOrdinal = (long?)root["SessionTraceOrdinal"] ?? item!.SessionTraceOrdinal ?? 0L; + var kind = (string?)root["Kind"] ?? string.Empty; + var schemaVersion = (int?)root["KindSchemaVersion"] ?? 0; + var occurredAt = (DateTime?)root["OccurredAtUtc"] ?? item!.EnqueuedAtUtc; + var sourceOrigin = (string?)root["SourceOrigin"] ?? string.Empty; + + return new SignalRecord + { + TenantId = tenantId, + SessionId = sessionId, + SessionSignalOrdinal = ordinal, + SessionTraceOrdinal = traceOrdinal, + Kind = kind, + KindSchemaVersion = schemaVersion, + OccurredAtUtc = occurredAt, + SourceOrigin = sourceOrigin, + PayloadJson = item.PayloadJson!, + }; + } + + /// + /// Parses a DecisionTransition item's payload into the flat + /// storage shape. Projects index-discriminator + /// columns eagerly so the queue-driven index fan-out (M5.b) can read them off the + /// primary row without re-parsing the JSON blob. + /// + public static DecisionTransitionRecord? ParseTransition(TelemetryItemDto item, string tenantId, string sessionId) + { + if (string.IsNullOrEmpty(item?.PayloadJson)) return null; + JObject root; + try { root = JObject.Parse(item.PayloadJson!); } + catch (JsonException) { return null; } + + var stepIndex = (int?)root["StepIndex"] ?? 0; + var traceOrdinal = (long?)root["SessionTraceOrdinal"] ?? item!.SessionTraceOrdinal ?? 0L; + var signalOrdinal = (long?)root["SignalOrdinalRef"] ?? 0L; + var occurredAt = (DateTime?)root["OccurredAtUtc"] ?? item!.EnqueuedAtUtc; + var trigger = (string?)root["Trigger"] ?? string.Empty; + var fromStage = (string?)root["FromStage"] ?? string.Empty; + var toStage = (string?)root["ToStage"] ?? string.Empty; + var taken = (bool?)root["Taken"] ?? false; + var deadEndReason = (string?)root["DeadEndReason"]; + var reducerVersion = (string?)root["ReducerVersion"] ?? string.Empty; + + // Classifier verdict — optional nested object. + string? verdictId = null, verdictLevel = null; + if (root["ClassifierVerdict"] is JObject verdict) + { + verdictId = (string?)verdict["ClassifierId"] ?? (string?)verdict["Id"]; + verdictLevel = (string?)verdict["HypothesisLevel"] ?? (string?)verdict["Level"]; + } + + return new DecisionTransitionRecord + { + TenantId = tenantId, + SessionId = sessionId, + StepIndex = stepIndex, + SessionTraceOrdinal = traceOrdinal, + SignalOrdinalRef = signalOrdinal, + OccurredAtUtc = occurredAt, + Trigger = trigger, + FromStage = fromStage, + ToStage = toStage, + Taken = taken, + DeadEndReason = deadEndReason, + ReducerVersion = reducerVersion, + IsTerminal = IsTerminalStage(toStage), + ClassifierVerdictId = verdictId, + ClassifierHypothesisLevel = verdictLevel, + PayloadJson = item.PayloadJson!, + }; + } + + /// + /// Terminal-stage classification mirrors the agent-side + /// SessionStageExtensions.IsTerminal() (DecisionCore) so backend index queries can + /// reach the same conclusion without linking DecisionCore. New terminal stages must be + /// added in both places — enforced by the SessionStageExtensions contract test on + /// the agent side. + /// + internal static bool IsTerminalStage(string toStage) + { + // Authoritative list mirrors DecisionCore SessionStageExtensions.IsTerminal(). + // Keep in sync — new terminal stages need both sites updated. + switch (toStage) + { + case "Completed": + case "Failed": + case "WhiteGloveSealed": + case "WhiteGloveCompletedPart2": + return true; + default: + return false; + } + } + } +} diff --git a/src/Backend/AutopilotMonitor.Functions/Security/EndpointAccessPolicyCatalog.cs b/src/Backend/AutopilotMonitor.Functions/Security/EndpointAccessPolicyCatalog.cs index 204ba072..3e8e8fd4 100644 --- a/src/Backend/AutopilotMonitor.Functions/Security/EndpointAccessPolicyCatalog.cs +++ b/src/Backend/AutopilotMonitor.Functions/Security/EndpointAccessPolicyCatalog.cs @@ -105,6 +105,7 @@ public static class EndpointAccessPolicyCatalog // ── DeviceOrBootstrapAuth ─────────────────────────────────────── new("POST", "agent/register-session", EndpointPolicy.DeviceOrBootstrapAuth), new("POST", "agent/ingest", EndpointPolicy.DeviceOrBootstrapAuth), + new("POST", "agent/telemetry", EndpointPolicy.DeviceOrBootstrapAuth), new("GET", "agent/config", EndpointPolicy.DeviceOrBootstrapAuth), new("POST", "agent/upload-url", EndpointPolicy.DeviceOrBootstrapAuth), new("POST", "agent/error", EndpointPolicy.DeviceOrBootstrapAuth), diff --git a/src/Shared/AutopilotMonitor.Shared/Models/TelemetryItemDto.cs b/src/Shared/AutopilotMonitor.Shared/Models/TelemetryItemDto.cs new file mode 100644 index 00000000..9c001843 --- /dev/null +++ b/src/Shared/AutopilotMonitor.Shared/Models/TelemetryItemDto.cs @@ -0,0 +1,44 @@ +using System; + +namespace AutopilotMonitor.Shared.Models +{ + /// + /// Wire-format DTO for a single telemetry item sent by the V2 agent to + /// POST /api/agent/telemetry (Plan §2.7a / §M5). The agent's in-memory class + /// AutopilotMonitor.Agent.V2.Core.Transport.Telemetry.TelemetryItem is ctor-validated + /// and immutable; this DTO mirrors the JSON shape so the backend can deserialise without + /// depending on agent assemblies. + /// + /// Serialisation: Newtonsoft.Json, PascalCase, as string + /// ("Event"/"Signal"/"DecisionTransition") for forward-compat. + /// + /// + public sealed class TelemetryItemDto + { + /// + /// Item type — routes to the destination table. Values: Event, Signal, + /// DecisionTransition. Unknown values are rejected at parse time. + /// + public string Kind { get; set; } = string.Empty; + + /// Azure-Table PartitionKey, typically {tenantId}_{sessionId}. + public string PartitionKey { get; set; } = string.Empty; + + public string RowKey { get; set; } = string.Empty; + + /// Transport-cursor — monotonic per session across all item kinds. + public long TelemetryItemId { get; set; } + + /// Null for agent-global items (no session). + public long? SessionTraceOrdinal { get; set; } + + /// Already-serialised JSON of the wrapped payload (EnrollmentEvent / DecisionSignal / DecisionTransition). + public string PayloadJson { get; set; } = string.Empty; + + public bool RequiresImmediateFlush { get; set; } + + public DateTime EnqueuedAtUtc { get; set; } + + public int RetryCount { get; set; } + } +}