diff --git a/src/Agent/AutopilotMonitor.Agent.V2.Core.Tests/Monitoring/Transport/EmergencyReporterTests.cs b/src/Agent/AutopilotMonitor.Agent.V2.Core.Tests/Monitoring/Transport/EmergencyReporterTests.cs
index 690960da..0a3fce24 100644
--- a/src/Agent/AutopilotMonitor.Agent.V2.Core.Tests/Monitoring/Transport/EmergencyReporterTests.cs
+++ b/src/Agent/AutopilotMonitor.Agent.V2.Core.Tests/Monitoring/Transport/EmergencyReporterTests.cs
@@ -3,6 +3,7 @@ using System.Collections.Generic;
using System.Net.Http;
using System.Threading.Tasks;
using AutopilotMonitor.Agent.V2.Core.Monitoring.Transport;
+using AutopilotMonitor.Agent.V2.Core.Tests.Harness;
using AutopilotMonitor.Shared.Models;
using Xunit;
@@ -49,6 +50,74 @@ namespace AutopilotMonitor.Agent.V2.Core.Tests.Monitoring.Transport
private static EmergencyReporter Reporter(FakeApiClient client)
=> new EmergencyReporter(client, "session", "tenant", "0.0.0", logger: null);
+ private static EmergencyReporter Reporter(FakeApiClient client, VirtualClock clock)
+ => new EmergencyReporter(client, "session", "tenant", "0.0.0", logger: null, clock: clock);
+
+ private static VirtualClock Clock() => new VirtualClock(new DateTime(2026, 10, 9, 10, 0, 0, DateTimeKind.Utc));
+
+ [Fact]
+ public async Task Session_unknown_report_is_not_suppressed_by_the_cooldown_of_an_earlier_report()
+ {
+ // Terminal report, sent at most once per process right before the termination — a
+ // routine report one minute earlier must not cost it. Every other type keeps the cooldown.
+ var clock = Clock();
+ var client = new FakeApiClient();
+ client.Results.Enqueue(true);
+ client.Results.Enqueue(true);
+ client.Results.Enqueue(true);
+ var reporter = Reporter(client, clock);
+
+ await reporter.TrySendAsync(AgentErrorType.IngestFailed, "m", httpStatusCode: 503);
+ clock.Advance(TimeSpan.FromMinutes(1));
+ await reporter.TrySendAsync(AgentErrorType.TelemetryRejectedSessionUnknown, "m", httpStatusCode: 410);
+
+ Assert.Equal(2, client.Calls);
+ Assert.Equal(AgentErrorType.TelemetryRejectedSessionUnknown, client.Reports[1].ErrorType);
+
+ // Control: a third report of another type inside the window is still held back.
+ clock.Advance(TimeSpan.FromMinutes(1));
+ await reporter.TrySendAsync(AgentErrorType.ConfigFetchFailed, "m");
+ Assert.Equal(2, client.Calls);
+ }
+
+ [Fact]
+ public async Task Second_session_unknown_report_within_the_window_is_still_suppressed()
+ {
+ var clock = Clock();
+ var client = new FakeApiClient();
+ client.Results.Enqueue(true);
+ client.Results.Enqueue(true);
+ var reporter = Reporter(client, clock);
+
+ await reporter.TrySendAsync(AgentErrorType.TelemetryRejectedSessionUnknown, "m", httpStatusCode: 410);
+ clock.Advance(TimeSpan.FromMinutes(1));
+ await reporter.TrySendAsync(AgentErrorType.TelemetryRejectedSessionUnknown, "m", httpStatusCode: 410);
+
+ Assert.Equal(1, client.Calls);
+ }
+
+ [Fact]
+ public async Task Session_unknown_report_still_counts_against_the_per_session_maximum()
+ {
+ // Five delivered reports (ten minutes apart, distinct keys) exhaust the budget; the
+ // exemption waives only the cooldown, not the maximum.
+ var clock = Clock();
+ var client = new FakeApiClient();
+ for (var i = 0; i < 6; i++) client.Results.Enqueue(true);
+ var reporter = Reporter(client, clock);
+
+ foreach (var status in new[] { 500, 502, 503, 504, 429 })
+ {
+ await reporter.TrySendAsync(AgentErrorType.IngestFailed, "m", httpStatusCode: status);
+ clock.Advance(TimeSpan.FromMinutes(10));
+ }
+ Assert.Equal(5, client.Calls);
+
+ await reporter.TrySendAsync(AgentErrorType.TelemetryRejectedSessionUnknown, "m", httpStatusCode: 410);
+
+ Assert.Equal(5, client.Calls);
+ }
+
[Fact]
public async Task Failed_send_rolls_back_reservation_so_a_later_call_retries()
{
@@ -142,6 +211,20 @@ namespace AutopilotMonitor.Agent.V2.Core.Tests.Monitoring.Transport
Assert.Equal(2, client.Calls);
}
+ [Fact]
+ public async Task Session_unknown_report_carries_error_type_and_http_410()
+ {
+ var client = new FakeApiClient();
+ client.Results.Enqueue(true);
+ var reporter = Reporter(client);
+
+ await reporter.TrySendAsync(AgentErrorType.TelemetryRejectedSessionUnknown, "m", httpStatusCode: 410);
+
+ var report = Assert.Single(client.Reports);
+ Assert.Equal(AgentErrorType.TelemetryRejectedSessionUnknown, report.ErrorType);
+ Assert.Equal(410, report.HttpStatusCode);
+ }
+
[Fact]
public async Task Session_age_is_carried_in_the_report_and_absent_by_default()
{
diff --git a/src/Agent/AutopilotMonitor.Agent.V2.Core.Tests/Program/ServerControlPlaneSessionGoneTests.cs b/src/Agent/AutopilotMonitor.Agent.V2.Core.Tests/Program/ServerControlPlaneSessionGoneTests.cs
new file mode 100644
index 00000000..a8b89349
--- /dev/null
+++ b/src/Agent/AutopilotMonitor.Agent.V2.Core.Tests/Program/ServerControlPlaneSessionGoneTests.cs
@@ -0,0 +1,170 @@
+#nullable enable
+using System;
+using System.Collections.Generic;
+using System.IO;
+using System.Threading;
+using System.Threading.Tasks;
+using AutopilotMonitor.Agent.V2.Core.Configuration;
+using AutopilotMonitor.Agent.V2.Core.Logging;
+using AutopilotMonitor.Agent.V2.Core.Monitoring.Runtime;
+using AutopilotMonitor.Agent.V2.Core.Monitoring.Transport;
+using AutopilotMonitor.Agent.V2.Core.Orchestration;
+using AutopilotMonitor.Agent.V2.Core.Tests.Harness;
+using AutopilotMonitor.Agent.V2.Core.Tests.Orchestration;
+using AutopilotMonitor.Shared.Models;
+using Xunit;
+
+namespace AutopilotMonitor.Agent.V2.Core.Tests.Program
+{
+ ///
+ /// Host wiring for TelemetryUploadOrchestrator.SessionGone
+ /// (HTTP 410 on the current session's telemetry): exactly one
+ /// emergency report carrying the
+ /// status, then a synthetic forced terminate_session tagged origin=session_gone
+ /// — and nothing at all once a shutdown is already in progress.
+ ///
+ public sealed class ServerControlPlaneSessionGoneTests
+ {
+ private sealed class RecordingEmergencyReporter : EmergencyReporter
+ {
+ private readonly List _sequence;
+ public List<(AgentErrorType ErrorType, string Message, int? HttpStatusCode)> Reports { get; } =
+ new List<(AgentErrorType, string, int?)>();
+
+ public RecordingEmergencyReporter(List sequence)
+ : base(apiClient: null!, sessionId: "S1", tenantId: "T1", agentVersion: "0.0.0", logger: null!)
+ {
+ _sequence = sequence;
+ }
+
+ public override Task TrySendAsync(
+ AgentErrorType errorType,
+ string message,
+ int? httpStatusCode = null,
+ long? sequenceNumber = null,
+ int attempts = 1,
+ TimeSpan? perAttemptTimeout = null,
+ TimeSpan? retryDelay = null,
+ double? sessionAgeHours = null,
+ RegistrationFailureSummary? priorRegistrationFailure = null)
+ {
+ Reports.Add((errorType, message, httpStatusCode));
+ _sequence.Add("report");
+ return Task.CompletedTask;
+ }
+ }
+
+ private sealed class RecordingDispatcher : ServerActionDispatcher
+ {
+ private readonly List _sequence;
+ public List Dispatched { get; } = new List();
+
+ public RecordingDispatcher(AgentConfiguration config, AgentLogger logger, InformationalEventPost post, List sequence)
+ : base(
+ configuration: config,
+ logger: logger,
+ rotateConfigAsync: () => Task.FromResult(true),
+ uploadDiagnosticsAsync: _ => Task.FromResult(null!),
+ onTerminateRequested: _ => Task.CompletedTask,
+ post: post)
+ {
+ _sequence = sequence;
+ }
+
+ public override Task DispatchAsync(List actions)
+ {
+ Dispatched.AddRange(actions);
+ _sequence.Add("dispatch");
+ return Task.CompletedTask;
+ }
+ }
+
+ private sealed class Rig : IDisposable
+ {
+ public TempDirectory Tmp { get; } = new TempDirectory();
+ public AgentLogger Logger { get; }
+ public List Sequence { get; } = new List();
+ public RecordingEmergencyReporter Reporter { get; }
+ public RecordingDispatcher Dispatcher { get; }
+ public ManualResetEventSlim Shutdown { get; } = new ManualResetEventSlim(false);
+ public ManualResetEventSlim ShutdownComplete { get; } = new ManualResetEventSlim(false);
+
+ public Rig()
+ {
+ Logger = new AgentLogger(Path.Combine(Tmp.Path, "logs"), AgentLogLevel.Info);
+ var config = new AgentConfiguration { SessionId = "S1", TenantId = "T1", ApiBaseUrl = "http://localhost" };
+ var post = new InformationalEventPost(
+ new FakeSignalIngressSink(),
+ new VirtualClock(new DateTime(2026, 10, 9, 10, 0, 0, DateTimeKind.Utc)));
+ Reporter = new RecordingEmergencyReporter(Sequence);
+ Dispatcher = new RecordingDispatcher(config, Logger, post, Sequence);
+ }
+
+ public void Handle(string reason = "session gone: http 410") =>
+ global::AutopilotMonitor.Agent.V2.Runtime.ServerControlPlane.HandleSessionGone(
+ reason, Reporter, Dispatcher, Shutdown, ShutdownComplete, Logger);
+
+ public void Dispose()
+ {
+ Shutdown.Dispose();
+ ShutdownComplete.Dispose();
+ Tmp.Dispose();
+ }
+ }
+
+ [Fact]
+ public void Reports_once_with_410_then_dispatches_forced_terminate_tagged_session_gone()
+ {
+ using var rig = new Rig();
+
+ rig.Handle();
+
+ Assert.Equal(new[] { "report", "dispatch" }, rig.Sequence);
+
+ var report = Assert.Single(rig.Reporter.Reports);
+ Assert.Equal(AgentErrorType.TelemetryRejectedSessionUnknown, report.ErrorType);
+ Assert.Equal(410, report.HttpStatusCode);
+ Assert.Contains("410", report.Message);
+
+ var action = Assert.Single(rig.Dispatcher.Dispatched);
+ Assert.Equal(ServerActionTypes.TerminateSession, action.Type);
+ Assert.Contains("410", action.Reason);
+ Assert.NotNull(action.Params);
+ var parameters = action.Params!;
+ Assert.Equal("true", parameters["forceSelfDestruct"]);
+ Assert.Equal("0", parameters["gracePeriodSeconds"]);
+ Assert.Equal(TerminationOrigins.SessionGone, parameters["origin"]);
+ }
+
+ [Theory]
+ [InlineData(true, false)]
+ [InlineData(false, true)]
+ public void Does_nothing_when_a_shutdown_is_already_in_progress(bool shutdownSet, bool shutdownCompleteSet)
+ {
+ using var rig = new Rig();
+ if (shutdownSet) rig.Shutdown.Set();
+ if (shutdownCompleteSet) rig.ShutdownComplete.Set();
+
+ rig.Handle();
+
+ Assert.Empty(rig.Reporter.Reports);
+ Assert.Empty(rig.Dispatcher.Dispatched);
+ }
+
+ [Fact]
+ public void ReadOrigin_returns_the_origin_param_or_null()
+ {
+ Assert.Equal(TerminationOrigins.SessionGone,
+ global::AutopilotMonitor.Agent.V2.Runtime.ServerControlPlane.ReadOrigin(
+ new Dictionary { ["origin"] = TerminationOrigins.SessionGone }));
+ Assert.Equal(TerminationOrigins.KillSignal,
+ global::AutopilotMonitor.Agent.V2.Runtime.ServerControlPlane.ReadOrigin(
+ new Dictionary { ["origin"] = TerminationOrigins.KillSignal, ["forceSelfDestruct"] = "true" }));
+ Assert.Null(global::AutopilotMonitor.Agent.V2.Runtime.ServerControlPlane.ReadOrigin(
+ new Dictionary { ["adminOutcome"] = "Failed" }));
+ Assert.Null(global::AutopilotMonitor.Agent.V2.Runtime.ServerControlPlane.ReadOrigin(
+ new Dictionary { ["origin"] = string.Empty }));
+ Assert.Null(global::AutopilotMonitor.Agent.V2.Runtime.ServerControlPlane.ReadOrigin(null!));
+ }
+ }
+}
diff --git a/src/Agent/AutopilotMonitor.Agent.V2.Core.Tests/Termination/EnrollmentTerminationHandlerTests.cs b/src/Agent/AutopilotMonitor.Agent.V2.Core.Tests/Termination/EnrollmentTerminationHandlerTests.cs
index 25b06b5f..aa8b3b86 100644
--- a/src/Agent/AutopilotMonitor.Agent.V2.Core.Tests/Termination/EnrollmentTerminationHandlerTests.cs
+++ b/src/Agent/AutopilotMonitor.Agent.V2.Core.Tests/Termination/EnrollmentTerminationHandlerTests.cs
@@ -310,6 +310,19 @@ namespace AutopilotMonitor.Agent.V2.Core.Tests.Termination
new EnrollmentTerminatedEventArgs(
reason, outcome, stage.ToString(), at ?? EndUtc);
+ ///
+ /// The args ServerControlPlane.OnTerminateRequested synthesises for a
+ /// terminate_session action tagged with .
+ ///
+ private static EnrollmentTerminatedEventArgs ServerArgs(string origin) =>
+ new EnrollmentTerminatedEventArgs(
+ EnrollmentTerminationReason.DecisionTerminalStage,
+ EnrollmentTerminationOutcome.Failed,
+ SessionStage.Failed.ToString(),
+ EndUtc,
+ details: "Server-requested termination",
+ origin: origin);
+
[Fact]
public void Handle_writes_final_status_json_in_state_directory()
{
@@ -458,6 +471,52 @@ namespace AutopilotMonitor.Agent.V2.Core.Tests.Termination
Assert.Equal(0, rig.CleanupService.Invocations);
}
+ // ============================================================= origin=session_gone (HTTP 410)
+
+ [Fact]
+ public void Handle_session_gone_skips_spool_drain_and_diagnostics_but_still_cleans_up()
+ {
+ // The backend no longer knows the session: every drain would only burn its budget on
+ // 410s and a diagnostics package would describe a deleted session. Marker, cleanup
+ // and the (forced) self-destruct still run.
+ using var rig = new Rig();
+ rig.State = new DecisionStateBuilder(DecisionState.CreateInitial("S1", "T1")) { Stage = SessionStage.Failed }.Build();
+ rig.SpoolDrainPeriodOverride = TimeSpan.FromSeconds(5);
+ var spoolPolls = 0;
+ rig.PendingItemCountAccessor = () => { spoolPolls++; return 5; }; // never drains — a poll would burn the budget
+ var cfg = rig.BuildConfig(diagEnabled: true, diagMode: "Always", selfDestruct: true);
+
+ rig.Build(cfg).Handle(sender: null!, ServerArgs(TerminationOrigins.SessionGone));
+
+ Assert.Equal(0, spoolPolls);
+ Assert.Equal(0, rig.DiagnosticsUploads);
+ Assert.DoesNotContain(Constants.EventTypes.DiagnosticsCollecting, rig.EmittedEventTypes);
+ Assert.Contains(Constants.EventTypes.AgentShuttingDown, rig.EmittedEventTypes);
+ Assert.True(File.Exists(Path.Combine(rig.StateDir, "enrollment-complete.marker")));
+ Assert.Equal(1, rig.CleanupService.Invocations);
+ Assert.Equal(1, rig.ShutdownSignalled);
+ }
+
+ [Fact]
+ public void Handle_kill_signal_origin_still_drains_spool_and_uploads_diagnostics()
+ {
+ // The administrator kill signal keeps its full sequence — only session_gone skips.
+ using var rig = new Rig();
+ rig.State = new DecisionStateBuilder(DecisionState.CreateInitial("S1", "T1")) { Stage = SessionStage.Failed }.Build();
+ rig.SpoolDrainPeriodOverride = TimeSpan.FromSeconds(5);
+ var spoolPolls = 0;
+ rig.PendingItemCountAccessor = () => { spoolPolls++; return 0; };
+ var cfg = rig.BuildConfig(diagEnabled: true, diagMode: "Always", selfDestruct: true);
+
+ rig.Build(cfg).Handle(sender: null!, ServerArgs(TerminationOrigins.KillSignal));
+
+ Assert.True(spoolPolls > 0);
+ Assert.Equal(1, rig.DiagnosticsUploads);
+ Assert.Contains(Constants.EventTypes.DiagnosticsCollecting, rig.EmittedEventTypes);
+ Assert.Equal(1, rig.CleanupService.Invocations);
+ Assert.Equal(1, rig.ShutdownSignalled);
+ }
+
[Fact]
public void Handle_signals_shutdown_exactly_once_even_on_error()
{
diff --git a/src/Agent/AutopilotMonitor.Agent.V2.Core.Tests/Transport/BackendTelemetryUploaderTests.cs b/src/Agent/AutopilotMonitor.Agent.V2.Core.Tests/Transport/BackendTelemetryUploaderTests.cs
index 53e3b84c..32cf0386 100644
--- a/src/Agent/AutopilotMonitor.Agent.V2.Core.Tests/Transport/BackendTelemetryUploaderTests.cs
+++ b/src/Agent/AutopilotMonitor.Agent.V2.Core.Tests/Transport/BackendTelemetryUploaderTests.cs
@@ -275,9 +275,44 @@ namespace AutopilotMonitor.Agent.V2.Core.Tests.Transport
Assert.False(result.Success);
Assert.False(result.IsTransient);
+ Assert.False(result.IsSessionGone);
Assert.Contains(((int)status).ToString(), result.ErrorReason);
}
+ [Fact]
+ public async Task Gone_410_maps_to_SessionGone()
+ {
+ // 410 = the backend no longer knows this session. Not transient (no retry), not auth,
+ // not poison, not a size problem — a distinct terminal kind the drain reacts to.
+ using var rig = new Rig();
+ rig.Handler.QueueStatus(HttpStatusCode.Gone, body: "session unknown");
+
+ var result = await rig.Sut.UploadBatchAsync(new[] { NewEventItem(1) }, CancellationToken.None);
+
+ Assert.False(result.Success);
+ Assert.True(result.IsSessionGone);
+ Assert.False(result.IsTransient);
+ Assert.False(result.IsAuthFailure);
+ Assert.False(result.IsPoison);
+ Assert.False(result.RequiresSplit);
+ Assert.Contains("410", result.ErrorReason);
+ }
+
+ [Fact]
+ public async Task Gone_410_with_poison_body_is_still_SessionGone_never_poison()
+ {
+ // A poison body on 410 must not be honoured: dropping items + emitting
+ // telemetry_upload_poisoned would only produce the next 410 (emit/reject loop).
+ using var rig = new Rig();
+ rig.Handler.QueueStatus(HttpStatusCode.Gone, body: "{\"poison\":true,\"rejectedRowKeys\":[\"rk1\"],\"reason\":\"x\"}");
+
+ var result = await rig.Sut.UploadBatchAsync(new[] { NewEventItem(1) }, CancellationToken.None);
+
+ Assert.True(result.IsSessionGone);
+ Assert.False(result.IsPoison);
+ Assert.Null(result.PoisonRowKeys);
+ }
+
[Theory]
[InlineData(HttpStatusCode.RequestTimeout)] // 408
[InlineData((HttpStatusCode)429)] // TooManyRequests
diff --git a/src/Agent/AutopilotMonitor.Agent.V2.Core.Tests/Transport/FakeBackendTelemetryUploader.cs b/src/Agent/AutopilotMonitor.Agent.V2.Core.Tests/Transport/FakeBackendTelemetryUploader.cs
index 8d959548..e76601ef 100644
--- a/src/Agent/AutopilotMonitor.Agent.V2.Core.Tests/Transport/FakeBackendTelemetryUploader.cs
+++ b/src/Agent/AutopilotMonitor.Agent.V2.Core.Tests/Transport/FakeBackendTelemetryUploader.cs
@@ -68,6 +68,12 @@ namespace AutopilotMonitor.Agent.V2.Core.Tests.Transport
return this;
}
+ public FakeBackendTelemetryUploader QueueSessionGone(string reason = "session gone: http 410")
+ {
+ _script.Enqueue(_ => UploadResult.SessionGone(reason));
+ return this;
+ }
+
public FakeBackendTelemetryUploader QueueThrow(Exception ex)
{
_script.Enqueue(_ => throw ex);
diff --git a/src/Agent/AutopilotMonitor.Agent.V2.Core.Tests/Transport/TelemetrySpoolTests.cs b/src/Agent/AutopilotMonitor.Agent.V2.Core.Tests/Transport/TelemetrySpoolTests.cs
index 66af48bb..761e044d 100644
--- a/src/Agent/AutopilotMonitor.Agent.V2.Core.Tests/Transport/TelemetrySpoolTests.cs
+++ b/src/Agent/AutopilotMonitor.Agent.V2.Core.Tests/Transport/TelemetrySpoolTests.cs
@@ -10,10 +10,15 @@ namespace AutopilotMonitor.Agent.V2.Core.Tests.Transport
{
public sealed class TelemetrySpoolTests
{
- private static TelemetryItemDraft EventDraft(string rowKey = "row-1") =>
+ private const string OwnKey = "tenant_session";
+ private const string ForeignKey = "tenant_old-session";
+
+ private static TelemetryItemDraft EventDraft(string rowKey = "row-1") => EventDraft(rowKey, OwnKey);
+
+ private static TelemetryItemDraft EventDraft(string rowKey, string partitionKey) =>
new TelemetryItemDraft(
kind: TelemetryItemKind.Event,
- partitionKey: "tenant_session",
+ partitionKey: partitionKey,
rowKey: rowKey,
payloadJson: "{\"EventType\":\"agent_started\"}",
isSessionScoped: true);
@@ -195,6 +200,86 @@ namespace AutopilotMonitor.Agent.V2.Core.Tests.Transport
Assert.Equal(1, next.TelemetryItemId);
}
+ // ============================================================ Startup partition-key filter
+
+ [Fact]
+ public void Ctor_drops_leading_pending_items_of_another_session_and_persists_cursor()
+ {
+ // Same spool directory, new session id: the previous session's unsent items would only
+ // draw a 410 that must not end the new session. They are dropped on startup; the own
+ // item behind them stays pending.
+ using var tmp = new TempDirectory();
+ var s1 = new TelemetrySpool(tmp.Path, Clock());
+ s1.Enqueue(EventDraft("old-a", ForeignKey));
+ s1.Enqueue(EventDraft("old-b", ForeignKey));
+ s1.Enqueue(EventDraft("own", OwnKey));
+
+ var s2 = new TelemetrySpool(tmp.Path, Clock(), expectedPartitionKey: OwnKey);
+
+ Assert.Equal(1, s2.LastUploadedItemId);
+ Assert.Equal(2, s2.LastAssignedItemId);
+ Assert.Equal(1, s2.PendingItemCount);
+ Assert.Equal("own", Assert.Single(s2.Peek(10)).RowKey);
+
+ // Cursor persisted: a later start without the filter sees the same tail with the same bytes.
+ var s3 = new TelemetrySpool(tmp.Path, Clock());
+ Assert.Equal(1, s3.LastUploadedItemId);
+ Assert.Equal(1, s3.PendingItemCount);
+ Assert.Equal(s3.PendingBytes, s2.PendingBytes);
+
+ var next = s2.Enqueue(EventDraft("next", OwnKey));
+ Assert.Equal(3, next.TelemetryItemId);
+ }
+
+ [Fact]
+ public void Ctor_drops_nothing_when_a_foreign_item_follows_an_own_item()
+ {
+ // Impossible by construction (one process = one session); if it happens the order is
+ // not trustworthy, so the filter keeps everything instead of guessing.
+ using var tmp = new TempDirectory();
+ var s1 = new TelemetrySpool(tmp.Path, Clock());
+ s1.Enqueue(EventDraft("old-a", ForeignKey));
+ s1.Enqueue(EventDraft("own", OwnKey));
+ s1.Enqueue(EventDraft("old-b", ForeignKey));
+
+ var s2 = new TelemetrySpool(tmp.Path, Clock(), expectedPartitionKey: OwnKey);
+
+ Assert.Equal(-1, s2.LastUploadedItemId);
+ Assert.Equal(3, s2.PendingItemCount);
+ Assert.Equal(s1.PendingBytes, s2.PendingBytes);
+ }
+
+ [Fact]
+ public void Ctor_keeps_pending_items_that_carry_the_expected_key()
+ {
+ // WhiteGlove Part 2 reuses the Part-1 session id: its leftovers carry the expected key.
+ using var tmp = new TempDirectory();
+ var s1 = new TelemetrySpool(tmp.Path, Clock());
+ s1.Enqueue(EventDraft("a", OwnKey));
+ s1.Enqueue(EventDraft("b", OwnKey));
+ s1.Enqueue(EventDraft("c", OwnKey));
+
+ var s2 = new TelemetrySpool(tmp.Path, Clock(), expectedPartitionKey: OwnKey);
+
+ Assert.Equal(-1, s2.LastUploadedItemId);
+ Assert.Equal(3, s2.PendingItemCount);
+ Assert.Equal(s1.PendingBytes, s2.PendingBytes);
+ }
+
+ [Fact]
+ public void Ctor_without_expected_key_keeps_foreign_items()
+ {
+ using var tmp = new TempDirectory();
+ var s1 = new TelemetrySpool(tmp.Path, Clock());
+ s1.Enqueue(EventDraft("old-a", ForeignKey));
+ s1.Enqueue(EventDraft("old-b", ForeignKey));
+
+ var s2 = new TelemetrySpool(tmp.Path, Clock());
+
+ Assert.Equal(-1, s2.LastUploadedItemId);
+ Assert.Equal(2, s2.PendingItemCount);
+ }
+
[Fact]
public void Enqueue_with_RequiresImmediateFlush_raises_ImmediateFlushRequested()
{
diff --git a/src/Agent/AutopilotMonitor.Agent.V2.Core.Tests/Transport/TelemetryUploadOrchestratorTests.cs b/src/Agent/AutopilotMonitor.Agent.V2.Core.Tests/Transport/TelemetryUploadOrchestratorTests.cs
index 1a61f1b1..2651acc6 100644
--- a/src/Agent/AutopilotMonitor.Agent.V2.Core.Tests/Transport/TelemetryUploadOrchestratorTests.cs
+++ b/src/Agent/AutopilotMonitor.Agent.V2.Core.Tests/Transport/TelemetryUploadOrchestratorTests.cs
@@ -10,10 +10,14 @@ namespace AutopilotMonitor.Agent.V2.Core.Tests.Transport
{
public sealed class TelemetryUploadOrchestratorTests
{
- private static TelemetryItemDraft Draft(string rowKey) =>
+ private const string OwnKey = "tenant_session";
+
+ private static TelemetryItemDraft Draft(string rowKey) => Draft(rowKey, OwnKey);
+
+ private static TelemetryItemDraft Draft(string rowKey, string partitionKey) =>
new TelemetryItemDraft(
kind: TelemetryItemKind.Event,
- partitionKey: "tenant_session",
+ partitionKey: partitionKey,
rowKey: rowKey,
payloadJson: $"{{\"rk\":\"{rowKey}\"}}",
isSessionScoped: true);
@@ -157,6 +161,118 @@ namespace AutopilotMonitor.Agent.V2.Core.Tests.Transport
Assert.Equal(1, blockedCount);
}
+ // ============================================================ HTTP 410 — session gone
+
+ [Fact]
+ public async Task SessionGone_for_current_session_stops_drain_retains_cursor_and_raises_once()
+ {
+ using var tmp = new TempDirectory();
+ var spool = new TelemetrySpool(tmp.Path, Clock());
+ spool.Enqueue(Draft("a"));
+ spool.Enqueue(Draft("b"));
+
+ var uploader = new FakeBackendTelemetryUploader().QueueSessionGone().QueueSessionGone();
+ using var sut = new TelemetryUploadOrchestrator(spool, uploader, Clock(), currentPartitionKey: OwnKey);
+ var gone = new System.Collections.Generic.List();
+ int blockedCount = 0;
+ PoisonReport? poison = null;
+ sut.SessionGone += (_, reason) => gone.Add(reason);
+ sut.UploadBlocked += (_, __) => blockedCount++;
+ sut.BatchPoisoned += (_, p) => poison = p;
+
+ var first = await sut.DrainAllAsync();
+
+ Assert.False(first.Success);
+ Assert.Equal(1, first.FailedBatches);
+ Assert.Equal(0, first.UploadedItems);
+ Assert.Equal(1, uploader.CallCount); // no retries — nothing can make a 410 succeed
+ Assert.Equal(-1, sut.LastUploadedItemId); // retained — items stay on disk for the diag ZIP
+ Assert.Single(gone);
+ Assert.Contains("410", gone[0]);
+ Assert.Equal(0, blockedCount); // not a "blocked" pipeline
+ Assert.Null(poison);
+
+ // One-shot: the next drain hits the 410 again but does not raise again.
+ var second = await sut.DrainAllAsync();
+ Assert.False(second.Success);
+ Assert.Equal(2, uploader.CallCount);
+ Assert.Equal(-1, sut.LastUploadedItemId);
+ Assert.Single(gone);
+ Assert.Equal(0, blockedCount);
+ }
+
+ [Fact]
+ public async Task SessionGone_for_leading_foreign_items_drops_only_them_and_continues()
+ {
+ // Leftovers of a previous session id at the head of the spool (defence in depth — the
+ // spool filters these at startup). A 410 for them must not end the current session:
+ // drop exactly the leading foreign run, then carry on with the current session's items.
+ using var tmp = new TempDirectory();
+ var spool = new TelemetrySpool(tmp.Path, Clock());
+ spool.Enqueue(Draft("old-a", "tenant_old-session"));
+ spool.Enqueue(Draft("old-b", "tenant_old-session"));
+ spool.Enqueue(Draft("new", OwnKey));
+
+ var uploader = new FakeBackendTelemetryUploader().QueueSessionGone().QueueOk();
+ using var sut = new TelemetryUploadOrchestrator(spool, uploader, Clock(), currentPartitionKey: OwnKey);
+ int goneCount = 0, blockedCount = 0;
+ sut.SessionGone += (_, __) => goneCount++;
+ sut.UploadBlocked += (_, __) => blockedCount++;
+
+ var result = await sut.DrainAllAsync();
+
+ Assert.True(result.Success);
+ Assert.Equal(0, result.FailedBatches);
+ Assert.Equal(1, result.UploadedItems);
+ Assert.Equal(2, uploader.CallCount);
+ Assert.Equal(3, uploader.Received[0].Count); // first batch carried all three
+ Assert.Equal("new", Assert.Single(uploader.Received[1]).RowKey); // retry carried only the own item
+ Assert.Equal(2, sut.LastUploadedItemId);
+ Assert.Equal(0, goneCount);
+ Assert.Equal(0, blockedCount);
+ }
+
+ [Fact]
+ public async Task SessionGone_is_not_latched_before_a_handler_is_subscribed()
+ {
+ // The host subscribes after EnrollmentOrchestrator.Start; a 410 on the very first drain
+ // must still reach it on the next drain instead of being swallowed by the one-shot guard.
+ using var tmp = new TempDirectory();
+ var spool = new TelemetrySpool(tmp.Path, Clock());
+ spool.Enqueue(Draft("only"));
+
+ var uploader = new FakeBackendTelemetryUploader().QueueSessionGone().QueueSessionGone();
+ using var sut = new TelemetryUploadOrchestrator(spool, uploader, Clock(), currentPartitionKey: OwnKey);
+
+ await sut.DrainAllAsync(); // nobody listening yet
+
+ int goneCount = 0;
+ sut.SessionGone += (_, __) => goneCount++;
+ await sut.DrainAllAsync();
+
+ Assert.Equal(1, goneCount);
+ Assert.Equal(-1, sut.LastUploadedItemId);
+ }
+
+ [Fact]
+ public async Task SessionGone_without_a_current_key_treats_every_batch_as_own()
+ {
+ using var tmp = new TempDirectory();
+ var spool = new TelemetrySpool(tmp.Path, Clock());
+ spool.Enqueue(Draft("x", "tenant_whatever"));
+
+ var uploader = new FakeBackendTelemetryUploader().QueueSessionGone();
+ using var sut = new TelemetryUploadOrchestrator(spool, uploader, Clock());
+ int goneCount = 0;
+ sut.SessionGone += (_, __) => goneCount++;
+
+ var result = await sut.DrainAllAsync();
+
+ Assert.False(result.Success);
+ Assert.Equal(1, goneCount);
+ Assert.Equal(-1, sut.LastUploadedItemId);
+ }
+
[Fact]
public async Task TooLarge_splits_batch_and_uploads_everything()
{
diff --git a/src/Agent/AutopilotMonitor.Agent.V2.Core/Monitoring/Transport/EmergencyReporter.cs b/src/Agent/AutopilotMonitor.Agent.V2.Core/Monitoring/Transport/EmergencyReporter.cs
index 19d88456..a884aa57 100644
--- a/src/Agent/AutopilotMonitor.Agent.V2.Core/Monitoring/Transport/EmergencyReporter.cs
+++ b/src/Agent/AutopilotMonitor.Agent.V2.Core/Monitoring/Transport/EmergencyReporter.cs
@@ -2,6 +2,7 @@ using System;
using System.Collections.Generic;
using System.Threading.Tasks;
using AutopilotMonitor.Agent.V2.Core.Logging;
+using AutopilotMonitor.DecisionCore.Engine;
using AutopilotMonitor.Shared.Models;
namespace AutopilotMonitor.Agent.V2.Core.Monitoring.Transport
@@ -14,6 +15,7 @@ namespace AutopilotMonitor.Agent.V2.Core.Monitoring.Transport
/// - Same error key (ErrorType + HTTP status code) is only ever sent once per session
/// - At most emergency reports total per session
/// - At most one report every minutes regardless of error type
+ /// (the terminal report is exempt)
///
/// A failure in this reporter is silently swallowed — it must never cascade into the main loop.
///
@@ -32,6 +34,7 @@ namespace AutopilotMonitor.Agent.V2.Core.Monitoring.Transport
private readonly string _tenantId;
private readonly string _agentVersion;
private readonly AgentLogger _logger;
+ private readonly IClock _clock;
private readonly HashSet _reportedKeys = new HashSet();
private readonly object _antiFloodLock = new object();
@@ -43,13 +46,15 @@ namespace AutopilotMonitor.Agent.V2.Core.Monitoring.Transport
string sessionId,
string tenantId,
string agentVersion,
- AgentLogger logger)
+ AgentLogger logger,
+ IClock clock = null)
{
_apiClient = apiClient;
_sessionId = sessionId;
_tenantId = tenantId;
_agentVersion = agentVersion;
_logger = logger;
+ _clock = clock ?? SystemClock.Instance;
}
///
@@ -100,6 +105,7 @@ namespace AutopilotMonitor.Agent.V2.Core.Monitoring.Transport
// All anti-flood checks are protected by a lock to prevent race conditions when
// TrySendAsync is called concurrently (fire-and-forget from multiple threads).
var key = $"{errorType}:{httpStatusCode}";
+ var now = _clock.UtcNow;
int currentReportCount;
DateTime? previousLastReportTime;
@@ -117,8 +123,13 @@ namespace AutopilotMonitor.Agent.V2.Core.Monitoring.Transport
return;
}
- if (_lastReportTime != null &&
- (DateTime.UtcNow - _lastReportTime.Value).TotalMinutes < MinIntervalMinutes)
+ // The session-unknown report is terminal and sent at most once per process (the
+ // termination follows immediately): a routine report minutes earlier must not cost
+ // the last word about the session. Only the cooldown is waived; key dedup and the
+ // per-session maximum still apply.
+ var exemptFromCooldown = errorType == AgentErrorType.TelemetryRejectedSessionUnknown;
+ if (!exemptFromCooldown && _lastReportTime != null &&
+ (now - _lastReportTime.Value).TotalMinutes < MinIntervalMinutes)
{
_logger?.Debug($"[EmergencyChannel] Suppressed (cooldown active, last report at {_lastReportTime.Value:HH:mm:ss}): {key}");
return;
@@ -127,7 +138,7 @@ namespace AutopilotMonitor.Agent.V2.Core.Monitoring.Transport
previousLastReportTime = _lastReportTime;
_reportedKeys.Add(key);
_reportCount++;
- _lastReportTime = DateTime.UtcNow;
+ _lastReportTime = now;
currentReportCount = _reportCount;
}
@@ -145,7 +156,7 @@ namespace AutopilotMonitor.Agent.V2.Core.Monitoring.Transport
SessionAgeHours = sessionAgeHours,
PriorRegistrationFailure = priorRegistrationFailure,
AgentVersion = _agentVersion,
- Timestamp = DateTime.UtcNow
+ Timestamp = now
};
var delivered = false;
diff --git a/src/Agent/AutopilotMonitor.Agent.V2.Core/Orchestration/EnrollmentOrchestrator.cs b/src/Agent/AutopilotMonitor.Agent.V2.Core/Orchestration/EnrollmentOrchestrator.cs
index c4c6c4b5..89da3464 100644
--- a/src/Agent/AutopilotMonitor.Agent.V2.Core/Orchestration/EnrollmentOrchestrator.cs
+++ b/src/Agent/AutopilotMonitor.Agent.V2.Core/Orchestration/EnrollmentOrchestrator.cs
@@ -371,8 +371,13 @@ namespace AutopilotMonitor.Agent.V2.Core.Orchestration
// 3) Telemetry-Transport. Batch size flows from AgentConfiguration.MaxBatchSize
// via Program.cs (P1 fix: previously the remote-config knob was merged but
// never applied, so tenants saw drainInterval/MaxBatchSize have no effect).
- _spool = new TelemetrySpool(_transportDirectory, _clock, _logger);
- _transport = new TelemetryUploadOrchestrator(_spool, _uploader, _clock, batchSize: _uploadBatchSize);
+ // The partition key names the registered session: the spool drops leftovers of a
+ // previous session id on startup, and the orchestrator tells a 410 for this session
+ // (SessionGone → host terminates) from a 410 for such leftovers (dropped).
+ var partitionKey = TelemetryPartitionKey.ForSession(_tenantId, _sessionId);
+ _spool = new TelemetrySpool(_transportDirectory, _clock, _logger, expectedPartitionKey: partitionKey);
+ _transport = new TelemetryUploadOrchestrator(
+ _spool, _uploader, _clock, batchSize: _uploadBatchSize, currentPartitionKey: partitionKey, logger: _logger);
// TRACE-H1 — surface skipped oversized items + blocked-pipeline conditions as timeline
// events. Raised after the drain guard releases, so posting back through the ingress
diff --git a/src/Agent/AutopilotMonitor.Agent.V2.Core/Orchestration/EnrollmentTerminatedEventArgs.cs b/src/Agent/AutopilotMonitor.Agent.V2.Core/Orchestration/EnrollmentTerminatedEventArgs.cs
index f8ae9c2e..9a42139e 100644
--- a/src/Agent/AutopilotMonitor.Agent.V2.Core/Orchestration/EnrollmentTerminatedEventArgs.cs
+++ b/src/Agent/AutopilotMonitor.Agent.V2.Core/Orchestration/EnrollmentTerminatedEventArgs.cs
@@ -73,13 +73,23 @@ namespace AutopilotMonitor.Agent.V2.Core.Orchestration
///
public string? StopReason { get; }
+ ///
+ /// The origin param of a server-requested terminate_session action
+ /// (); null for every termination the agent decided
+ /// itself. makes the termination handler
+ /// skip the spool drain and the diagnostics upload — the backend no longer knows the
+ /// session, so both could only fail.
+ ///
+ public string? Origin { get; }
+
public EnrollmentTerminatedEventArgs(
EnrollmentTerminationReason reason,
EnrollmentTerminationOutcome outcome,
string? stageName,
DateTime terminatedAtUtc,
string? details = null,
- string? stopReason = null)
+ string? stopReason = null,
+ string? origin = null)
{
Reason = reason;
Outcome = outcome;
@@ -87,6 +97,20 @@ namespace AutopilotMonitor.Agent.V2.Core.Orchestration
TerminatedAtUtc = terminatedAtUtc;
Details = details;
StopReason = stopReason;
+ Origin = origin;
}
}
+
+ ///
+ /// Values of the origin param on a synthesised terminate_session
+ /// ServerAction, carried into .
+ ///
+ public static class TerminationOrigins
+ {
+ /// Administrator kill signal (DeviceKillSignal on a 2xx telemetry response).
+ public const string KillSignal = "kill_signal";
+
+ /// The backend answered HTTP 410 for the current session's telemetry.
+ public const string SessionGone = "session_gone";
+ }
}
diff --git a/src/Agent/AutopilotMonitor.Agent.V2.Core/Telemetry/Events/TelemetryEventEmitter.cs b/src/Agent/AutopilotMonitor.Agent.V2.Core/Telemetry/Events/TelemetryEventEmitter.cs
index faa08123..83119f4f 100644
--- a/src/Agent/AutopilotMonitor.Agent.V2.Core/Telemetry/Events/TelemetryEventEmitter.cs
+++ b/src/Agent/AutopilotMonitor.Agent.V2.Core/Telemetry/Events/TelemetryEventEmitter.cs
@@ -45,7 +45,7 @@ namespace AutopilotMonitor.Agent.V2.Core.Telemetry.Events
_sequenceCounter = sequenceCounter ?? throw new ArgumentNullException(nameof(sequenceCounter));
_sessionId = sessionId;
_tenantId = tenantId;
- _partitionKey = $"{tenantId}_{sessionId}";
+ _partitionKey = TelemetryPartitionKey.ForSession(tenantId, sessionId);
_traceEventsEnabled = traceEventsEnabled;
}
diff --git a/src/Agent/AutopilotMonitor.Agent.V2.Core/Telemetry/Signals/TelemetrySignalEmitter.cs b/src/Agent/AutopilotMonitor.Agent.V2.Core/Telemetry/Signals/TelemetrySignalEmitter.cs
index b6282a43..c267b4b9 100644
--- a/src/Agent/AutopilotMonitor.Agent.V2.Core/Telemetry/Signals/TelemetrySignalEmitter.cs
+++ b/src/Agent/AutopilotMonitor.Agent.V2.Core/Telemetry/Signals/TelemetrySignalEmitter.cs
@@ -42,7 +42,7 @@ namespace AutopilotMonitor.Agent.V2.Core.Telemetry.Signals
if (string.IsNullOrEmpty(sessionId)) throw new ArgumentException("SessionId is mandatory.", nameof(sessionId));
if (string.IsNullOrEmpty(tenantId)) throw new ArgumentException("TenantId is mandatory.", nameof(tenantId));
_transport = transport ?? throw new ArgumentNullException(nameof(transport));
- _partitionKey = $"{tenantId}_{sessionId}";
+ _partitionKey = TelemetryPartitionKey.ForSession(tenantId, sessionId);
}
///
diff --git a/src/Agent/AutopilotMonitor.Agent.V2.Core/Telemetry/Transitions/TelemetryTransitionEmitter.cs b/src/Agent/AutopilotMonitor.Agent.V2.Core/Telemetry/Transitions/TelemetryTransitionEmitter.cs
index 137b1c48..d215bbf9 100644
--- a/src/Agent/AutopilotMonitor.Agent.V2.Core/Telemetry/Transitions/TelemetryTransitionEmitter.cs
+++ b/src/Agent/AutopilotMonitor.Agent.V2.Core/Telemetry/Transitions/TelemetryTransitionEmitter.cs
@@ -43,7 +43,7 @@ namespace AutopilotMonitor.Agent.V2.Core.Telemetry.Transitions
if (string.IsNullOrEmpty(sessionId)) throw new ArgumentException("SessionId is mandatory.", nameof(sessionId));
if (string.IsNullOrEmpty(tenantId)) throw new ArgumentException("TenantId is mandatory.", nameof(tenantId));
_transport = transport ?? throw new ArgumentNullException(nameof(transport));
- _partitionKey = $"{tenantId}_{sessionId}";
+ _partitionKey = TelemetryPartitionKey.ForSession(tenantId, sessionId);
}
///
diff --git a/src/Agent/AutopilotMonitor.Agent.V2.Core/Termination/EnrollmentTerminationHandler.cs b/src/Agent/AutopilotMonitor.Agent.V2.Core/Termination/EnrollmentTerminationHandler.cs
index df5c869f..073631f3 100644
--- a/src/Agent/AutopilotMonitor.Agent.V2.Core/Termination/EnrollmentTerminationHandler.cs
+++ b/src/Agent/AutopilotMonitor.Agent.V2.Core/Termination/EnrollmentTerminationHandler.cs
@@ -306,7 +306,7 @@ namespace AutopilotMonitor.Agent.V2.Core.Termination
});
DelayLateEventGrace();
- DrainSpool();
+ DrainSpool(args);
// Option 2 (WG Part 1 graceful-exit hardening, 2026-04-30): write the
// clean-exit marker BEFORE _signalShutdown returns control to the main
@@ -329,7 +329,7 @@ namespace AutopilotMonitor.Agent.V2.Core.Termination
RunUploadDiagnosticsWithEvents(args);
WriteEnrollmentCompleteMarker(args);
- RunStandaloneRebootIfRequested();
+ RunStandaloneRebootIfRequested(args);
RunSelfDestructIfAppropriate(args);
// Option 2 (same hardening) — covers the standard Completed / Failed terminal
// path too. Cleanup PowerShell does not touch this marker, so writing it before
@@ -979,6 +979,14 @@ namespace AutopilotMonitor.Agent.V2.Core.Termination
///
private void RunUploadDiagnosticsWithEvents(EnrollmentTerminatedEventArgs args, string suffixOverride = null)
{
+ if (IsSessionGone(args))
+ {
+ // The backend answered 410 for this session: the package would describe a session
+ // that no longer exists, and its diagnostics_* events could never be accepted.
+ _logger.Info("EnrollmentTerminationHandler: diagnostics upload skipped — backend no longer knows this session.");
+ return;
+ }
+
var mode = _configuration.DiagnosticsUploadMode ?? "Off";
if (!_configuration.DiagnosticsUploadEnabled || string.Equals(mode, "Off", StringComparison.OrdinalIgnoreCase))
{
@@ -1124,7 +1132,7 @@ namespace AutopilotMonitor.Agent.V2.Core.Termination
/// but enables RebootOnComplete, the agent's final act is shutdown.exe /r
/// with the configured delay, giving the user a visible countdown.
///
- private void RunStandaloneRebootIfRequested()
+ private void RunStandaloneRebootIfRequested(EnrollmentTerminatedEventArgs args)
{
if (_configuration.SelfDestructOnComplete) return;
if (!_configuration.RebootOnComplete) return;
@@ -1148,7 +1156,7 @@ namespace AutopilotMonitor.Agent.V2.Core.Termination
ImmediateUpload = true,
});
- DrainSpool();
+ DrainSpool(args);
try
{
@@ -1185,7 +1193,7 @@ namespace AutopilotMonitor.Agent.V2.Core.Termination
// just above. The WG-Part-1 and standalone-reboot paths already drain first; this
// closes the gap on the self-destruct path. Bounded + best-effort (never blocks
// termination on a wedged backend).
- DrainSpool();
+ DrainSpool(args);
var service = _cleanupServiceFactory();
service.ExecuteSelfDestruct();
@@ -1318,8 +1326,23 @@ namespace AutopilotMonitor.Agent.V2.Core.Termination
}
}
- private void DrainSpool()
+ ///
+ /// : the backend no longer knows the session.
+ /// Every drain would only collect more 410s against a bounded budget, and a diagnostics
+ /// package would describe a deleted session — both are skipped; marker, cleanup and
+ /// self-destruct run as usual.
+ ///
+ private static bool IsSessionGone(EnrollmentTerminatedEventArgs args) =>
+ string.Equals(args.Origin, TerminationOrigins.SessionGone, StringComparison.Ordinal);
+
+ private void DrainSpool(EnrollmentTerminatedEventArgs args)
{
+ if (IsSessionGone(args))
+ {
+ _logger.Info("EnrollmentTerminationHandler: spool drain skipped — backend no longer knows this session.");
+ return;
+ }
+
// Block briefly so pending events can land before the next destructive step
// (shutdown.exe, self-destruct). Two phases share the same bounded budget:
//
diff --git a/src/Agent/AutopilotMonitor.Agent.V2.Core/Transport/Telemetry/BackendTelemetryUploader.cs b/src/Agent/AutopilotMonitor.Agent.V2.Core/Transport/Telemetry/BackendTelemetryUploader.cs
index ecf102d6..ff7b4590 100644
--- a/src/Agent/AutopilotMonitor.Agent.V2.Core/Transport/Telemetry/BackendTelemetryUploader.cs
+++ b/src/Agent/AutopilotMonitor.Agent.V2.Core/Transport/Telemetry/BackendTelemetryUploader.cs
@@ -165,6 +165,10 @@ namespace AutopilotMonitor.Agent.V2.Core.Transport.Telemetry
{
_logger?.Warning($"BackendTelemetryUploader: ingest TRANSIENT (corr={correlationId}, items={items.Count}, durationMs={sw.ElapsedMilliseconds}, status={(int)response.StatusCode}): {result.ErrorReason}");
}
+ else if (result.IsSessionGone)
+ {
+ _logger?.Error($"BackendTelemetryUploader: ingest rejected — backend no longer knows this session (corr={correlationId}, items={items.Count}, durationMs={sw.ElapsedMilliseconds}, status={(int)response.StatusCode}): {result.ErrorReason}");
+ }
else
{
_logger?.Error($"BackendTelemetryUploader: ingest PERMANENT fail (corr={correlationId}, items={items.Count}, durationMs={sw.ElapsedMilliseconds}, status={(int)response.StatusCode}): {result.ErrorReason}");
@@ -208,6 +212,12 @@ namespace AutopilotMonitor.Agent.V2.Core.Transport.Telemetry
case HttpStatusCode.RequestEntityTooLarge: // 413 — the data is fine, only the batch
return UploadResult.TooLarge(shortReason); // size is wrong → split & retry (P1).
+
+ case HttpStatusCode.Gone: // 410 — the backend no longer knows this session.
+ // Decided BEFORE the poison-body check: a poison body on 410 would make the
+ // drain drop items and emit telemetry_upload_poisoned, which the backend
+ // rejects with 410 again — an emit/reject loop for a session that is over.
+ return UploadResult.SessionGone($"session gone: {shortReason}");
}
if (statusCode >= 500 && statusCode <= 599) return UploadResult.Transient(shortReason);
diff --git a/src/Agent/AutopilotMonitor.Agent.V2.Core/Transport/Telemetry/TelemetryPartitionKey.cs b/src/Agent/AutopilotMonitor.Agent.V2.Core/Transport/Telemetry/TelemetryPartitionKey.cs
new file mode 100644
index 00000000..7b1b4b2d
--- /dev/null
+++ b/src/Agent/AutopilotMonitor.Agent.V2.Core/Transport/Telemetry/TelemetryPartitionKey.cs
@@ -0,0 +1,14 @@
+#nullable enable
+namespace AutopilotMonitor.Agent.V2.Core.Transport.Telemetry
+{
+ ///
+ /// The one composition of a session's Azure-Table partition key. The three emitters stamp
+ /// it on every ; the spool and the upload orchestrator compare
+ /// against it to tell the current session's items from leftovers of a previous session id.
+ /// A drift between producer and comparer would silently drop or retain the wrong items.
+ ///
+ public static class TelemetryPartitionKey
+ {
+ public static string ForSession(string tenantId, string sessionId) => $"{tenantId}_{sessionId}";
+ }
+}
diff --git a/src/Agent/AutopilotMonitor.Agent.V2.Core/Transport/Telemetry/TelemetrySpool.cs b/src/Agent/AutopilotMonitor.Agent.V2.Core/Transport/Telemetry/TelemetrySpool.cs
index b7b39924..3dc6fc7a 100644
--- a/src/Agent/AutopilotMonitor.Agent.V2.Core/Transport/Telemetry/TelemetrySpool.cs
+++ b/src/Agent/AutopilotMonitor.Agent.V2.Core/Transport/Telemetry/TelemetrySpool.cs
@@ -74,7 +74,16 @@ namespace AutopilotMonitor.Agent.V2.Core.Transport.Telemetry
public int ByteLength { get; }
}
- public TelemetrySpool(string directoryPath, IClock clock, Logging.AgentLogger? logger = null)
+ ///
+ /// Partition key of the session this process serves ().
+ /// When set, pending items of another session id found at the head of the spool are dropped
+ /// on startup — see . Null disables the filter.
+ ///
+ public TelemetrySpool(
+ string directoryPath,
+ IClock clock,
+ Logging.AgentLogger? logger = null,
+ string? expectedPartitionKey = null)
{
if (string.IsNullOrEmpty(directoryPath))
{
@@ -96,6 +105,7 @@ namespace AutopilotMonitor.Agent.V2.Core.Transport.Telemetry
// count as "pending" and need to be rehydrated into _pending.
_lastUploadedItemId = _cursor.Load();
_lastAssignedItemId = ScanSpoolAndRehydratePending();
+ DropLeadingForeignPending(expectedPartitionKey);
_peakPendingCount = _pending.Count;
}
@@ -312,5 +322,68 @@ namespace AutopilotMonitor.Agent.V2.Core.Transport.Telemetry
}
return highest;
}
+
+ ///
+ /// Startup filter: drops the LEADING run of pending items whose partition key is not
+ /// , advancing and persisting the cursor past them.
+ /// Such items are leftovers of a previous session id in the same spool directory; the
+ /// backend answers 410 for them, and that 410 must never end the session this process
+ /// serves. One process serves one session, so a foreign item AFTER an own item cannot
+ /// occur — if it does, the order is not trustworthy and nothing is dropped. WhiteGlove
+ /// Part 2 reuses the Part-1 session id, so its leftovers carry the expected key and stay.
+ /// Called once from the ctor, after rehydration.
+ ///
+ private void DropLeadingForeignPending(string? expectedPartitionKey)
+ {
+ if (string.IsNullOrEmpty(expectedPartitionKey) || _pending.Count == 0) return;
+
+ var leadingForeign = 0;
+ long lastForeignId = -1;
+ var seenOwn = false;
+ foreach (var entry in _pending)
+ {
+ var isForeign = !string.Equals(entry.Item.PartitionKey, expectedPartitionKey, StringComparison.Ordinal);
+ if (!isForeign)
+ {
+ seenOwn = true;
+ continue;
+ }
+
+ if (!seenOwn)
+ {
+ leadingForeign++;
+ lastForeignId = entry.Item.TelemetryItemId;
+ continue;
+ }
+
+ _logger?.Error(
+ $"TelemetrySpool: pending itemId={entry.Item.TelemetryItemId} carries a foreign partition key after items of the current session — spool order is inconsistent; keeping all {_pending.Count} pending item(s).");
+ return;
+ }
+
+ if (leadingForeign == 0) return;
+
+ try
+ {
+ _cursor.Save(lastForeignId);
+ }
+ catch (Exception ex)
+ {
+ // Keep the items rather than risk a cursor that disagrees with disk; the drain
+ // loop drops the same leading run on its first 410 (defence in depth).
+ _logger?.Warning($"TelemetrySpool: could not persist the cursor past {leadingForeign} foreign pending item(s): {ex.Message}");
+ return;
+ }
+
+ _lastUploadedItemId = lastForeignId;
+ for (var i = 0; i < leadingForeign; i++)
+ {
+ _pendingBytes -= _pending.First!.Value.ByteLength;
+ _pending.RemoveFirst();
+ }
+
+ _logger?.Warning(
+ $"TelemetrySpool: dropped {leadingForeign} pending item(s) of a previous session id (cursor advanced to {lastForeignId}, {_pending.Count} pending item(s) of the current session kept).");
+ }
}
}
diff --git a/src/Agent/AutopilotMonitor.Agent.V2.Core/Transport/Telemetry/TelemetryUploadOrchestrator.cs b/src/Agent/AutopilotMonitor.Agent.V2.Core/Transport/Telemetry/TelemetryUploadOrchestrator.cs
index 9875ab0e..2a499dcd 100644
--- a/src/Agent/AutopilotMonitor.Agent.V2.Core/Transport/Telemetry/TelemetryUploadOrchestrator.cs
+++ b/src/Agent/AutopilotMonitor.Agent.V2.Core/Transport/Telemetry/TelemetryUploadOrchestrator.cs
@@ -45,12 +45,28 @@ namespace AutopilotMonitor.Agent.V2.Core.Transport.Telemetry
// telemetry_upload_blocked exactly once per process rather than on every 30 s drain.
private bool _uploadBlockedReported;
+ // Partition key of the session this process serves; null = every batch counts as own.
+ private readonly string? _currentPartitionKey;
+ private readonly Logging.AgentLogger? _logger;
+
+ // One-shot guard for SessionGone. Latched only when a handler is subscribed: the host
+ // wires the handler after EnrollmentOrchestrator.Start, and a 410 on the very first
+ // drain must not be swallowed before anyone listens.
+ private bool _sessionGoneReported;
+
+ ///
+ /// of the registered session. A 410 on a
+ /// batch that starts with this key ends the session (); a 410 on
+ /// a batch that starts with any other key only drops those leftover items.
+ ///
public TelemetryUploadOrchestrator(
ITelemetrySpool spool,
IBackendTelemetryUploader uploader,
IClock clock,
int batchSize = 100,
- IReadOnlyList? retryBackoffs = null)
+ IReadOnlyList? retryBackoffs = null,
+ string? currentPartitionKey = null,
+ Logging.AgentLogger? logger = null)
{
_spool = spool ?? throw new ArgumentNullException(nameof(spool));
_uploader = uploader ?? throw new ArgumentNullException(nameof(uploader));
@@ -61,6 +77,8 @@ namespace AutopilotMonitor.Agent.V2.Core.Transport.Telemetry
_effectiveBatchSize = batchSize;
_retryBackoffs = retryBackoffs ?? DefaultRetryBackoffs;
+ _currentPartitionKey = currentPartitionKey;
+ _logger = logger;
}
///
@@ -97,6 +115,15 @@ namespace AutopilotMonitor.Agent.V2.Core.Transport.Telemetry
///
public event EventHandler? UploadBlocked;
+ ///
+ /// Raised once per process (after the drain guard releases) when the backend answered
+ /// HTTP 410 for a batch of the CURRENT session: the session is over server-side (never
+ /// registered, deleted, or being deleted) and no retry can succeed. The batch is retained
+ /// (cursor stays) and the drain stops; the host reports once over the emergency channel
+ /// and terminates the agent with self-destruct. Carries the backend error reason.
+ ///
+ public event EventHandler? SessionGone;
+
///
/// true when the last successful upload returned
/// and the paused-until window has not expired. Drain is a no-op in this state.
@@ -124,6 +151,7 @@ namespace AutopilotMonitor.Agent.V2.Core.Transport.Telemetry
List? pendingSignals = null;
List? pendingPoison = null;
string? pendingBlockedReason = null;
+ string? pendingSessionGoneReason = null;
DrainResult result;
try
@@ -247,6 +275,35 @@ namespace AutopilotMonitor.Agent.V2.Core.Transport.Telemetry
new PoisonReport(1, oversizeId, "oversize: " + outcome.ErrorReason, PoisonKind.Oversize));
break;
}
+ else if (outcome.IsSessionGone)
+ {
+ // HTTP 410 — the backend no longer knows the session these items belong to.
+ var foreignRun = CountLeadingForeignItems(batch);
+ if (foreignRun > 0)
+ {
+ // Leftovers of a previous session id (the spool drops these at startup;
+ // this is defence in depth). Drop only that leading run and carry on: a
+ // legitimate new session must never be ended because of old items.
+ var throughId = batch[foreignRun - 1].TelemetryItemId;
+ _spool.MarkUploaded(throughId);
+ _logger?.Warning(
+ $"TelemetryUploadOrchestrator: dropped {foreignRun} item(s) of a previous session id the backend rejected with 410 (cursor advanced to {throughId}).");
+ continue;
+ }
+
+ // The current session is over server-side. RETAIN (cursor stays, the items
+ // remain on disk for the diagnostics ZIP), stop this drain, and surface it
+ // once — not as telemetry_upload_blocked, which would only queue another
+ // item behind a batch that can never be accepted.
+ failedBatches++;
+ lastError = outcome.ErrorReason;
+ if (!_sessionGoneReported && SessionGone != null)
+ {
+ _sessionGoneReported = true;
+ pendingSessionGoneReason = outcome.ErrorReason;
+ }
+ break;
+ }
else
{
// P1 — RETAIN. Transient (retry next drain), auth-permanent (uploader already
@@ -301,9 +358,32 @@ namespace AutopilotMonitor.Agent.V2.Core.Transport.Telemetry
catch { /* handler must not abort the drain result */ }
}
+ if (pendingSessionGoneReason != null)
+ {
+ try { SessionGone?.Invoke(this, pendingSessionGoneReason); }
+ catch { /* handler must not abort the drain result */ }
+ }
+
return result;
}
+ ///
+ /// Number of items at the head of whose partition key is not the
+ /// current session's. 0 when no current key is configured (every batch counts as own).
+ ///
+ private int CountLeadingForeignItems(IReadOnlyList batch)
+ {
+ if (_currentPartitionKey == null) return 0;
+
+ var count = 0;
+ while (count < batch.Count
+ && !string.Equals(batch[count].PartitionKey, _currentPartitionKey, StringComparison.Ordinal))
+ {
+ count++;
+ }
+ return count;
+ }
+
private static bool CarriesSignal(UploadResult outcome) =>
outcome.DeviceBlocked
|| outcome.DeviceKillSignal
diff --git a/src/Agent/AutopilotMonitor.Agent.V2.Core/Transport/Telemetry/UploadResult.cs b/src/Agent/AutopilotMonitor.Agent.V2.Core/Transport/Telemetry/UploadResult.cs
index ae4f14c8..9f138f89 100644
--- a/src/Agent/AutopilotMonitor.Agent.V2.Core/Transport/Telemetry/UploadResult.cs
+++ b/src/Agent/AutopilotMonitor.Agent.V2.Core/Transport/Telemetry/UploadResult.cs
@@ -30,7 +30,8 @@ namespace AutopilotMonitor.Agent.V2.Core.Transport.Telemetry
bool isAuthFailure = false,
bool requiresSplit = false,
bool isPoison = false,
- IReadOnlyList? poisonRowKeys = null)
+ IReadOnlyList? poisonRowKeys = null,
+ bool isSessionGone = false)
{
Success = success;
ErrorReason = errorReason;
@@ -44,6 +45,7 @@ namespace AutopilotMonitor.Agent.V2.Core.Transport.Telemetry
RequiresSplit = requiresSplit;
IsPoison = isPoison;
PoisonRowKeys = poisonRowKeys;
+ IsSessionGone = isSessionGone;
}
public bool Success { get; }
@@ -81,6 +83,16 @@ namespace AutopilotMonitor.Agent.V2.Core.Transport.Telemetry
/// The RowKeys the backend named as poison (item-level). Non-empty when .
public IReadOnlyList? PoisonRowKeys { get; }
+ ///
+ /// HTTP 410: the backend no longer knows the session the batch belongs to (never
+ /// registered, deleted, or being deleted). Not transient, not auth, not poison — retrying
+ /// can never succeed. The orchestrator stops the drain (cursor stays) and raises
+ /// when the batch belongs to the
+ /// current session, or drops the batch's leading items when they are leftovers of a
+ /// previous session id.
+ ///
+ public bool IsSessionGone { get; }
+
///
/// Non-terminal quarantine signalled by the backend. The agent should stop draining
/// uploads until (if set) elapses — the session stays alive so
@@ -165,5 +177,14 @@ namespace AutopilotMonitor.Agent.V2.Core.Transport.Telemetry
new UploadResult(false, reason ?? throw new ArgumentNullException(nameof(reason)), isTransient: false,
deviceBlocked: false, unblockAt: null, deviceKillSignal: false, adminAction: null, actions: null,
isPoison: true, poisonRowKeys: rejectedRowKeys ?? throw new ArgumentNullException(nameof(rejectedRowKeys)));
+
+ ///
+ /// HTTP 410 — the backend no longer knows the session. Flagged ;
+ /// never retried, never split, never treated as poison.
+ ///
+ public static UploadResult SessionGone(string reason) =>
+ new UploadResult(false, reason ?? throw new ArgumentNullException(nameof(reason)), isTransient: false,
+ deviceBlocked: false, unblockAt: null, deviceKillSignal: false, adminAction: null, actions: null,
+ isSessionGone: true);
}
}
diff --git a/src/Agent/AutopilotMonitor.Agent.V2/Runtime/AgentRuntimeHost.cs b/src/Agent/AutopilotMonitor.Agent.V2/Runtime/AgentRuntimeHost.cs
index 472a3e9f..0466e1bc 100644
--- a/src/Agent/AutopilotMonitor.Agent.V2/Runtime/AgentRuntimeHost.cs
+++ b/src/Agent/AutopilotMonitor.Agent.V2/Runtime/AgentRuntimeHost.cs
@@ -439,6 +439,8 @@ namespace AutopilotMonitor.Agent.V2.Runtime
lifecyclePost,
() => terminationHandler,
agentConfig,
+ auth.EmergencyReporter,
+ shutdown,
shutdownComplete,
logger);
diff --git a/src/Agent/AutopilotMonitor.Agent.V2/Runtime/ServerControlPlane.cs b/src/Agent/AutopilotMonitor.Agent.V2/Runtime/ServerControlPlane.cs
index 885e4197..ccc04e81 100644
--- a/src/Agent/AutopilotMonitor.Agent.V2/Runtime/ServerControlPlane.cs
+++ b/src/Agent/AutopilotMonitor.Agent.V2/Runtime/ServerControlPlane.cs
@@ -1,10 +1,12 @@
using System;
using System.Collections.Generic;
+using System.Net;
using System.Threading;
using System.Threading.Tasks;
using AutopilotMonitor.Agent.V2.Core.Configuration;
using AutopilotMonitor.Agent.V2.Core.Logging;
using AutopilotMonitor.Agent.V2.Core.Monitoring.Runtime;
+using AutopilotMonitor.Agent.V2.Core.Monitoring.Transport;
using AutopilotMonitor.Agent.V2.Core.Orchestration;
using AutopilotMonitor.Agent.V2.Core.Termination;
using AutopilotMonitor.Shared;
@@ -103,6 +105,9 @@ namespace AutopilotMonitor.Agent.V2.Runtime
/// dispatcher because these are real actions, not admin-button echoes.
/// - DeviceBlocked — non-terminal upload pause. Transport already reacted in
/// TelemetryUploadOrchestrator.ApplyControlSignals; here we only log.
+ /// - SessionGone (, HTTP 410
+ /// on the current session's telemetry) — one emergency report, then the kill-signal
+ /// termination path tagged origin=session_gone. See .
///
///
/// Shutdown race guard: if the agent has already finished its own termination
@@ -117,9 +122,14 @@ namespace AutopilotMonitor.Agent.V2.Runtime
InformationalEventPost post,
Func terminationHandlerAccessor,
AgentConfiguration agentConfig,
+ EmergencyReporter emergencyReporter,
+ ManualResetEventSlim shutdown,
ManualResetEventSlim shutdownComplete,
AgentLogger logger)
{
+ orchestrator.Transport.SessionGone += (sender, reason) =>
+ HandleSessionGone(reason, emergencyReporter, dispatcher, shutdown, shutdownComplete, logger);
+
orchestrator.Transport.ServerResponseReceived += (sender, upload) =>
{
if (shutdownComplete.IsSet)
@@ -140,7 +150,7 @@ namespace AutopilotMonitor.Agent.V2.Runtime
{
{ "forceSelfDestruct", "true" },
{ "gracePeriodSeconds", "0" },
- { "origin", "kill_signal" },
+ { "origin", TerminationOrigins.KillSignal },
},
};
try { dispatcher.DispatchAsync(new List { killAction }).GetAwaiter().GetResult(); }
@@ -181,6 +191,76 @@ namespace AutopilotMonitor.Agent.V2.Runtime
};
}
+ // Bounded blocking send for the one session-gone report: the terminate that follows ends
+ // the process and the cleanup script deletes the agent, so a fire-and-forget send would
+ // be abandoned mid-flight. Retries stay inside one anti-flood reservation (ONE report).
+ private const int SessionGoneReportAttempts = 2;
+ private static readonly TimeSpan SessionGoneReportTimeout = TimeSpan.FromSeconds(15);
+ private static readonly TimeSpan SessionGoneReportRetryDelay = TimeSpan.FromSeconds(3);
+ private static readonly TimeSpan SessionGoneReportBudget = TimeSpan.FromSeconds(40);
+
+ ///
+ /// The backend answered HTTP 410 for the current session's telemetry: the session is over
+ /// server-side. (a) Send ONE
+ /// report over the emergency channel — the telemetry channel is exactly what was refused,
+ /// and the backend accepts this report for unknown sessions. (b) Terminate through the same
+ /// synthesised terminate_session path as the administrator kill signal (forced
+ /// self-destruct, no grace), tagged so the
+ /// termination handler skips the spool drain and the diagnostics upload. A shutdown that is
+ /// already in progress owns the outcome — then nothing happens here.
+ ///
+ internal static void HandleSessionGone(
+ string reason,
+ EmergencyReporter emergencyReporter,
+ ServerActionDispatcher dispatcher,
+ ManualResetEventSlim shutdown,
+ ManualResetEventSlim shutdownComplete,
+ AgentLogger logger)
+ {
+ if (shutdownComplete.IsSet || shutdown.IsSet)
+ {
+ logger.Debug("ServerControlPlane: session gone (HTTP 410) while shutdown is already in progress — ignoring.");
+ return;
+ }
+
+ logger.Error($"Backend no longer knows this session (HTTP 410: {reason}) — reporting once over the emergency channel, then terminating with self-destruct.");
+
+ if (emergencyReporter != null)
+ {
+ try
+ {
+ emergencyReporter
+ .TrySendAsync(
+ AgentErrorType.TelemetryRejectedSessionUnknown,
+ $"Telemetry upload rejected with HTTP 410 — backend no longer knows this session ({reason}). Agent terminating with self-destruct.",
+ httpStatusCode: (int)HttpStatusCode.Gone,
+ attempts: SessionGoneReportAttempts,
+ perAttemptTimeout: SessionGoneReportTimeout,
+ retryDelay: SessionGoneReportRetryDelay)
+ .Wait(SessionGoneReportBudget);
+ }
+ catch (Exception ex)
+ {
+ logger.Debug($"Session-gone report send failed (best-effort): {ex.Message}");
+ }
+ }
+
+ var terminateAction = new ServerAction
+ {
+ Type = ServerActionTypes.TerminateSession,
+ Reason = "Backend no longer knows this session (HTTP 410)",
+ QueuedAt = DateTime.UtcNow,
+ Params = new Dictionary
+ {
+ { "forceSelfDestruct", "true" },
+ { "gracePeriodSeconds", "0" },
+ { "origin", TerminationOrigins.SessionGone },
+ },
+ };
+ try { dispatcher.DispatchAsync(new List { terminateAction }).GetAwaiter().GetResult(); }
+ catch (Exception ex) { logger.Error("ServerActionDispatcher.DispatchAsync (session gone) threw during server-response wiring.", ex); }
+ }
+
///
/// Soft admin-override handling. Portal Mark-Succeeded / Mark-Failed is an echo of an
/// admin button click, not a ServerAction: it bypasses the dispatcher (no
@@ -281,6 +361,17 @@ namespace AutopilotMonitor.Agent.V2.Runtime
: EnrollmentTerminationOutcome.Failed;
}
+ ///
+ /// The origin param of a synthesised terminate_session action
+ /// (), or null when absent — it travels into
+ /// .
+ ///
+ internal static string ReadOrigin(IReadOnlyDictionary parameters)
+ {
+ if (parameters == null) return null;
+ return parameters.TryGetValue("origin", out var origin) && !string.IsNullOrEmpty(origin) ? origin : null;
+ }
+
// ============================================================ Private terminate-callback
///
@@ -326,8 +417,9 @@ namespace AutopilotMonitor.Agent.V2.Runtime
// before, masquerading every admin override as an error in SummaryDialog +
// firing spurious diagnostics uploads).
var mappedOutcome = MapAdminOutcome(action?.Params);
+ var origin = ReadOrigin(action?.Params);
- logger.Warning($"ServerAction terminate_session received (ruleId={action?.RuleId}, reason={action?.Reason}, outcome={mappedOutcome}) — invoking termination handler.");
+ logger.Warning($"ServerAction terminate_session received (ruleId={action?.RuleId}, reason={action?.Reason}, outcome={mappedOutcome}, origin={origin ?? "none"}) — invoking termination handler.");
// Synthesise a Terminated event as if the kernel fired it.
terminationHandler.Handle(
sender: null,
@@ -336,7 +428,8 @@ namespace AutopilotMonitor.Agent.V2.Runtime
outcome: mappedOutcome,
stageName: orchestrator.CurrentState?.Stage.ToString(),
terminatedAtUtc: DateTime.UtcNow,
- details: $"Server-requested termination: ruleId={action?.RuleId}, reason={action?.Reason}"));
+ details: $"Server-requested termination: ruleId={action?.RuleId}, reason={action?.Reason}",
+ origin: origin));
// Plan §6.2 synchronous-shutdown — block the ingest dispatcher thread until the
// main-thread cleanup (orchestrator.Stop, client disposal) has fully run. This
diff --git a/src/Backend/AutopilotMonitor.Functions.Tests/DerivedWriterLivenessTests.cs b/src/Backend/AutopilotMonitor.Functions.Tests/DerivedWriterLivenessTests.cs
new file mode 100644
index 00000000..1360ef53
--- /dev/null
+++ b/src/Backend/AutopilotMonitor.Functions.Tests/DerivedWriterLivenessTests.cs
@@ -0,0 +1,126 @@
+using System;
+using System.Collections.Generic;
+using System.Threading;
+using System.Threading.Tasks;
+using AutopilotMonitor.Functions.Services;
+using AutopilotMonitor.Functions.Services.Analyze;
+using AutopilotMonitor.Functions.Services.Deletion;
+using AutopilotMonitor.Functions.Services.Vulnerability;
+using AutopilotMonitor.Shared.DataAccess;
+using AutopilotMonitor.Shared.Models;
+using AutopilotMonitor.Shared.Models.Deletion;
+using Azure.Data.Tables;
+using Microsoft.Extensions.Caching.Memory;
+using Microsoft.Extensions.Logging.Abstractions;
+using Moq;
+using Xunit;
+
+namespace AutopilotMonitor.Functions.Tests;
+
+///
+/// The two queue handlers derive rows for a session (rule results, reports, inventory counters,
+/// CveIndex). For a session the backend does not know, or one a cascade is tearing down, they
+/// must stop before the first read of session data — nothing derived may ever outlive the
+/// session's cascade. Both handlers are driven up to that gate with every later dependency
+/// unavailable, so a regression that moves the gate shows up as a null dereference or an
+/// unexpected mock call.
+///
+public class DerivedWriterLivenessTests
+{
+ private const string TenantId = "11111111-1111-1111-1111-111111111111";
+ private const string SessionId = "22222222-2222-2222-2222-222222222222";
+
+ private static SessionDeletionGuard GuardFor(TableEntity? row)
+ {
+ var reader = new Mock();
+ reader.Setup(r => r.GetSessionRowAsync(TenantId, SessionId, It.IsAny>(), It.IsAny()))
+ .ReturnsAsync(row);
+ return new SessionDeletionGuard(reader.Object, NullLogger.Instance);
+ }
+
+ private static Mock AdminConfigWithCorrelationEnabled()
+ {
+ var adminConfig = new Mock(
+ Mock.Of(), NullLogger.Instance,
+ new MemoryCache(new MemoryCacheOptions()));
+ adminConfig.Setup(a => a.GetConfigurationAsync())
+ .ReturnsAsync(new AdminConfiguration { VulnerabilityCorrelationEnabled = true });
+ return adminConfig;
+ }
+
+ public static IEnumerable