AutopilotMonitor/src/Backend/AutopilotMonitor.Functions/Functions/Ingest/IngestTelemetryFunction.cs

using System.IO;
using System.Net;
using Azure;
using AutopilotMonitor.Functions.DataAccess.TableStorage;
using AutopilotMonitor.Functions.Helpers;
using AutopilotMonitor.Functions.Security;
using AutopilotMonitor.Functions.Services;
using AutopilotMonitor.Functions.Services.Deletion;
using AutopilotMonitor.Shared.DataAccess;
using AutopilotMonitor.Shared.Models;
using Microsoft.Azure.Functions.Worker;
using Microsoft.Azure.Functions.Worker.Extensions.SignalRService;
using Microsoft.Azure.Functions.Worker.Http;
using Microsoft.Extensions.Logging;
using Newtonsoft.Json;

namespace AutopilotMonitor.Functions.Functions.Ingest
{
    /// <summary>
    /// Agent ingest endpoint. Consumes a heterogeneous batch of <see cref="TelemetryItemDto"/>s
    /// (Events + Signals + DecisionTransitions in a single JSON array, gzip-compressed on
    /// the wire — the <c>UseRequestDecompression</c> middleware decompresses before we
    /// parse). Plan §2.7a / §M5 / M4.6.ε.
    /// <para>
    /// <b>Routing by <see cref="TelemetryItemDto.Kind"/>:</b>
    /// <list type="bullet">
    ///   <item><c>Event</c> → <see cref="EventIngestProcessor"/> (rule engine, app-install
    ///   aggregation, SignalR, vulnerability correlation, webhooks, SLA breach, AdminAction
    ///   detection, ServerAction delivery)</item>
    ///   <item><c>Signal</c> → <see cref="ISignalRepository.StoreBatchAsync"/></item>
    ///   <item><c>DecisionTransition</c> → <see cref="IDecisionTransitionRepository.StoreBatchAsync"/></item>
    /// </list>
    /// </para>
    /// <para>
    /// <b>Response (M4.6.ε):</b> the agent parses <c>DeviceBlocked</c>/<c>UnblockAt</c>/
    /// <c>DeviceKillSignal</c>/<c>AdminAction</c>/<c>Actions</c> from the 2xx body and routes
    /// kill-switches through its <c>ServerActionDispatcher</c>.
    /// </para>
    /// </summary>
    public sealed class IngestTelemetryFunction
    {
        private readonly ILogger<IngestTelemetryFunction> _logger;
        private readonly ISessionRepository _sessionRepo;
        private readonly ISignalRepository _signalRepo;
        private readonly IDecisionTransitionRepository _transitionRepo;
        private readonly EventIngestProcessor _eventProcessor;
        private readonly TenantConfigurationService _configService;
        private readonly AdminConfigurationService _adminConfigService;
        private readonly RateLimitService _rateLimitService;
        private readonly AutopilotDeviceValidator _autopilotDeviceValidator;
        private readonly CorporateIdentifierValidator _corporateIdentifierValidator;
        private readonly DeviceAssociationValidator _deviceAssociationValidator;
        private readonly CloudPcDeviceValidator _cloudPcDeviceValidator;
        private readonly IntuneDeviceBindingValidator _intuneDeviceBindingValidator;
        private readonly BootstrapSessionService _bootstrapSessionService;
        private readonly KillSwitchEvaluator _killSwitchEvaluator;
        private readonly SessionDeletionGuard _deletionGuard;
        private readonly SessionOwnerBindingObserver _ownerBinding;

