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; }
}
///