From 01e29da0f2e5e3bb3baebf56f28bfc2c84b32d72 Mon Sep 17 00:00:00 2001 From: Oliver Kieselbach Date: Tue, 14 Apr 2026 23:42:10 +0200 Subject: [PATCH] feat: Server SessionActions and action signal consolidation on agent side --- .../Core/EventUploadOrchestratorTests.cs | 74 ++++-- .../Core/EventUploadOrchestrator.cs | 141 ++++++----- .../Monitoring/Core/MonitoringService.cs | 194 +++++++++++++++ .../Monitoring/Core/ServerActionDispatcher.cs | 223 ++++++++++++++++++ .../AutopilotMonitor.Agent/buildcounter.txt | 2 +- .../ServerActionQueueTests.cs | 77 ++++++ .../TableStorage/TableSessionRepository.cs | 6 + .../Ingest/IngestEventsFunction.Processing.cs | 30 ++- .../Functions/Ingest/IngestEventsFunction.cs | 6 +- .../Sessions/QueueSessionActionFunction.cs | 134 +++++++++++ .../Security/EndpointAccessPolicyCatalog.cs | 1 + .../TableStorageService.ServerActions.cs | 153 ++++++++++++ .../Services/TableStorageService.Sessions.cs | 4 + .../DataAccess/ISessionRepository.cs | 14 ++ .../Models/ServerAction.cs | 70 ++++++ .../Models/SessionApiModels.cs | 22 ++ 16 files changed, 1080 insertions(+), 71 deletions(-) create mode 100644 src/Agent/AutopilotMonitor.Agent.Core/Monitoring/Core/ServerActionDispatcher.cs create mode 100644 src/Backend/AutopilotMonitor.Functions.Tests/ServerActionQueueTests.cs create mode 100644 src/Backend/AutopilotMonitor.Functions/Functions/Sessions/QueueSessionActionFunction.cs create mode 100644 src/Backend/AutopilotMonitor.Functions/Services/TableStorageService.ServerActions.cs create mode 100644 src/Shared/AutopilotMonitor.Shared/Models/ServerAction.cs diff --git a/src/Agent/AutopilotMonitor.Agent.Core.Tests/Core/EventUploadOrchestratorTests.cs b/src/Agent/AutopilotMonitor.Agent.Core.Tests/Core/EventUploadOrchestratorTests.cs index f5504c6c..7b04e3b1 100644 --- a/src/Agent/AutopilotMonitor.Agent.Core.Tests/Core/EventUploadOrchestratorTests.cs +++ b/src/Agent/AutopilotMonitor.Agent.Core.Tests/Core/EventUploadOrchestratorTests.cs @@ -78,6 +78,28 @@ namespace AutopilotMonitor.Agent.Core.Tests.Core return orch.Object; } + /// + /// Returns a Mock ServerActionDispatcher whose DispatchAsync captures the batch into the + /// supplied callback. Used by tests that verify the synthesis contract between the orchestrator + /// and the dispatcher (kill / admin-action → terminate_session ServerAction). + /// + private ServerActionDispatcher CreateCapturingDispatcher(Action> onDispatched) + { + var mock = new Mock( + _config, + TestLogger.Instance, + (Func>)(() => Task.FromResult(true)), + (Func>)(_ => Task.FromResult(new DiagnosticsUploadResult())), + (Func)(_ => Task.CompletedTask), + (Action)(_ => { })) + { CallBase = false }; + + mock.Setup(d => d.DispatchAsync(It.IsAny>())) + .Returns>(batch => { onDispatched(batch); return Task.CompletedTask; }); + + return mock.Object; + } + private List CreateTestEvents(int count = 3) { return Enumerable.Range(1, count) @@ -185,22 +207,30 @@ namespace AutopilotMonitor.Agent.Core.Tests.Core // ===== Admin Signal Processing ===== [Fact] - public async Task UploadEvents_KillSignal_SelfDestructsAndExits() + public async Task UploadEvents_KillSignal_SynthesizesTerminateActionWithForceSelfDestruct() { + // After the consolidation the orchestrator no longer runs kill-specific inline logic. + // It synthesizes a terminate_session ServerAction and hands it to the dispatcher. + // The actual self-destruct + exit happens in MonitoringService.HandleTerminateSessionAsync + // (see dedicated tests there). We verify the synthesis contract here. var events = CreateTestEvents(); _mockSpool.Setup(s => s.GetBatch(It.IsAny())).Returns(events); - _mockSpool.Setup(s => s.GetCount()).Returns(0); _mockApiClient.Setup(a => a.IngestEventsAsync(It.IsAny())) .ReturnsAsync(new IngestEventsResponse { DeviceKillSignal = true }); + List dispatched = null; + var dispatcher = CreateCapturingDispatcher(captured => dispatched = captured); + var orch = CreateOrchestrator(); + orch.SetServerActionDispatcher(dispatcher); await orch.UploadEventsAsync(); - Assert.Single(_shutdownEvents); - Assert.Equal("remote_kill_signal", _shutdownEvents[0].type); - Assert.Equal(true, _shutdownEvents[0].data["selfDestruct"]); - _mockCleanup.Verify(c => c.ExecuteSelfDestruct(), Times.Once); - Assert.Equal(0, _exitCode); + Assert.NotNull(dispatched); + Assert.Single(dispatched); + Assert.Equal(ServerActionTypes.TerminateSession, dispatched[0].Type); + Assert.Equal("true", dispatched[0].Params["forceSelfDestruct"]); + Assert.Equal("0", dispatched[0].Params["gracePeriodSeconds"]); + Assert.Equal("kill_signal", dispatched[0].Params["origin"]); } [Fact] @@ -220,24 +250,34 @@ namespace AutopilotMonitor.Agent.Core.Tests.Core } [Fact] - public async Task UploadEvents_AdminOverride_Succeeded_EmitsEnrollmentComplete() + public async Task UploadEvents_AdminOverride_Succeeded_SynthesizesTerminateWithOutcome() { + // AdminAction is a soft-shutdown path — no forceSelfDestruct. The adminOutcome param + // tells MonitoringService.HandleTerminateSessionAsync to emit enrollment_complete + // instead of agent_shutdown. var events = CreateTestEvents(); _mockSpool.Setup(s => s.GetBatch(It.IsAny())).Returns(events); _mockApiClient.Setup(a => a.IngestEventsAsync(It.IsAny())) .ReturnsAsync(new IngestEventsResponse { AdminAction = "Succeeded" }); _terminalEventSeen = false; + List dispatched = null; + var dispatcher = CreateCapturingDispatcher(captured => dispatched = captured); + var orch = CreateOrchestrator(); + orch.SetServerActionDispatcher(dispatcher); await orch.UploadEventsAsync(); - Assert.Single(_emittedEvents); - Assert.Equal("enrollment_complete", _emittedEvents[0].EventType); - Assert.Equal("AdminOverride", _emittedEvents[0].Source); + Assert.NotNull(dispatched); + Assert.Single(dispatched); + Assert.Equal(ServerActionTypes.TerminateSession, dispatched[0].Type); + Assert.Equal("Succeeded", dispatched[0].Params["adminOutcome"]); + Assert.Equal("admin_action", dispatched[0].Params["origin"]); + Assert.False(dispatched[0].Params.ContainsKey("forceSelfDestruct")); } [Fact] - public async Task UploadEvents_AdminOverride_Failed_EmitsEnrollmentFailed() + public async Task UploadEvents_AdminOverride_Failed_SynthesizesTerminateWithFailedOutcome() { var events = CreateTestEvents(); _mockSpool.Setup(s => s.GetBatch(It.IsAny())).Returns(events); @@ -245,11 +285,17 @@ namespace AutopilotMonitor.Agent.Core.Tests.Core .ReturnsAsync(new IngestEventsResponse { AdminAction = "Failed" }); _terminalEventSeen = false; + List dispatched = null; + var dispatcher = CreateCapturingDispatcher(captured => dispatched = captured); + var orch = CreateOrchestrator(); + orch.SetServerActionDispatcher(dispatcher); await orch.UploadEventsAsync(); - Assert.Single(_emittedEvents); - Assert.Equal("enrollment_failed", _emittedEvents[0].EventType); + Assert.NotNull(dispatched); + Assert.Single(dispatched); + Assert.Equal(ServerActionTypes.TerminateSession, dispatched[0].Type); + Assert.Equal("Failed", dispatched[0].Params["adminOutcome"]); } [Fact] diff --git a/src/Agent/AutopilotMonitor.Agent.Core/Monitoring/Core/EventUploadOrchestrator.cs b/src/Agent/AutopilotMonitor.Agent.Core/Monitoring/Core/EventUploadOrchestrator.cs index a661fcd6..a1f166c2 100644 --- a/src/Agent/AutopilotMonitor.Agent.Core/Monitoring/Core/EventUploadOrchestrator.cs +++ b/src/Agent/AutopilotMonitor.Agent.Core/Monitoring/Core/EventUploadOrchestrator.cs @@ -26,6 +26,7 @@ namespace AutopilotMonitor.Agent.Core.Monitoring.Core private readonly Action _emitEvent; private readonly Action> _emitShutdownEvent; private readonly Func _isTerminalEventSeen; + private ServerActionDispatcher _serverActionDispatcher; private readonly Timer _uploadTimer; private readonly Timer _debounceTimer; @@ -86,6 +87,16 @@ namespace AutopilotMonitor.Agent.Core.Monitoring.Core _logger.Info("EventUploadOrchestrator started (debounce + fallback timers)"); } + /// + /// Wires the server-action dispatcher. Called by after the + /// orchestrator is constructed, once all other services exist. Optional — if null, actions + /// in ingest responses are simply ignored. + /// + public void SetServerActionDispatcher(ServerActionDispatcher dispatcher) + { + _serverActionDispatcher = dispatcher; + } + /// /// Stops all upload timers. Called during terminal event processing to prevent /// further automatic uploads while the completion handler drives uploads manually. @@ -151,44 +162,9 @@ namespace AutopilotMonitor.Agent.Core.Monitoring.Core var response = await _apiClient.IngestEventsAsync(request); - if (response.DeviceKillSignal) - { - _logger.Warning($"=== REMOTE KILL SIGNAL received from administrator. Spool: {_spool.GetCount()} events pending. Initiating self-destruct... ==="); - - StopTimers(); - - _emitShutdownEvent( - "remote_kill_signal", - "Agent terminated by remote kill signal from administrator", - new Dictionary - { - { "pendingEvents", _spool.GetCount() }, - { "selfDestruct", true } - }); - try - { - var shutdownEvents = _spool.GetBatch(_configuration.MaxBatchSize); - if (shutdownEvents.Count > 0) - { - await _apiClient.IngestEventsAsync(new IngestEventsRequest - { - SessionId = _configuration.SessionId, - TenantId = _configuration.TenantId, - Events = shutdownEvents - }); - } - } - catch (Exception ex) - { - _logger.Warning($"Failed to upload shutdown event before kill: {ex.Message}"); - } - - _configuration.SelfDestructOnComplete = true; - _cleanupService.ExecuteSelfDestruct(); - ExitProcess(0); - return; - } - + // DeviceBlocked is intentionally NOT synthesized into a ServerAction: it's a + // non-terminal quarantine (UnblockAt can auto-resume the session later). Treat it + // as a straight "stop uploads for now" signal. if (response.DeviceBlocked) { var unblockMsg = response.UnblockAt.HasValue @@ -200,26 +176,83 @@ namespace AutopilotMonitor.Agent.Core.Monitoring.Core return; } - if (!string.IsNullOrEmpty(response.AdminAction) && !_isTerminalEventSeen()) - { - var succeeded = string.Equals(response.AdminAction, "Succeeded", StringComparison.OrdinalIgnoreCase); - _logger.Warning($"=== ADMIN OVERRIDE: Session marked as {response.AdminAction} by administrator. Initiating cleanup... ==="); + // Synthesize legacy kill + admin-override signals into ServerActions so the whole + // "server told us to stop" surface runs through a single shutdown handler. This + // removes three divergent code paths (kill / admin / terminate) and guarantees + // every terminal path respects SelfDestructOnComplete — no more zombie agents + // after the operator clicks Mark-Failed in the portal. + List synthesized = null; - _emitEvent(new EnrollmentEvent + if (response.DeviceKillSignal) + { + synthesized = new List { - SessionId = _configuration.SessionId, - TenantId = _configuration.TenantId, - EventType = succeeded ? "enrollment_complete" : "enrollment_failed", - Severity = succeeded ? EventSeverity.Info : EventSeverity.Warning, - Source = "AdminOverride", - Phase = EnrollmentPhase.Complete, - Message = $"Session {response.AdminAction.ToLower()} by administrator — cleanup initiated", - Timestamp = DateTime.UtcNow, - Data = new Dictionary + new ServerAction { - { "adminAction", response.AdminAction } + Type = ServerActionTypes.TerminateSession, + Reason = "DeviceKillSignal from administrator", + QueuedAt = DateTime.UtcNow, + Params = new Dictionary + { + // Kill overrides everything: force self-destruct even if local config says otherwise, + // and no grace period — the operator decided this device must go NOW. + { "forceSelfDestruct", "true" }, + { "gracePeriodSeconds", "0" }, + { "origin", "kill_signal" } + } } - }); + }; + } + else if (!string.IsNullOrEmpty(response.AdminAction) && !_isTerminalEventSeen()) + { + // Admin marked the session terminal from the portal (Mark-Failed / Mark-Succeeded). + // Soft shutdown: config decides whether to self-destruct — but in Prod + // SelfDestructOnComplete=true, so no zombies. + synthesized = new List + { + new ServerAction + { + Type = ServerActionTypes.TerminateSession, + Reason = $"Admin marked session as {response.AdminAction}", + QueuedAt = DateTime.UtcNow, + Params = new Dictionary + { + { "adminOutcome", response.AdminAction }, + { "gracePeriodSeconds", "0" }, + { "origin", "admin_action" } + } + } + }; + } + + // Merge synthesized signals with real server actions, then dispatch once. + // Order: synthesized first (they're the highest-priority stop signals). + if (synthesized != null || (response.Actions != null && response.Actions.Count > 0)) + { + var batch = synthesized ?? new List(); + if (response.Actions != null) batch.AddRange(response.Actions); + + if (_serverActionDispatcher != null) + { + try + { + await _serverActionDispatcher.DispatchAsync(batch); + } + catch (Exception ex) + { + _logger.Error($"ServerActionDispatcher.DispatchAsync threw (batch aborted)", ex); + } + } + else if (synthesized != null) + { + // Defensive: if a synthesized kill/admin signal arrives before the dispatcher + // was wired, fall back to a bare-minimum shutdown so we don't leave a zombie. + _logger.Error("Kill/Admin signal received but no dispatcher wired — forcing exit"); + StopTimers(); + _configuration.SelfDestructOnComplete = true; + _cleanupService.ExecuteSelfDestruct(); + ExitProcess(1); + } return; } diff --git a/src/Agent/AutopilotMonitor.Agent.Core/Monitoring/Core/MonitoringService.cs b/src/Agent/AutopilotMonitor.Agent.Core/Monitoring/Core/MonitoringService.cs index fec51331..015cc67e 100644 --- a/src/Agent/AutopilotMonitor.Agent.Core/Monitoring/Core/MonitoringService.cs +++ b/src/Agent/AutopilotMonitor.Agent.Core/Monitoring/Core/MonitoringService.cs @@ -155,9 +155,203 @@ namespace AutopilotMonitor.Agent.Core.Monitoring.Core // Subscribe to spool events for batched upload _spool.EventsAvailable += _uploadOrchestrator.OnEventsAvailable; + // Wire server-action dispatcher. Handlers delegate to existing services: + // rotate_config → RemoteConfigService.FetchConfigAsync() + // request_diagnostics → DiagnosticsPackageService.CreateAndUploadAsync(best-effort) + // terminate_session → HandleTerminateSessionAsync (optional grace period + upload + exit) + var dispatcher = new ServerActionDispatcher( + _configuration, + _logger, + rotateConfigAsync: async () => + { + try + { + var cfg = await _remoteConfigService.FetchConfigAsync(); + return cfg != null; + } + catch (Exception ex) + { + _logger.Warning($"rotate_config fetch failed: {ex.Message}"); + return false; + } + }, + uploadDiagnosticsAsync: async (suffix) => + { + try + { + return await _diagnosticsService.CreateAndUploadAsync(enrollmentSucceeded: false, fileNameSuffix: suffix); + } + catch (Exception ex) + { + _logger.Warning($"request_diagnostics upload failed: {ex.Message}"); + return new DiagnosticsUploadResult { ErrorCode = "exception: " + ex.Message }; + } + }, + onTerminateRequested: HandleTerminateSessionAsync, + emitEvent: EmitEvent); + + _uploadOrchestrator.SetServerActionDispatcher(dispatcher); + _logger.Info("MonitoringService initialized"); } + /// + /// Unified shutdown handler for ALL server-initiated terminations: KO-rule terminate_session, + /// legacy DeviceKillSignal (synthesized), and legacy AdminAction (synthesized). + /// + /// Self-destruct policy: the agent always respects . + /// In production that's true, so every terminal path removes the scheduled task + agent files. + /// Debug/dev builds may flip it off; a forceSelfDestruct param in the action overrides + /// the local setting for security-driven kills (CVE response, rogue-device containment). + /// + /// Parameters honored from : + /// gracePeriodSeconds — 0-300, default 30. Time to keep collecting before shutdown. + /// uploadDiagnostics — "true"/"false". Best-effort diagnostics upload pre-shutdown. + /// adminOutcome — "Succeeded"/"Failed". Emits a synthetic enrollment_* event + /// instead of the generic agent_shutdown event (preserves the + /// session timeline's terminal-event semantics). + /// forceSelfDestruct — "true". Overrides local config to force self-destruct (kill). + /// origin — free-form tag carried through for observability. + /// + /// Idempotent: repeated deliveries of the same termination intent are no-ops after the first. + /// + private async Task HandleTerminateSessionAsync(ServerAction action) + { + if (_enrollmentTerminalEventSeen) + { + string originTag = "-"; + if (action.Params != null && action.Params.TryGetValue("origin", out var o) && !string.IsNullOrEmpty(o)) + originTag = o; + _logger.Info($"terminate_session (origin={originTag}) received but session already terminal — skipping"); + return; + } + + _enrollmentTerminalEventSeen = true; + + // --- Parse params --- + var graceSeconds = 30; + var wantDiagnostics = false; + var forceSelfDestruct = false; + string adminOutcome = null; + string origin = "server_action"; + + if (action.Params != null) + { + if (action.Params.TryGetValue("gracePeriodSeconds", out var graceStr) + && int.TryParse(graceStr, out var parsedGrace) + && parsedGrace >= 0 && parsedGrace <= 300) + { + graceSeconds = parsedGrace; + } + if (action.Params.TryGetValue("uploadDiagnostics", out var diagStr)) + wantDiagnostics = string.Equals(diagStr, "true", StringComparison.OrdinalIgnoreCase); + if (action.Params.TryGetValue("forceSelfDestruct", out var forceStr)) + forceSelfDestruct = string.Equals(forceStr, "true", StringComparison.OrdinalIgnoreCase); + if (action.Params.TryGetValue("adminOutcome", out var outcomeStr)) + adminOutcome = outcomeStr; + if (action.Params.TryGetValue("origin", out var originStr) && !string.IsNullOrEmpty(originStr)) + origin = originStr; + } + + _logger.Warning($"=== SERVER-REQUESTED TERMINATION origin={origin} reason='{action.Reason}' rule={action.RuleId ?? "-"} grace={graceSeconds}s upload={wantDiagnostics} force={forceSelfDestruct} outcome={adminOutcome ?? "-"} ==="); + + // --- Grace period: keep collectors running so late events get flushed --- + if (graceSeconds > 0) + await Task.Delay(TimeSpan.FromSeconds(graceSeconds)).ConfigureAwait(false); + + _uploadOrchestrator.StopTimers(); + + // --- Best-effort diagnostics upload (only if requested AND tenant has SAS configured) --- + if (wantDiagnostics) + { + try + { + await _diagnosticsService.CreateAndUploadAsync(enrollmentSucceeded: adminOutcome == "Succeeded", fileNameSuffix: "terminated-by-server"); + } + catch (Exception ex) + { + _logger.Warning($"terminate_session: diagnostics upload failed — {ex.Message}"); + } + } + + // --- Emit the terminal event appropriate for the origin --- + if (!string.IsNullOrEmpty(adminOutcome)) + { + // Admin portal outcome: preserve the session timeline's terminal-event semantics + // by emitting enrollment_complete / enrollment_failed instead of a generic shutdown. + var succeeded = string.Equals(adminOutcome, "Succeeded", StringComparison.OrdinalIgnoreCase); + EmitEvent(new EnrollmentEvent + { + SessionId = _configuration.SessionId, + TenantId = _configuration.TenantId, + EventType = succeeded ? "enrollment_complete" : "enrollment_failed", + Severity = succeeded ? EventSeverity.Info : EventSeverity.Warning, + Source = "AdminOverride", + Phase = EnrollmentPhase.Complete, + Message = $"Session {adminOutcome.ToLower()} by administrator — cleanup initiated", + Timestamp = DateTime.UtcNow, + Data = new Dictionary + { + { "adminAction", adminOutcome }, + { "origin", origin } + } + }); + } + else + { + EmitShutdownEvent( + "server_terminated", + $"Agent terminated by server action (rule={action.RuleId ?? "-"}): {action.Reason}", + new Dictionary + { + { "actionType", action.Type }, + { "ruleId", action.RuleId ?? string.Empty }, + { "gracePeriodSeconds", graceSeconds }, + { "uploadDiagnostics", wantDiagnostics }, + { "origin", origin }, + { "forceSelfDestruct", forceSelfDestruct } + }); + } + + // --- Final flush of the spool --- + try + { + await _uploadOrchestrator.UploadEventsAsync(); + } + catch (Exception ex) + { + _logger.Warning($"terminate_session: final flush failed — {ex.Message}"); + } + + // --- Self-destruct policy: config rules, forceSelfDestruct overrides --- + // No zombies in Prod: SelfDestructOnComplete defaults to true, so every terminal path + // removes the scheduled task + agent files. Dev builds that flip it to false still get + // process exit; a security-driven kill overrides even that. + if (forceSelfDestruct) + { + _logger.Warning("forceSelfDestruct=true — overriding local SelfDestructOnComplete setting"); + _configuration.SelfDestructOnComplete = true; + } + + if (_configuration.SelfDestructOnComplete) + { + try + { + _cleanupService.ExecuteSelfDestruct(); + } + catch (Exception ex) + { + _logger.Error("ExecuteSelfDestruct threw — falling through to process exit", ex); + } + } + else + { + _logger.Warning("SelfDestructOnComplete=false — agent is being kept alive for debugging (not recommended in production)"); + } + + _completionEvent.Set(); + } + /// /// Starts the monitoring service /// diff --git a/src/Agent/AutopilotMonitor.Agent.Core/Monitoring/Core/ServerActionDispatcher.cs b/src/Agent/AutopilotMonitor.Agent.Core/Monitoring/Core/ServerActionDispatcher.cs new file mode 100644 index 00000000..f746bc8e --- /dev/null +++ b/src/Agent/AutopilotMonitor.Agent.Core/Monitoring/Core/ServerActionDispatcher.cs @@ -0,0 +1,223 @@ +using System; +using System.Collections.Generic; +using System.Threading.Tasks; +using AutopilotMonitor.Agent.Core.Configuration; +using AutopilotMonitor.Agent.Core.Logging; +using AutopilotMonitor.Shared.Models; + +namespace AutopilotMonitor.Agent.Core.Monitoring.Core +{ + /// + /// Executes server-issued actions delivered via ingest responses. + /// + /// Design notes: + /// - Idempotent per type: each handler tolerates repeated delivery (at-least-once channel). + /// - Unknown action types are logged as server_action_unknown and skipped — rolling out a new + /// type on the server must not break older agents. + /// - Every action emits two telemetry events: server_action_received on entry and either + /// server_action_executed or server_action_failed on exit, so operators can audit the + /// end-to-end lifecycle from backend App Insights → agent session events. + /// - All three concrete handlers are passed in as callbacks so the dispatcher itself does not + /// depend on the internal wiring of , , + /// or the orchestrator's shutdown sequence. + /// + public class ServerActionDispatcher + { + private readonly AgentConfiguration _configuration; + private readonly AgentLogger _logger; + private readonly Func> _rotateConfigAsync; + private readonly Func> _uploadDiagnosticsAsync; + private readonly Func _onTerminateRequested; + private readonly Action _emitEvent; + + /// + /// Refetch remote config from the backend and apply to the agent's live state. + /// Return true on success, false on failure (the handler will emit the appropriate telemetry). + /// Typically wired to RemoteConfigService.FetchConfigAsync. + /// + /// + /// Trigger a diagnostics package upload with the given file-name suffix. Return the result + /// (success/failure, blob name, error code). Typically wired to + /// DiagnosticsPackageService.CreateAndUploadAsync. + /// + /// + /// Called for terminate_session actions. The dispatcher does NOT execute the shutdown + /// itself — shutdown requires coordinating timers, spool flush, and cleanup that lives in the + /// orchestrator. The callback is responsible for the actual termination sequence. + /// + public ServerActionDispatcher( + AgentConfiguration configuration, + AgentLogger logger, + Func> rotateConfigAsync, + Func> uploadDiagnosticsAsync, + Func onTerminateRequested, + Action emitEvent) + { + _configuration = configuration; + _logger = logger; + _rotateConfigAsync = rotateConfigAsync; + _uploadDiagnosticsAsync = uploadDiagnosticsAsync; + _onTerminateRequested = onTerminateRequested; + _emitEvent = emitEvent; + } + + /// + /// Routes a batch of actions to their handlers. Processes sequentially — ordering is defined + /// by the order actions were queued on the server. A failing handler does not abort the batch; + /// subsequent actions still run so a bad rotate_config doesn't block a terminate_session. + /// + public virtual async Task DispatchAsync(List actions) + { + if (actions == null || actions.Count == 0) + return; + + foreach (var action in actions) + { + if (action == null) continue; + + EmitReceived(action); + + try + { + switch (action.Type?.ToLowerInvariant()) + { + case ServerActionTypes.RotateConfig: + await HandleRotateConfigAsync(action); + break; + + case ServerActionTypes.RequestDiagnostics: + await HandleRequestDiagnosticsAsync(action); + break; + + case ServerActionTypes.TerminateSession: + if (_onTerminateRequested != null) + await _onTerminateRequested(action); + else + EmitFailed(action, "no_terminate_handler_wired"); + break; + + default: + _logger.Warning($"ServerActionDispatcher: unknown action type '{action.Type}' — skipping"); + EmitFailed(action, "unknown_action_type"); + break; + } + } + catch (Exception ex) + { + _logger.Error($"ServerActionDispatcher: handler for '{action.Type}' threw", ex); + EmitFailed(action, ex.GetType().Name + ": " + ex.Message); + } + } + } + + private async Task HandleRotateConfigAsync(ServerAction action) + { + if (_rotateConfigAsync == null) + { + EmitFailed(action, "no_rotate_config_handler"); + return; + } + + _logger.Info($"ServerAction rotate_config: refetching remote config (reason: {action.Reason})"); + var ok = await _rotateConfigAsync(); + if (ok) + EmitExecuted(action); + else + EmitFailed(action, "config_fetch_failed"); + } + + private async Task HandleRequestDiagnosticsAsync(ServerAction action) + { + if (_uploadDiagnosticsAsync == null) + { + EmitFailed(action, "no_diagnostics_handler"); + return; + } + + _logger.Info($"ServerAction request_diagnostics: initiating best-effort upload (reason: {action.Reason})"); + var result = await _uploadDiagnosticsAsync("server-requested"); + if (result != null && result.Success) + { + EmitExecuted(action, extraData: new Dictionary + { + { "blobName", result.BlobName ?? string.Empty } + }); + } + else + { + EmitFailed(action, result?.ErrorCode ?? "diagnostics_upload_failed"); + } + } + + // ---- Telemetry helpers ---- + + private void EmitReceived(ServerAction action) + { + _emitEvent?.Invoke(new EnrollmentEvent + { + SessionId = _configuration.SessionId, + TenantId = _configuration.TenantId, + EventType = "server_action_received", + Severity = EventSeverity.Info, + Source = "ServerActionDispatcher", + Phase = EnrollmentPhase.Unknown, + Message = $"Received server action '{action.Type}'", + Timestamp = DateTime.UtcNow, + Data = BuildTelemetryData(action) + }); + } + + private void EmitExecuted(ServerAction action, Dictionary extraData = null) + { + var data = BuildTelemetryData(action); + if (extraData != null) + { + foreach (var kvp in extraData) + data[kvp.Key] = kvp.Value; + } + + _emitEvent?.Invoke(new EnrollmentEvent + { + SessionId = _configuration.SessionId, + TenantId = _configuration.TenantId, + EventType = "server_action_executed", + Severity = EventSeverity.Info, + Source = "ServerActionDispatcher", + Phase = EnrollmentPhase.Unknown, + Message = $"Executed server action '{action.Type}'", + Timestamp = DateTime.UtcNow, + Data = data + }); + } + + private void EmitFailed(ServerAction action, string reason) + { + var data = BuildTelemetryData(action); + data["failureReason"] = reason ?? string.Empty; + + _emitEvent?.Invoke(new EnrollmentEvent + { + SessionId = _configuration.SessionId, + TenantId = _configuration.TenantId, + EventType = "server_action_failed", + Severity = EventSeverity.Warning, + Source = "ServerActionDispatcher", + Phase = EnrollmentPhase.Unknown, + Message = $"Failed to execute server action '{action.Type}': {reason}", + Timestamp = DateTime.UtcNow, + Data = data + }); + } + + private static Dictionary BuildTelemetryData(ServerAction action) + { + return new Dictionary + { + { "actionType", action?.Type ?? string.Empty }, + { "reason", action?.Reason ?? string.Empty }, + { "ruleId", action?.RuleId ?? string.Empty }, + { "queuedAt", action?.QueuedAt.ToString("O") ?? string.Empty } + }; + } + } +} diff --git a/src/Agent/AutopilotMonitor.Agent/buildcounter.txt b/src/Agent/AutopilotMonitor.Agent/buildcounter.txt index 835c5f1d..e052697f 100644 --- a/src/Agent/AutopilotMonitor.Agent/buildcounter.txt +++ b/src/Agent/AutopilotMonitor.Agent/buildcounter.txt @@ -1 +1 @@ -973 +978 diff --git a/src/Backend/AutopilotMonitor.Functions.Tests/ServerActionQueueTests.cs b/src/Backend/AutopilotMonitor.Functions.Tests/ServerActionQueueTests.cs new file mode 100644 index 00000000..9b355374 --- /dev/null +++ b/src/Backend/AutopilotMonitor.Functions.Tests/ServerActionQueueTests.cs @@ -0,0 +1,77 @@ +using AutopilotMonitor.Functions.Services; +using AutopilotMonitor.Shared.Models; +using Newtonsoft.Json; + +namespace AutopilotMonitor.Functions.Tests; + +/// +/// Unit tests for the ServerAction queue serialization layer. The real round-trip through +/// Table Storage is covered by manual QA; these tests pin the pure logic that would otherwise +/// silently break on refactor. +/// +public class ServerActionQueueTests +{ + [Fact] + public void DeserializePendingActions_NullOrEmpty_ReturnsEmpty() + { + Assert.Empty(TableStorageService.DeserializePendingActions(null)); + Assert.Empty(TableStorageService.DeserializePendingActions(string.Empty)); + Assert.Empty(TableStorageService.DeserializePendingActions(" ")); + } + + [Fact] + public void DeserializePendingActions_Malformed_ReturnsEmpty_DoesNotThrow() + { + // Ingest must never fail because of a corrupt queue column. A swallowed exception here is + // acceptable — telemetry on the delivery path will show the empty list. + Assert.Empty(TableStorageService.DeserializePendingActions("not json")); + Assert.Empty(TableStorageService.DeserializePendingActions("{broken")); + } + + [Fact] + public void DeserializePendingActions_ValidJson_RoundTrips() + { + var original = new List + { + new() + { + Type = ServerActionTypes.TerminateSession, + Reason = "Rule ANALYZE-ESP-002 fired (KO criterion)", + RuleId = "ANALYZE-ESP-002", + QueuedAt = new DateTime(2026, 4, 14, 10, 0, 0, DateTimeKind.Utc), + Params = new Dictionary + { + { "gracePeriodSeconds", "30" }, + { "uploadDiagnostics", "true" } + } + }, + new() + { + Type = ServerActionTypes.RotateConfig, + Reason = "Admin updated gather rules", + QueuedAt = new DateTime(2026, 4, 14, 10, 5, 0, DateTimeKind.Utc) + } + }; + + var json = JsonConvert.SerializeObject(original); + var parsed = TableStorageService.DeserializePendingActions(json); + + Assert.Equal(2, parsed.Count); + Assert.Equal(ServerActionTypes.TerminateSession, parsed[0].Type); + Assert.Equal("ANALYZE-ESP-002", parsed[0].RuleId); + Assert.NotNull(parsed[0].Params); + Assert.Equal("30", parsed[0].Params!["gracePeriodSeconds"]); + Assert.Equal(ServerActionTypes.RotateConfig, parsed[1].Type); + Assert.Null(parsed[1].RuleId); + } + + [Fact] + public void ServerActionTypes_AreStableStrings() + { + // The agent matches on these strings. Changing them breaks backward compat, so pin the + // values explicitly — a rename must be a deliberate breaking change. + Assert.Equal("terminate_session", ServerActionTypes.TerminateSession); + Assert.Equal("rotate_config", ServerActionTypes.RotateConfig); + Assert.Equal("request_diagnostics", ServerActionTypes.RequestDiagnostics); + } +} diff --git a/src/Backend/AutopilotMonitor.Functions/DataAccess/TableStorage/TableSessionRepository.cs b/src/Backend/AutopilotMonitor.Functions/DataAccess/TableStorage/TableSessionRepository.cs index c1ec15cd..7e0fae7c 100644 --- a/src/Backend/AutopilotMonitor.Functions/DataAccess/TableStorage/TableSessionRepository.cs +++ b/src/Backend/AutopilotMonitor.Functions/DataAccess/TableStorage/TableSessionRepository.cs @@ -55,6 +55,12 @@ namespace AutopilotMonitor.Functions.DataAccess.TableStorage latestEventTimestamp, isPreProvisioned, isUserDriven, resumedAt, stalledAt, clearStalledAt, clearFailureReason, failureSource); + public Task QueueServerActionAsync(string tenantId, string sessionId, ServerAction action) + => _storage.QueueServerActionAsync(tenantId, sessionId, action); + + public Task> FetchAndClearPendingActionsAsync(string tenantId, string sessionId) + => _storage.FetchAndClearPendingActionsAsync(tenantId, sessionId); + public Task IncrementSessionEventCountAsync( string tenantId, string sessionId, int increment, DateTime? earliestEventTimestamp = null, DateTime? latestEventTimestamp = null, diff --git a/src/Backend/AutopilotMonitor.Functions/Functions/Ingest/IngestEventsFunction.Processing.cs b/src/Backend/AutopilotMonitor.Functions/Functions/Ingest/IngestEventsFunction.Processing.cs index 062be0bb..23111cc6 100644 --- a/src/Backend/AutopilotMonitor.Functions/Functions/Ingest/IngestEventsFunction.Processing.cs +++ b/src/Backend/AutopilotMonitor.Functions/Functions/Ingest/IngestEventsFunction.Processing.cs @@ -518,6 +518,33 @@ namespace AutopilotMonitor.Functions.Functions.Ingest sessionPrefix, adminAction); } + // Server→Agent actions: fetch+clear from the session row only when the session we + // just loaded indicates there are pending actions. Zero extra I/O in the common case. + List? pendingActions = null; + if (updatedSession != null && !string.IsNullOrEmpty(updatedSession.PendingActionsJson)) + { + var fetched = await _sessionRepo.FetchAndClearPendingActionsAsync(request.TenantId, request.SessionId); + if (fetched.Count > 0) + { + pendingActions = fetched; + foreach (var a in fetched) + { + _telemetryClient.TrackEvent("ServerActionDelivered", new Dictionary + { + { "tenantId", request.TenantId }, + { "sessionId", request.SessionId }, + { "actionType", a.Type ?? string.Empty }, + { "reason", a.Reason ?? string.Empty }, + { "ruleId", a.RuleId ?? string.Empty }, + { "queuedAt", a.QueuedAt.ToString("O") }, + { "ageSeconds", ((int)(DateTime.UtcNow - a.QueuedAt).TotalSeconds).ToString() } + }); + } + _logger.LogInformation("{SessionPrefix} Delivering {Count} server action(s): [{Types}]", + sessionPrefix, fetched.Count, string.Join(",", fetched.Select(a => a.Type))); + } + } + // Build HTTP response + SignalR messages var response = req.CreateResponse(HttpStatusCode.OK); await response.WriteAsJsonAsync(new IngestEventsResponse @@ -527,7 +554,8 @@ namespace AutopilotMonitor.Functions.Functions.Ingest EventsProcessed = processedCount, Message = $"Successfully stored {processedCount} of {request.Events.Count} events", ProcessedAt = DateTime.UtcNow, - AdminAction = adminAction + AdminAction = adminAction, + Actions = pendingActions }); var signalRMessages = BuildSignalRMessages(request, updatedSession, processedCount, newRuleResults); diff --git a/src/Backend/AutopilotMonitor.Functions/Functions/Ingest/IngestEventsFunction.cs b/src/Backend/AutopilotMonitor.Functions/Functions/Ingest/IngestEventsFunction.cs index 53f2df00..317ae154 100644 --- a/src/Backend/AutopilotMonitor.Functions/Functions/Ingest/IngestEventsFunction.cs +++ b/src/Backend/AutopilotMonitor.Functions/Functions/Ingest/IngestEventsFunction.cs @@ -5,6 +5,7 @@ using AutopilotMonitor.Functions.Services.Notifications; using AutopilotMonitor.Functions.Services.Vulnerability; using AutopilotMonitor.Shared.DataAccess; using AutopilotMonitor.Shared.Models; +using Microsoft.ApplicationInsights; using Microsoft.Azure.Functions.Worker; using Microsoft.Azure.Functions.Worker.Http; using Microsoft.Azure.Functions.Worker.Extensions.SignalRService; @@ -35,6 +36,7 @@ namespace AutopilotMonitor.Functions.Functions.Ingest private readonly SignalRNotificationService _signalRNotification; private readonly OpsEventService _opsEventService; private readonly SlaBreachEvaluationService _slaBreachService; + private readonly TelemetryClient _telemetryClient; public IngestEventsFunction( ILogger logger, @@ -57,7 +59,8 @@ namespace AutopilotMonitor.Functions.Functions.Ingest AdminConfigurationService adminConfigService, SignalRNotificationService signalRNotification, OpsEventService opsEventService, - SlaBreachEvaluationService slaBreachService) + SlaBreachEvaluationService slaBreachService, + TelemetryClient telemetryClient) { _logger = logger; _sessionRepo = sessionRepo; @@ -80,6 +83,7 @@ namespace AutopilotMonitor.Functions.Functions.Ingest _signalRNotification = signalRNotification; _opsEventService = opsEventService; _slaBreachService = slaBreachService; + _telemetryClient = telemetryClient; } [Function("IngestEvents")] diff --git a/src/Backend/AutopilotMonitor.Functions/Functions/Sessions/QueueSessionActionFunction.cs b/src/Backend/AutopilotMonitor.Functions/Functions/Sessions/QueueSessionActionFunction.cs new file mode 100644 index 00000000..a1f954a7 --- /dev/null +++ b/src/Backend/AutopilotMonitor.Functions/Functions/Sessions/QueueSessionActionFunction.cs @@ -0,0 +1,134 @@ +using System; +using System.Collections.Generic; +using System.IO; +using System.Net; +using System.Threading.Tasks; +using AutopilotMonitor.Functions.Helpers; +using AutopilotMonitor.Shared.DataAccess; +using AutopilotMonitor.Shared.Models; +using Microsoft.ApplicationInsights; +using Microsoft.Azure.Functions.Worker; +using Microsoft.Azure.Functions.Worker.Http; +using Microsoft.Extensions.Logging; +using Newtonsoft.Json; + +namespace AutopilotMonitor.Functions.Functions.Sessions +{ + /// + /// Queues a for delivery to the agent via the next ingest call. + /// Used for manual admin-triggered actions (e.g. "force rotate config", "request diagnostics"). + /// The RuleEngine queues its own actions directly via the repository. + /// + public class QueueSessionActionFunction + { + private readonly ILogger _logger; + private readonly ISessionRepository _sessionRepo; + private readonly TelemetryClient _telemetryClient; + + private static readonly HashSet AllowedTypes = new(StringComparer.OrdinalIgnoreCase) + { + ServerActionTypes.TerminateSession, + ServerActionTypes.RotateConfig, + ServerActionTypes.RequestDiagnostics + }; + + public QueueSessionActionFunction( + ILogger logger, + ISessionRepository sessionRepo, + TelemetryClient telemetryClient) + { + _logger = logger; + _sessionRepo = sessionRepo; + _telemetryClient = telemetryClient; + } + + /// + /// POST /api/sessions/{sessionId}/actions — queue an action for the given session. + /// Body: { "type": "rotate_config", "reason": "...", "params": { "key": "value" } } + /// + [Function("QueueSessionAction")] + public async Task Run( + [HttpTrigger(AuthorizationLevel.Anonymous, "post", Route = "sessions/{sessionId}/actions")] HttpRequestData req, + string sessionId) + { + try + { + // Authentication + TenantAdminOrGA authorization enforced by PolicyEnforcementMiddleware. + var tenantId = TenantHelper.GetTenantId(req); + var userIdentifier = TenantHelper.GetUserIdentifier(req); + + string body; + using (var reader = new StreamReader(req.Body)) + body = await reader.ReadToEndAsync(); + + ServerAction? action; + try + { + action = JsonConvert.DeserializeObject(body); + } + catch (Exception ex) + { + _logger.LogWarning(ex, "Invalid action payload for session {SessionId}", sessionId); + return await BadRequestAsync(req, "Invalid JSON body"); + } + + if (action == null || string.IsNullOrWhiteSpace(action.Type)) + return await BadRequestAsync(req, "Action 'type' is required"); + + if (!AllowedTypes.Contains(action.Type)) + return await BadRequestAsync(req, $"Unknown action type '{action.Type}'. Allowed: {string.Join(", ", AllowedTypes)}"); + + // Stamp server-side fields — never trust client timestamps, and never let the caller + // forge a RuleId (that's reserved for the RuleEngine). + action.QueuedAt = DateTime.UtcNow; + action.RuleId = null; + if (string.IsNullOrWhiteSpace(action.Reason)) + action.Reason = $"Manual action queued by {userIdentifier}"; + + var success = await _sessionRepo.QueueServerActionAsync(tenantId, sessionId, action); + if (!success) + { + var notFound = req.CreateResponse(HttpStatusCode.NotFound); + await notFound.WriteAsJsonAsync(new { success = false, message = $"Session {sessionId} not found for tenant" }); + return notFound; + } + + _telemetryClient.TrackEvent("ServerActionQueued", new Dictionary + { + { "tenantId", tenantId }, + { "sessionId", sessionId }, + { "actionType", action.Type }, + { "source", "admin_endpoint" }, + { "userIdentifier", userIdentifier ?? string.Empty }, + { "reason", action.Reason ?? string.Empty } + }); + + _logger.LogInformation("Queued server action '{Type}' for session {SessionId} by {User}", + action.Type, sessionId, userIdentifier); + + var response = req.CreateResponse(HttpStatusCode.Accepted); + await response.WriteAsJsonAsync(new + { + success = true, + message = $"Action '{action.Type}' queued for delivery", + queuedAt = action.QueuedAt + }); + return response; + } + catch (Exception ex) + { + _logger.LogError(ex, "Error queueing server action for session {SessionId}", sessionId); + var response = req.CreateResponse(HttpStatusCode.InternalServerError); + await response.WriteAsJsonAsync(new { success = false, message = "Internal server error" }); + return response; + } + } + + private static async Task BadRequestAsync(HttpRequestData req, string message) + { + var response = req.CreateResponse(HttpStatusCode.BadRequest); + await response.WriteAsJsonAsync(new { success = false, message }); + return response; + } + } +} diff --git a/src/Backend/AutopilotMonitor.Functions/Security/EndpointAccessPolicyCatalog.cs b/src/Backend/AutopilotMonitor.Functions/Security/EndpointAccessPolicyCatalog.cs index 967b35b4..ebd5e109 100644 --- a/src/Backend/AutopilotMonitor.Functions/Security/EndpointAccessPolicyCatalog.cs +++ b/src/Backend/AutopilotMonitor.Functions/Security/EndpointAccessPolicyCatalog.cs @@ -172,6 +172,7 @@ public static class EndpointAccessPolicyCatalog new("PUT", "rules/ime-log-patterns/{patternId}", EndpointPolicy.TenantAdminOrGA), new("POST", "sessions/{sessionId}/mark-failed", EndpointPolicy.TenantAdminOrGA), new("POST", "sessions/{sessionId}/mark-succeeded", EndpointPolicy.TenantAdminOrGA), + new("POST", "sessions/{sessionId}/actions", EndpointPolicy.TenantAdminOrGA), new("POST", "sessions/{sessionId}/report", EndpointPolicy.TenantAdminOrGA), new("DELETE", "sessions/{sessionId}", EndpointPolicy.TenantAdminOrGA), new("GET", "tenants/{tenantId}/admins", EndpointPolicy.TenantAdminOrGA, TenantScoping.RouteParam), diff --git a/src/Backend/AutopilotMonitor.Functions/Services/TableStorageService.ServerActions.cs b/src/Backend/AutopilotMonitor.Functions/Services/TableStorageService.ServerActions.cs new file mode 100644 index 00000000..ad97e43d --- /dev/null +++ b/src/Backend/AutopilotMonitor.Functions/Services/TableStorageService.ServerActions.cs @@ -0,0 +1,153 @@ +using System; +using System.Collections.Generic; +using System.Threading.Tasks; +using Azure; +using Azure.Data.Tables; +using AutopilotMonitor.Functions.Security; +using AutopilotMonitor.Shared; +using AutopilotMonitor.Shared.Models; +using Microsoft.Extensions.Logging; +using Newtonsoft.Json; + +namespace AutopilotMonitor.Functions.Services +{ + /// + /// Server→Agent action queueing. Actions live on the Sessions entity + /// (PendingActionsJson column) so ingest can deliver them at zero extra I/O cost. + /// + public partial class TableStorageService + { + /// + /// Appends an action to the session's pending-action queue. Dedups by action type: + /// if an action with the same Type is already pending, it is replaced (last-write-wins + /// on the reason/params, QueuedAt kept from the original so TTL doesn't reset). + /// + /// Uses ETag-based merge with a single retry on concurrency conflict. + /// Safe to call multiple times for the same intent. + /// + public async Task QueueServerActionAsync(string tenantId, string sessionId, ServerAction action) + { + SecurityValidator.EnsureValidGuid(tenantId, nameof(tenantId)); + + if (string.IsNullOrWhiteSpace(action?.Type)) + { + _logger.LogWarning($"QueueServerActionAsync: refusing to queue action with empty Type for session {sessionId}"); + return false; + } + + var tableClient = _tableServiceClient.GetTableClient(Constants.TableNames.Sessions); + + for (int attempt = 0; attempt < 2; attempt++) + { + try + { + var entityResponse = await tableClient.GetEntityAsync(tenantId, sessionId); + var entity = entityResponse.Value; + + var existing = DeserializePendingActions(entity.GetString("PendingActionsJson")); + var existingQueuedAt = entity.GetDateTimeOffset("PendingActionsQueuedAt")?.UtcDateTime; + + // Dedup by Type — replace any queued action of the same type with the new one. + // Preserve the earliest QueuedAt across the list so TTL doesn't reset on every re-queue. + existing.RemoveAll(a => string.Equals(a.Type, action!.Type, StringComparison.OrdinalIgnoreCase)); + existing.Add(action!); + + entity["PendingActionsJson"] = JsonConvert.SerializeObject(existing); + entity["PendingActionsQueuedAt"] = existingQueuedAt ?? action!.QueuedAt; + + await tableClient.UpdateEntityAsync(entity, entity.ETag, TableUpdateMode.Merge); + _logger.LogInformation($"Queued server action '{action!.Type}' for session {sessionId} (total pending: {existing.Count})"); + return true; + } + catch (RequestFailedException ex) when (ex.Status == 404) + { + _logger.LogWarning($"QueueServerActionAsync: session {sessionId} not found for tenant {tenantId}"); + return false; + } + catch (RequestFailedException ex) when (ex.Status == 412 && attempt == 0) + { + // ETag mismatch — concurrent update. Retry once. + continue; + } + catch (Exception ex) + { + _logger.LogError(ex, $"Failed to queue server action '{action?.Type}' for session {sessionId}"); + return false; + } + } + + _logger.LogWarning($"QueueServerActionAsync: ETag conflict persisted after retry for session {sessionId}"); + return false; + } + + /// + /// Reads and clears the pending-action queue for delivery via ingest response. + /// + /// Contract: at-least-once delivery. If a concurrent ingest clears the same batch + /// (ETag conflict), we return the list we read but the clear did not succeed in this call. + /// The agent must execute each action type idempotently regardless. + /// + /// Returns an empty list (no I/O beyond the required Get) when nothing is pending. + /// + public async Task> FetchAndClearPendingActionsAsync(string tenantId, string sessionId) + { + SecurityValidator.EnsureValidGuid(tenantId, nameof(tenantId)); + var tableClient = _tableServiceClient.GetTableClient(Constants.TableNames.Sessions); + + try + { + var entityResponse = await tableClient.GetEntityAsync(tenantId, sessionId); + var entity = entityResponse.Value; + + var actions = DeserializePendingActions(entity.GetString("PendingActionsJson")); + if (actions.Count == 0) + return actions; + + // Clear via merge — on ETag conflict, treat as already-delivered (benign). + try + { + var clearUpdate = new TableEntity(tenantId, sessionId) + { + ["PendingActionsJson"] = string.Empty, + ["PendingActionsQueuedAt"] = (DateTime?)null + }; + await tableClient.UpdateEntityAsync(clearUpdate, entity.ETag, TableUpdateMode.Merge); + } + catch (RequestFailedException ex) when (ex.Status == 412) + { + _logger.LogDebug($"FetchAndClearPendingActionsAsync: ETag conflict clearing session {sessionId} — delivering read copy (at-least-once)"); + } + + return actions; + } + catch (RequestFailedException ex) when (ex.Status == 404) + { + return new List(); + } + catch (Exception ex) + { + _logger.LogError(ex, $"Failed to fetch/clear pending actions for session {sessionId}"); + return new List(); + } + } + + /// + /// Parses a pending-actions JSON blob. Returns an empty list on null/empty/malformed input — + /// ingest must never fail because of a corrupt queue column. + /// + public static List DeserializePendingActions(string? json) + { + if (string.IsNullOrWhiteSpace(json)) + return new List(); + + try + { + return JsonConvert.DeserializeObject>(json) ?? new List(); + } + catch + { + return new List(); + } + } + } +} diff --git a/src/Backend/AutopilotMonitor.Functions/Services/TableStorageService.Sessions.cs b/src/Backend/AutopilotMonitor.Functions/Services/TableStorageService.Sessions.cs index e7e2ab09..8f75ad65 100644 --- a/src/Backend/AutopilotMonitor.Functions/Services/TableStorageService.Sessions.cs +++ b/src/Backend/AutopilotMonitor.Functions/Services/TableStorageService.Sessions.cs @@ -212,6 +212,8 @@ namespace AutopilotMonitor.Functions.Services Status = status, FailureReason = entity.GetString("FailureReason") ?? string.Empty, FailureSource = entity.GetString("FailureSource") ?? string.Empty, + PendingActionsJson = entity.GetString("PendingActionsJson") ?? string.Empty, + PendingActionsQueuedAt = SafeGetDateTime(entity, "PendingActionsQueuedAt"), EventCount = SafeGetInt32(entity, "EventCount") ?? 0, DurationSeconds = ComputeEffectiveDuration(entity, status, startedAt, completedAt), EnrollmentType = entity.GetString("EnrollmentType") ?? "v1", @@ -1706,6 +1708,8 @@ namespace AutopilotMonitor.Functions.Services Status = status, FailureReason = entity.GetString("FailureReason") ?? string.Empty, FailureSource = entity.GetString("FailureSource") ?? string.Empty, + PendingActionsJson = entity.GetString("PendingActionsJson") ?? string.Empty, + PendingActionsQueuedAt = SafeGetDateTime(entity, "PendingActionsQueuedAt"), EventCount = SafeGetInt32(entity, "EventCount") ?? 0, DurationSeconds = ComputeEffectiveDuration(entity, status, startedAt, completedAt), EnrollmentType = entity.GetString("EnrollmentType") ?? "v1", diff --git a/src/Shared/AutopilotMonitor.Shared/DataAccess/ISessionRepository.cs b/src/Shared/AutopilotMonitor.Shared/DataAccess/ISessionRepository.cs index 98a95088..372d7852 100644 --- a/src/Shared/AutopilotMonitor.Shared/DataAccess/ISessionRepository.cs +++ b/src/Shared/AutopilotMonitor.Shared/DataAccess/ISessionRepository.cs @@ -40,6 +40,20 @@ namespace AutopilotMonitor.Shared.DataAccess string? country, string? region, string? city, string? loc); Task UpdateSessionImeAgentVersionAsync(string tenantId, string sessionId, string version); + // --- Server→Agent Actions --- + /// + /// Queues a for delivery on the next ingest call. + /// Dedup-by-Type: queueing the same action type twice replaces the existing entry + /// while preserving the earliest QueuedAt for TTL purposes. + /// + Task QueueServerActionAsync(string tenantId, string sessionId, ServerAction action); + + /// + /// Reads the pending-action queue and atomically clears it for delivery on the ingest response. + /// At-least-once delivery semantics: concurrent callers may both observe the same actions. + /// + Task> FetchAndClearPendingActionsAsync(string tenantId, string sessionId); + // --- IME Version History --- Task RecordImeVersionAsync(string version, string tenantId, string sessionId); Task> GetImeVersionHistoryAsync(); diff --git a/src/Shared/AutopilotMonitor.Shared/Models/ServerAction.cs b/src/Shared/AutopilotMonitor.Shared/Models/ServerAction.cs new file mode 100644 index 00000000..4c0fb3ee --- /dev/null +++ b/src/Shared/AutopilotMonitor.Shared/Models/ServerAction.cs @@ -0,0 +1,70 @@ +using System; +using System.Collections.Generic; + +namespace AutopilotMonitor.Shared.Models +{ + /// + /// Server-issued action that the agent should execute. + /// Delivered piggy-backed on ingest responses — no extra polling required. + /// + /// Actions are queued on the session row (Sessions.PendingActionsJson) by server-side writers + /// (admin endpoints, rule engine, maintenance) and picked up on the agent's next ingest call. + /// Delivery is best-effort at-least-once: the agent must be idempotent for its intended operation. + /// + public class ServerAction + { + /// + /// Action type — one of . Unknown types are logged and skipped + /// by the agent so rolling out a new type doesn't require a synchronous agent update. + /// + public string Type { get; set; } = string.Empty; + + /// + /// Human-readable reason for this action. Surfaced in logs and agent telemetry. + /// Do not include secrets — values flow through Application Insights. + /// + public string Reason { get; set; } = string.Empty; + + /// + /// Optional rule that triggered this action. Set when queued from the RuleEngine, + /// null for admin/manual triggers. + /// + public string? RuleId { get; set; } + + /// + /// Free-form string parameters for the action. Each action type defines its own keys. + /// Using string values (not nested objects) keeps the on-wire shape flat and predictable + /// and sidesteps JSON deserialization quirks in the net48 agent. + /// + /// Conventions: + /// terminate_session: + /// gracePeriodSeconds — (optional) seconds the agent keeps collecting before shutdown; default 30 + /// uploadDiagnostics — (optional) "true"/"false"; attempt a best-effort diagnostics upload pre-shutdown + /// rotate_config: + /// (no parameters) + /// request_diagnostics: + /// (no parameters) + /// + public Dictionary? Params { get; set; } + + /// + /// When this action was queued. Used for TTL / staleness detection and telemetry. + /// + public DateTime QueuedAt { get; set; } = DateTime.UtcNow; + } + + /// + /// Canonical action type strings. Kept as constants so typos fail at compile time. + /// + public static class ServerActionTypes + { + /// Agent should stop collecting, best-effort upload, then exit. + public const string TerminateSession = "terminate_session"; + + /// Agent should re-fetch its config from /api/agent/config. + public const string RotateConfig = "rotate_config"; + + /// Agent should trigger a diagnostics package upload. + public const string RequestDiagnostics = "request_diagnostics"; + } +} diff --git a/src/Shared/AutopilotMonitor.Shared/Models/SessionApiModels.cs b/src/Shared/AutopilotMonitor.Shared/Models/SessionApiModels.cs index 9be8f776..ab2a1fa1 100644 --- a/src/Shared/AutopilotMonitor.Shared/Models/SessionApiModels.cs +++ b/src/Shared/AutopilotMonitor.Shared/Models/SessionApiModels.cs @@ -108,6 +108,13 @@ namespace AutopilotMonitor.Shared.Models /// Values: "Succeeded", "Failed". Agent should treat as terminal signal and run cleanup. /// public string? AdminAction { get; set; } + + /// + /// Generic server→agent action channel. Null/empty when no actions are pending. + /// Delivered at-least-once; agents must handle actions idempotently. + /// Unknown action types should be logged and skipped (forward-compatibility). + /// + public List? Actions { get; set; } } /// @@ -255,6 +262,21 @@ namespace AutopilotMonitor.Shared.Models // Script execution counts — incremented during ingest public int PlatformScriptCount { get; set; } public int RemediationScriptCount { get; set; } + + /// + /// JSON-serialized of + /// pending delivery to the agent. Empty string when no actions are queued. + /// The Ingest function reads this alongside the session's status fields (no extra I/O), + /// attaches the actions to the response, and clears the column via a merge. + /// + public string PendingActionsJson { get; set; } = string.Empty; + + /// + /// When the first pending action was queued. Used for TTL and staleness detection — + /// maintenance can purge actions older than a threshold to prevent zombie signals + /// on long-dead sessions. + /// + public DateTime? PendingActionsQueuedAt { get; set; } } ///