        public IngestTelemetryFunction(
            ILogger<IngestTelemetryFunction> logger,
            ISessionRepository sessionRepo,
            ISignalRepository signalRepo,
            IDecisionTransitionRepository transitionRepo,
            EventIngestProcessor eventProcessor,
            TenantConfigurationService configService,
            AdminConfigurationService adminConfigService,
            RateLimitService rateLimitService,
            AutopilotDeviceValidator autopilotDeviceValidator,
            CorporateIdentifierValidator corporateIdentifierValidator,
            DeviceAssociationValidator deviceAssociationValidator,
            CloudPcDeviceValidator cloudPcDeviceValidator,
            IntuneDeviceBindingValidator intuneDeviceBindingValidator,
            BootstrapSessionService bootstrapSessionService,
            KillSwitchEvaluator killSwitchEvaluator,
            SessionDeletionGuard deletionGuard,
            SessionOwnerBindingObserver ownerBinding)
        {
            _logger = logger;
            _sessionRepo = sessionRepo;
            _signalRepo = signalRepo;
            _transitionRepo = transitionRepo;
            _eventProcessor = eventProcessor;
            _configService = configService;
            _adminConfigService = adminConfigService;
            _rateLimitService = rateLimitService;
            _autopilotDeviceValidator = autopilotDeviceValidator;
            _corporateIdentifierValidator = corporateIdentifierValidator;
            _deviceAssociationValidator = deviceAssociationValidator;
            _cloudPcDeviceValidator = cloudPcDeviceValidator;
            _intuneDeviceBindingValidator = intuneDeviceBindingValidator;
            _bootstrapSessionService = bootstrapSessionService;
            _killSwitchEvaluator = killSwitchEvaluator;
            _deletionGuard = deletionGuard;
            _ownerBinding = ownerBinding;
        }

        [Function("IngestTelemetry")]
        public async Task<IngestEventsOutput> 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 AsOutput(await WriteErrorAsync(req, HttpStatusCode.BadRequest, "X-Tenant-Id header is required"));
                }

                var (validation, errorResponse) = await req.ValidateSecurityAsync(
                    tenantIdHeader,
                    _configService,
                    _adminConfigService,
                    _rateLimitService,
                    _autopilotDeviceValidator,
                    _corporateIdentifierValidator,
                    _logger,
                    bootstrapSessionService: _bootstrapSessionService,
                    deviceAssociationValidator: _deviceAssociationValidator,
                    cloudPcDeviceValidator: _cloudPcDeviceValidator,
                    intuneDeviceBindingValidator: _intuneDeviceBindingValidator);

                if (errorResponse != null) return AsOutput(errorResponse);

                // Device + version kill-switches: short-circuit before parsing the body. Serial
                // leg (caller-declared header) AND certificate-identity leg (cert Subject CN — the
                // one an agent that omits or forges its serial cannot dodge). A session-scoped
                // block (watchdog auto-block) cannot be decided yet — the session id lives in the
                // body — so it is re-evaluated below once parsed. Evaluation + kill ops-event live
                // in KillSwitchEvaluator, shared with the agent-config channel.
                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 preBodyVerdict = await _killSwitchEvaluator.EvaluateAsync(
                    tenantIdHeader, serialNumberHeader, agentVersionHeader, channel: "telemetry",
                    intuneDeviceId: validation.IntuneDeviceId);
                DeviceIdentityBinding.Stamp(req, preBodyVerdict.IdentityBinding);
                if (preBodyVerdict.IsBlocked && !preBodyVerdict.IsSessionScoped)
                    return AsOutput(await WriteDeviceBlockedAsync(req, preBodyVerdict));

                // Defense in depth: cap decompressed body size before buffering it all into
                // memory + deserialising (same tenant-config knob the removed V1 NDJSON path
                // used). Request body here is already gzip-decompressed by
                // UseRequestDecompression — so the cap is on the actual JSON bytes we'll feed
                // to the parser.
                var tenantConfig = await _configService.GetConfigurationAsync(tenantIdHeader);
                var maxPayloadBytes = (tenantConfig?.MaxNdjsonPayloadSizeMB ?? 5) * 1024 * 1024;

                bool exceeded;
                List<TelemetryItemDto>? items;
                try
                {
                    (exceeded, items) = await ReadBodyWithSizeCapAsync(req.Body, maxPayloadBytes);
                }
                catch (JsonException ex)
                {
                    _logger.LogWarning(ex, "IngestTelemetry: malformed JSON body");
                    return AsOutput(await WriteErrorAsync(req, HttpStatusCode.BadRequest, "Malformed JSON body"));
                }

                if (exceeded)
                {
                    _logger.LogWarning(
                        "IngestTelemetry: payload exceeds {Max}MB cap for tenant {Tenant}",
                        maxPayloadBytes / (1024 * 1024), tenantIdHeader);
                    return AsOutput(await WriteErrorAsync(
                        req, HttpStatusCode.RequestEntityTooLarge,
                        $"Payload exceeds {maxPayloadBytes / (1024 * 1024)} MB cap"));
                }

                if (items == null || items.Count == 0)
                {
                    return AsOutput(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 must belong to one
                // session — anything else is a client bug and gets the whole batch rejected (see
                // the uniformity check below).
                if (!TryParsePartitionKey(items[0].PartitionKey, out var bodyTenantId, out var sessionId))
                {
                    return AsOutput(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 AsOutput(await WriteErrorAsync(req, HttpStatusCode.Forbidden, "TenantId mismatch between header and payload"));
                }

                // Defense-in-depth: every item must carry the same PartitionKey as the first. Without
                // this check a mixed-session batch would stamp every item with session A's identity
                // inside PersistItemsAsync, silently relocating session B's signals/transitions into
                // session A's primary rows.
                if (FindMismatchingPartitionKey(items, out var mismatchIndex, out var mismatchedValue))
                {
                    _logger.LogWarning(
                        "IngestTelemetry: PartitionKey mismatch at item[{Index}] — expected={Expected}, got={Got}",
                        mismatchIndex, items[0].PartitionKey, mismatchedValue);
                    return AsOutput(await WriteErrorAsync(
                        req, HttpStatusCode.BadRequest,
                        "All telemetry items in a batch must share the same PartitionKey"));
                }

                // Session-scoped block, session now known: either this is one of the blocked
                // sessions (stop) or a new enrollment on the same device (auto-unblock — the
                // watchdog blocked a runaway session, not the device). The version leg runs
                // again on purpose: the device leg short-circuited it before the body was parsed.
                if (preBodyVerdict.IsSessionScoped)
                {
                    var sessionVerdict = await _killSwitchEvaluator.EvaluateAsync(
                        bodyTenantId, serialNumberHeader, agentVersionHeader, channel: "telemetry",
                        intuneDeviceId: validation.IntuneDeviceId, sessionId: sessionId);
                    if (sessionVerdict.IsBlocked)
                        return AsOutput(await WriteDeviceBlockedAsync(req, sessionVerdict));
                }

                // Cascade-delete guard: refuse the batch with 410 Gone when a V2 cascade owns the
                // Sessions row (states Preparing/Queued/Running/Poisoned). Without this check, the
                // hot-path writers below (StoreEventsBatchAsync, signal/transition StoreBatchAsync,
                // UpdateSessionImeAgentVersionAsync upsert, …) would land rows past the lock and
                // leave orphan data the manifest cannot describe. One read per batch; absent
                // Sessions row → silent pass (caller handles session-not-found in its own write).
                Azure.Data.Tables.TableEntity? guardSessionRow;
                try
                {
                    guardSessionRow = await _deletionGuard.EnsureWritableAndGetRowAsync(bodyTenantId, sessionId, "V2.IngestTelemetry");
                }
                catch (SessionDeletionLockedException locked)
                {
                    _logger.LogInformation(
                        "IngestTelemetry: refused batch — cascade in flight tenant={Tenant} session={Session} state={State} manifestId={ManifestId}",
                        bodyTenantId, sessionId, locked.CurrentState, locked.ManifestId);
                    return AsOutput(await WriteSessionLockedAsync(req, locked));
                }

                // The header serial is caller-declared; the Sessions row carries the serial the
                // session was registered under. When they differ, the ROW's serial is held against
                // the block list too — a mid-session header change must not dodge a block placed
                // on the registered serial. Cached like every other lookup (negative entries for
                // healthy devices), so the common case costs nothing.
                var rowSerial = guardSessionRow?.GetString("SerialNumber");
                if (!string.IsNullOrWhiteSpace(rowSerial)
                    && !string.Equals(rowSerial.Trim(), serialNumberHeader?.Trim(), StringComparison.OrdinalIgnoreCase))
                {
                    var rowSerialVerdict = await _killSwitchEvaluator.EvaluateAsync(
                        bodyTenantId, rowSerial, agentVersion: null, channel: "telemetry", sessionId: sessionId);
                    if (rowSerialVerdict.IsBlocked)
                        return AsOutput(await WriteDeviceBlockedAsync(req, rowSerialVerdict));
                }

                // Reuse the guard's full-row read: its Status feeds the stall-heal check inside
                // EventIngestProcessor, saving one Sessions point-read per batch.
                var preFetchedStatus = TryReadSessionStatus(guardSessionRow);

                // SESSION-OWNER-BINDING-SHADOW: is the device behind this certificate/token the
                // device this session belongs to? Same guard row, zero extra reads. Stage 1 records
                // the outcome (request-row dimension + Warning + throttled ops event) and stamps
                // legacy claims / rebinds; it never refuses the batch. Covers the Signal-only path
                // too — that is where pending ServerActions get fetched-and-cleared.
                var ownerDecision = _ownerBinding.Observe(req, bodyTenantId, sessionId, guardSessionRow, validation, "agent/telemetry");
                if (guardSessionRow != null)
                    await _ownerBinding.StampAsync(bodyTenantId, sessionId, ownerDecision);

                // P14: device-clock send time of this upload (one value per request). Separates
                // spool delay from device-vs-server clock offset downstream; absent/garbage → null.
                var sentAtHeader = req.Headers.Contains("X-Send-Time-Utc")
                    ? req.Headers.GetValues("X-Send-Time-Utc").FirstOrDefault()
                    : null;
                var sentAt = ParseSendTimeHeader(sentAtHeader);

                // Partition + persist. Events are routed through EventIngestProcessor which runs the
                // full pipeline (rule engine / app-install aggregation / SignalR / webhooks / ...);
                // Signal + Transition go straight to their repositories.
                var outcome = await PersistItemsAsync(items, bodyTenantId, sessionId, validation, preFetchedStatus, sentAt);

                _logger.LogInformation(
                    "IngestTelemetry: tenant={Tenant} session={Session} events={E} signals={S} transitions={T} unknown={U}",
                    bodyTenantId, sessionId, outcome.EventCount, outcome.SignalCount, outcome.TransitionCount, outcome.UnknownCount);

                var response = req.CreateResponse(HttpStatusCode.OK);
                await response.WriteAsJsonAsync(new IngestEventsResponse
                {
                    Success = true,
                    EventsReceived = items.Count,
                    EventsProcessed = outcome.EventCount + outcome.SignalCount + outcome.TransitionCount,
                    Message = $"Stored {outcome.EventCount} events, {outcome.SignalCount} signals, {outcome.TransitionCount} transitions",
                    ProcessedAt = DateTime.UtcNow,
                    AdminAction = outcome.AdminAction,
                    Actions = outcome.PendingActions,
                });

                return new IngestEventsOutput
                {
                    HttpResponse = response,
                    SignalRMessages = outcome.SignalRMessages,
                };
            }
            catch (TableEntityTooLargeException ex)
            {
                // One item can never fit a table row. 413 makes the agent halve the batch and,
                // once the item is alone, quarantine it — a 500 would be replayed forever.
                _logger.LogWarning(ex, "IngestTelemetry: telemetry item {Partition}/{Row} exceeds the entity limit ({Bytes} bytes)",
                    ex.PartitionKey, ex.RowKey, ex.EstimatedBytes);
                return AsOutput(await WriteErrorAsync(req, HttpStatusCode.RequestEntityTooLarge,
                    "A telemetry item exceeds the storage entity limit"));
            }
            catch (RequestFailedException ex)
            {
                var (status, message, retryAfter) = ClassifyStorageFailure(ex);
                if (status == HttpStatusCode.InternalServerError)
                    _logger.LogError(ex, "IngestTelemetry: storage request failed with status {Status} ({Code})", ex.Status, ex.ErrorCode);
                else
                    _logger.LogWarning(ex, "IngestTelemetry: storage request failed with status {Status} ({Code}) — answering {Http}", ex.Status, ex.ErrorCode, (int)status);
                return AsOutput(await WriteErrorAsync(req, status, message, retryAfter));
            }
            catch (Exception ex)
            {
                _logger.LogError(ex, "IngestTelemetry: unhandled exception");
                return AsOutput(await WriteErrorAsync(req, HttpStatusCode.InternalServerError, "Internal server error"));
            }
        }

        /// <summary>Seconds the agent should wait before replaying a batch after a storage outage.</summary>
        internal const int StorageRetryAfterSeconds = 5;

        /// <summary>
        /// Maps a storage failure onto the status the agent's uploader understands: 413 makes it
        /// halve the batch, 503 (with <c>Retry-After</c>) makes it replay later; the agent treats
        /// 500 as transient too but never shrinks on it, so an oversize batch surfacing as 500
        /// would wedge the device (audit 2026-09-05 F07).
        /// </summary>
        internal static (HttpStatusCode Status, string Message, int? RetryAfterSeconds) ClassifyStorageFailure(RequestFailedException ex)
        {
            if (StorageErrors.IsPayloadTooLarge(ex))
                return (HttpStatusCode.RequestEntityTooLarge, "Telemetry batch exceeds the storage transaction limit; retry with a smaller batch", null);
            if (StorageErrors.IsTransient(ex))
                return (HttpStatusCode.ServiceUnavailable, "Storage temporarily unavailable; retry later", StorageRetryAfterSeconds);
            return (HttpStatusCode.InternalServerError, "Internal server error", null);
        }

        private static IngestEventsOutput AsOutput(HttpResponseData response)
            => new IngestEventsOutput
            {
                HttpResponse = response,
                SignalRMessages = Array.Empty<SignalRMessageAction>(),
            };

        /// <summary>
        /// 200 with <c>DeviceBlocked=true</c> (and <c>DeviceKillSignal=true</c> for a kill): the
        /// agent pauses its upload loop until <c>UnblockAt</c>, or self-destructs on a kill.
        /// </summary>
        private static async Task<HttpResponseData> WriteDeviceBlockedAsync(HttpRequestData req, KillSwitchVerdict verdict)
        {
            var response = req.CreateResponse(HttpStatusCode.OK);
            await response.WriteAsJsonAsync(new IngestEventsResponse
            {
                Success = false,
                DeviceBlocked = true,
                DeviceKillSignal = verdict.IsKill,
                UnblockAt = verdict.UnblockAt,
                Message = verdict.Message,
                ProcessedAt = DateTime.UtcNow,
            });
            return response;
        }

        /// <summary>
        /// Responds 410 Gone when the V2 cascade-delete guard refuses the batch. Per plan §5 PR3
        /// wiring table, telemetry ingest of a locked session is a terminal condition for the
        /// agent: the session is being torn down server-side and any further writes would create
        /// orphan rows. 410 is the documented status; the body shape mirrors the device-blocked
        /// response so existing agent code paths (which already short-circuit on Success=false)
        /// drop the batch without retry.
        /// </summary>
        private static async Task<HttpResponseData> WriteSessionLockedAsync(HttpRequestData req, SessionDeletionLockedException locked)
        {
            var response = req.CreateResponse(HttpStatusCode.Gone);
            await response.WriteAsJsonAsync(new IngestEventsResponse
            {
                Success = false,
                Message = $"Session is being deleted by an administrator (state={locked.CurrentState}); further telemetry will be rejected.",
                ProcessedAt = DateTime.UtcNow,
            });
            return response;
        }

        /// <summary>
        /// Extracts <c>Status</c> from a Sessions row the deletion guard already loaded. Sessions
        /// writes Status as a STRING (<c>status.ToString()</c> in UpdateSessionStatusAsync — never
        /// an int), so this mirrors the canonical mapper's parse: <c>Enum.TryParse</c>,
        /// case-insensitive. Returns null for missing/unparseable values (incl. a defensive int
        /// fallback for any legacy numeric shape) — callers then fall back to their own read.
        /// </summary>
        internal static SessionStatus? TryReadSessionStatus(Azure.Data.Tables.TableEntity? sessionRow)
        {
            if (sessionRow == null || !sessionRow.TryGetValue("Status", out var statusValue))
                return null;

            return statusValue switch
            {
                string s when Enum.TryParse<SessionStatus>(s, ignoreCase: true, out var parsed) => parsed,
                int i when Enum.IsDefined(typeof(SessionStatus), i) => (SessionStatus)i,
                _ => null,
            };
        }

        /// <summary>
        /// Partitions the incoming batch by <see cref="TelemetryItemDto.Kind"/> and persists each
        /// kind through its destination path. Events go through <see cref="EventIngestProcessor"/>
        /// (the full event pipeline); Signals + Transitions land directly in their primary tables.
        /// The returned <see cref="IngestOutcome"/> carries both the per-kind counts and the
        /// control-signal / SignalR payload for the response.
        /// </summary>
        private async Task<IngestOutcome> PersistItemsAsync(
            IReadOnlyList<TelemetryItemDto> items,
            string tenantId,
            string sessionId,
            SecurityValidationResult validation,
            SessionStatus? preFetchedStatus,
            DateTime? sentAt)
        {
            var events      = new List<EnrollmentEvent>();
            var signals     = new List<SignalRecord>();
            var transitions = new List<DecisionTransitionRecord>();
            var unknown     = 0;

            // Parse-null tracking: the per-line skip is deliberate (one poisonous item must not
            // reject the batch), but the response stays Success=true and the agent deletes its
            // spool — so a dropped item is gone forever. Without these counters an agent-side
            // serialization regression would be fleet-wide silent data loss.
            int eventParseFailed = 0, signalParseFailed = 0, transitionParseFailed = 0;
            long? firstParseFailedItemId = null;

            void TrackParseFailure(ref int counter, TelemetryItemDto item)
            {
                counter++;
                firstParseFailedItemId ??= item.TelemetryItemId;
            }

            foreach (var item in items)
            {
                switch (item.Kind)
                {
                    case "Event":
                        var evt = TelemetryPayloadParser.ParseEvent(item, tenantId, sessionId);
                        if (evt != null) events.Add(evt);
                        else TrackParseFailure(ref eventParseFailed, item);
                        break;
                    case "Signal":
                        var sig = TelemetryPayloadParser.ParseSignal(item, tenantId, sessionId);
                        if (sig != null) signals.Add(sig);
                        else TrackParseFailure(ref signalParseFailed, item);
                        break;
                    case "DecisionTransition":
                        var tr = TelemetryPayloadParser.ParseTransition(item, tenantId, sessionId);
                        if (tr != null) transitions.Add(tr);
                        else TrackParseFailure(ref transitionParseFailed, item);
                        break;
                    default:
                        unknown++;
                        _logger.LogWarning("IngestTelemetry: unknown Kind '{Kind}' (TelemetryItemId={Id})", item.Kind, item.TelemetryItemId);
                        break;
                }
            }

            var parseFailed = eventParseFailed + signalParseFailed + transitionParseFailed;
            if (parseFailed > 0)
            {
                _logger.LogWarning(
                    "IngestTelemetry: dropped {Total} unparseable item(s) of {BatchSize} for session {SessionId} " +
                    "(events={Events}, signals={Signals}, transitions={Transitions}, firstItemId={FirstItemId}) — " +
                    "items are lost (agent clears its spool on success)",
                    parseFailed, items.Count, sessionId,
                    eventParseFailed, signalParseFailed, transitionParseFailed, firstParseFailedItemId);
            }

            // Signals + Transitions write directly; they don't feed into the event pipeline.
            var signalCount     = await _signalRepo.StoreBatchAsync(signals);
            var transitionCount = await _transitionRepo.StoreBatchAsync(transitions);

            int eventCount;
            string? adminAction;
            List<ServerAction>? pendingActions;
            SignalRMessageAction[] signalRMessages;

            if (events.Count > 0)
            {
                // Full event pipeline (rule engine, app-install aggregation, SignalR,
                // vulnerability correlation, webhooks, SLA breach, AdminAction detection,
                // ServerAction delivery).
                var eventRequest = new IngestEventsRequest
                {
                    SessionId = sessionId,
                    TenantId  = tenantId,
                    Events    = events,
                    SentAt    = sentAt,
                };
                var processed = await _eventProcessor.ProcessEventsAsync(eventRequest, validation, preFetchedStatus);

                eventCount      = processed.EventsProcessed;
                adminAction     = processed.AdminAction;
                pendingActions  = processed.PendingActions;
                signalRMessages = processed.SignalRMessages;
            }
            else
            {
                // Signal/Transition-only batch: no events means no event pipeline. We still honour
                // the control-signal contract (AdminAction + pending ServerActions) because the
                // agent reads them from every 2xx response regardless of which items it sent.
                eventCount = 0;
                var (aa, pa) = await ReadControlSignalsAsync(tenantId, sessionId);
                adminAction     = aa;
                pendingActions  = pa;
                signalRMessages = Array.Empty<SignalRMessageAction>();
            }

            return new IngestOutcome
            {
                EventCount      = eventCount,
                SignalCount     = signalCount,
                TransitionCount = transitionCount,
                UnknownCount    = unknown,
                AdminAction     = adminAction,
                PendingActions  = pendingActions,
                SignalRMessages = signalRMessages,
            };
        }

        /// <summary>
        /// Fetches <c>AdminAction</c> (session marked terminal out-of-band) and pending
        /// <c>ServerAction</c>s for Signal/Transition-only batches (no <see cref="EventIngestProcessor"/>
        /// invocation).
        /// </summary>
        private async Task<(string? adminAction, List<ServerAction>? pendingActions)>
            ReadControlSignalsAsync(string tenantId, string sessionId)
        {
            var session = await _sessionRepo.GetSessionAsync(tenantId, sessionId);
            if (session == null) return (null, null);

            // AdminAction carries the portal-button signal only. The old logic (Status ==
            // Succeeded/Failed → AdminAction) also fired for agent-reported completion and made
            // every follow-up signal/transition upload look like an out-of-band admin override.
            string? adminAction = session.AdminMarkedAction;

            List<ServerAction>? pendingActions = null;
            if (!string.IsNullOrEmpty(session.PendingActionsJson))
            {
                var fetched = await _sessionRepo.FetchAndClearPendingActionsAsync(tenantId, sessionId);
                if (fetched.Count > 0) pendingActions = fetched;
            }

            return (adminAction, pendingActions);
        }

        private sealed class IngestOutcome
        {
            public int EventCount;
            public int SignalCount;
            public int TransitionCount;
            public int UnknownCount;
            public string? AdminAction;
            public List<ServerAction>? PendingActions;
            public SignalRMessageAction[] SignalRMessages = Array.Empty<SignalRMessageAction>();
        }

        /// <summary>
        /// Buffers the request body under a strict byte cap, then deserialises the telemetry batch
        /// directly off the buffered bytes. Returns <c>(true, null)</c> as soon as the cap would be
        /// exceeded — the stream isn't fully drained, which bounds memory use even for a malicious
        /// sender (strict greater-than, so a payload equal to the cap is accepted).
        /// <para>
        /// Hot-path note: this deserialises through a <see cref="JsonTextReader"/> over the buffered
        /// <see cref="MemoryStream"/> rather than materialising a full UTF-16 <c>string</c> + a
        /// <c>ToArray()</c> copy first. That removes ~3× the body size in transient allocations per
        /// request on the highest-volume backend endpoint — the reader decodes UTF-8 incrementally
        /// in 8&#160;KB chunks. The cap is still applied <i>before</i> parsing, so an over-cap body
        /// never reaches the deserialiser.
        /// </para>
        /// Throws <see cref="JsonException"/> on malformed JSON (caller maps to 400). The
        /// deserialiser uses the same default settings as <c>JsonConvert.DeserializeObject</c>, so
        /// the parse semantics are unchanged — only the intermediate allocations are gone.
        /// </summary>
        internal static async Task<(bool exceeded, List<TelemetryItemDto>? items)> ReadBodyWithSizeCapAsync(
            Stream source, int maxBytes)
        {
            using var buffered = new MemoryStream();
            var buffer = new byte[8192];
            int read;
            long total = 0;
            while ((read = await source.ReadAsync(buffer, 0, buffer.Length)) > 0)
            {
                total += read;
                if (total > maxBytes) return (true, null);
                await buffered.WriteAsync(buffer, 0, read);
            }

            buffered.Position = 0;
            using var textReader = new StreamReader(buffered, System.Text.Encoding.UTF8);
            using var jsonReader = new JsonTextReader(textReader);
            var items = JsonSerializer.CreateDefault().Deserialize<List<TelemetryItemDto>>(jsonReader);
            return (false, items);
        }

        /// <summary>
        /// Scans the batch for any item whose <see cref="TelemetryItemDto.PartitionKey"/> differs
        /// from the first item's. Returns true + the offending index/value on mismatch; false if
        /// every item shares the same PartitionKey (or the batch has &lt; 2 items). Pure so the
        /// guard can be unit-tested without a live HTTP trigger.
        /// </summary>
        internal static bool FindMismatchingPartitionKey(
            IReadOnlyList<TelemetryItemDto> items,
            out int mismatchIndex,
            out string? mismatchedValue)
        {
            mismatchIndex = -1;
            mismatchedValue = null;
            if (items is null || items.Count < 2) return false;

            var expected = items[0].PartitionKey;
            for (var i = 1; i < items.Count; i++)
            {
                if (!string.Equals(items[i].PartitionKey, expected, StringComparison.Ordinal))
                {
                    mismatchIndex = i;
                    mismatchedValue = items[i].PartitionKey;
                    return true;
                }
            }
            return false;
        }

        /// <summary>
        /// Parses the <c>X-Send-Time-Utc</c> header (agent device-clock send time, ISO-8601
        /// round-trip). Returns null for absent/unparseable values and for anything before
        /// year 2000 — Table Storage rejects pre-1601 DateTimes outright, and a sub-2000
        /// value can only be garbage, never a measurable clock offset. Deliberately NO upper
        /// clamp: a future-dated send time IS the device-clock-error measurement this field
        /// exists to capture.
        /// </summary>
        internal static DateTime? ParseSendTimeHeader(string? headerValue)
        {
            if (string.IsNullOrWhiteSpace(headerValue)) return null;

            if (!DateTimeOffset.TryParse(
                    headerValue,
                    System.Globalization.CultureInfo.InvariantCulture,
                    System.Globalization.DateTimeStyles.RoundtripKind,
                    out var parsed))
                return null;

            var utc = parsed.UtcDateTime;
            return utc.Year >= 2000 ? utc : (DateTime?)null;
        }

        /// <summary>
        /// PartitionKey convention is <c>{tenantId}_{sessionId}</c> — 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).
        /// </summary>
        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<HttpResponseData> WriteErrorAsync(
            HttpRequestData req, HttpStatusCode status, string message, int? retryAfterSeconds = null)
        {
            var response = req.CreateResponse(status);
            if (retryAfterSeconds is int seconds)
                response.Headers.Add("Retry-After", seconds.ToString());
            await response.WriteAsJsonAsync(new IngestEventsResponse
            {
                Success = false,
                EventsReceived = 0,
                EventsProcessed = 0,
                Message = message,
                ProcessedAt = DateTime.UtcNow,
            });
            return response;
        }
    }

    /// <summary>
    /// Multi-binding output shape of the agent ingest endpoint: the HTTP response plus the
    /// SignalR messages for the live UI push.
    /// </summary>
    public class IngestEventsOutput
    {
        [HttpResult]
        public HttpResponseData? HttpResponse { get; set; }

        [SignalROutput(HubName = SignalRGroupHelper.HubName)]
        public SignalRMessageAction[]? SignalRMessages { get; set; }
    }
}