feat(agent-v2): V2 agent review remediation (waves 1-3)

Read-only V2 agent review remediation; V2-only, net48-safe, durability/recovery
guarantees preserved (SignalLog + Journal still fsync per step).

Wave 1 (small/independent):
- PERF-H3+MON-A2: shared SignalActivityClassifier (excludes DeadlineFired/classifier_tick
  + periodic & agent-health events) - fixes idle-stop defeated by the 30s tick and StallProbeHost
  measuring uptime instead of idleness (+ ResetProbes on real activity, generation-guarded).
- LIFE-F2: cleanup reboot uses shutdown.exe /r (Restart-Computer -Comment/-Delay never rebooted).
- MON-D1: one-shot collector_degraded on event-log watcher arming failure (ShellCore/Hello/ModernDeployment).
- TRACE-M1: AdminPreemption terminal events carry the decision audit trail.
- ARCH-F8: DesktopArrivalDetector reentrancy guard.
- MON-B8: stall active-install age filter (CMTrace timestamp, conservative).

Wave 2 (durability efficiency, guarantee-preserving):
- H1a: snapshot only on stage-change/effect-bearing steps (pass-throughs replayed; floor-bounded).
- H2: classifier_tick armed reactively on first WhiteGlove signal, not eagerly at SessionStarted.
- M1: IME state-save throttled (>=2s) + atomic Formatting.None File.Replace.
- M2: single IME directory enumeration per tick + guarded Trace lines.
- M5: cached DecisionCoreJsonSettings.Shared.

Wave 3 (robustness, minus LIFE-F4/F5):
- MON-A1: gather phase_change/on_event triggers driven by SignalIngress.SignalPosted.
- TRACE-H2: quarantine contract - TriggerQuarantine persists a marker honoured fail-closed on Start.
- TRACE-H1 (defensive per review P1): explicit item-level backend poison signal (body flag) is the
  only authorized discard; 413 -> split/shrink; unknown permanent 4xx -> retain + telemetry_upload_blocked
  (local diagnostic); auth-permanent retained. Survivor re-upload honours backend control signals;
  poison forensics distinguish Oversize vs BackendRejected.
- LIFE-F3: drain spool before launching the self-destruct cleanup script.
- LIFE-F6: cleanup verifies task removal (+schtasks fallback) before deleting files.

New event types: collector_degraded, state_quarantine_recovered, telemetry_upload_poisoned,
telemetry_upload_blocked. De-flaked OfficeBinaryWatcher bound test (synchronise-on-bound).
Tests: V2.Core 1314/1314, DecisionCore 358/358, MCP tsc clean.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
Oliver Kieselbach
2026-06-10 18:09:11 +02:00
parent d68f6d23e0
commit b45384fd01
43 changed files with 1805 additions and 222 deletions
@@ -64,14 +64,16 @@ namespace AutopilotMonitor.Agent.V2.Core.Tests.Monitoring.Office
watcher.Start();
watcher.ScheduleRecheck();
// Give it well beyond max*interval, then assert it has stopped and never exceeded the bound.
Thread.Sleep(400);
int settled = Volatile.Read(ref raises);
Thread.Sleep(200);
// Deterministic instead of fixed-sleep (the old version raced a delayed re-probe under
// load): the probe raises exactly maxRechecks times then stops, so first SYNCHRONISE on
// reaching the bound (generous timeout tolerates scheduler delay), then prove it stopped.
bool reachedBound = SpinUntil(() => Volatile.Read(ref raises) >= 3, 10000);
Assert.True(reachedBound, $"re-probe did not reach its bound in time; got {Volatile.Read(ref raises)}");
Assert.Equal(settled, Volatile.Read(ref raises)); // no further raises — bounded
Assert.True(settled <= 3, $"re-probe exceeded its bound: {settled}");
Assert.True(settled >= 1, "re-probe never fired");
int settled = Volatile.Read(ref raises);
Thread.Sleep(200); // room for a (buggy) 4th raise past the bound
Assert.Equal(settled, Volatile.Read(ref raises)); // stopped — no further raises
Assert.Equal(3, settled); // fired exactly maxRechecks, no more
}
[Fact]
@@ -145,14 +147,15 @@ namespace AutopilotMonitor.Agent.V2.Core.Tests.Monitoring.Office
Assert.Equal(afterDispose, Volatile.Read(ref raises)); // no raises after dispose
}
private static void SpinUntil(Func<bool> condition, int timeoutMs)
private static bool SpinUntil(Func<bool> condition, int timeoutMs)
{
var deadline = Environment.TickCount + timeoutMs;
while (Environment.TickCount < deadline)
{
if (condition()) return;
if (condition()) return true;
Thread.Sleep(10);
}
return condition();
}
}
}
@@ -0,0 +1,74 @@
using System;
using System.Collections.Generic;
using AutopilotMonitor.Agent.V2.Core.Monitoring.Enrollment.SystemSignals;
using AutopilotMonitor.Agent.V2.Core.Orchestration;
using AutopilotMonitor.Agent.V2.Core.Tests.Harness;
using AutopilotMonitor.Agent.V2.Core.Tests.Orchestration;
using AutopilotMonitor.DecisionCore.Engine;
using AutopilotMonitor.DecisionCore.Signals;
using Xunit;
namespace AutopilotMonitor.Agent.V2.Core.Tests.Monitoring.SystemSignals
{
/// <summary>
/// Review MON-D1: a watcher that fails to arm must surface a one-shot <c>collector_degraded</c>
/// Warning so the backend can tell a dead kernel watcher apart from a genuine no-signal session.
/// </summary>
public sealed class CollectorDegradationReporterTests
{
private static readonly DateTime At = new DateTime(2026, 6, 10, 9, 0, 0, DateTimeKind.Utc);
private static (InformationalEventPost post, FakeSignalIngressSink sink) BuildRig()
{
var sink = new FakeSignalIngressSink();
var post = new InformationalEventPost(sink, new VirtualClock(At));
return (post, sink);
}
[Fact]
public void Report_emits_warning_collector_degraded_with_collector_and_error_detail()
{
var (post, sink) = BuildRig();
CollectorDegradationReporter.Report(
post, sessionId: "s1", tenantId: "t1",
collectorName: "ShellCoreTracker",
reason: "watcher_arm_failed",
ex: new InvalidOperationException("boom"));
var posted = Assert.Single(sink.Posted);
Assert.Equal(DecisionSignalKind.InformationalEvent, posted.Kind);
Assert.Equal("collector_degraded", posted.Payload![SignalPayloadKeys.EventType]);
Assert.Equal("ShellCoreTracker", posted.Payload[SignalPayloadKeys.Source]);
Assert.Equal("Warning", posted.Payload[SignalPayloadKeys.Severity]);
var data = Assert.IsAssignableFrom<IReadOnlyDictionary<string, object>>(posted.TypedPayload!);
Assert.Equal("ShellCoreTracker", data["collector"]);
Assert.Equal("watcher_arm_failed", data["reason"]);
Assert.Equal("boom", data["error"]);
Assert.Equal(nameof(InvalidOperationException), data["errorType"]);
}
[Fact]
public void Report_without_exception_omits_error_fields()
{
var (post, sink) = BuildRig();
CollectorDegradationReporter.Report(
post, "s1", "t1", "ModernDeploymentTracker", "watcher_arm_failed:SomeChannel");
var posted = Assert.Single(sink.Posted);
var data = Assert.IsAssignableFrom<IReadOnlyDictionary<string, object>>(posted.TypedPayload!);
Assert.Equal("ModernDeploymentTracker", data["collector"]);
Assert.Equal("watcher_arm_failed:SomeChannel", data["reason"]);
Assert.False(data.ContainsKey("error"));
Assert.False(data.ContainsKey("errorType"));
}
[Fact]
public void Report_with_null_post_is_a_noop_and_does_not_throw()
{
CollectorDegradationReporter.Report(null, "s1", "t1", "X", "y");
}
}
}
@@ -0,0 +1,91 @@
using System;
using System.Linq;
using AutopilotMonitor.Agent.V2.Core.Monitoring.Enrollment.SystemSignals;
using Xunit;
namespace AutopilotMonitor.Agent.V2.Core.Tests.Monitoring.SystemSignals
{
/// <summary>
/// Review MON-B8: the AppWorkload active-install scan must age-filter so a wedged install whose
/// last "EnforcementState: Installing" line lingers in the 200 KB tail stops counting as
/// progress (it was suppressing <c>session_stalled</c> forever). Conservative: only lines with a
/// CMTrace timestamp provably older than the freshness window are dropped; unparseable /
/// unknown-age lines stay counted so a long-but-active installer is never mis-flagged.
/// </summary>
public sealed class StallProbeActiveInstallFilterTests
{
private const int FreshnessMinutes = 15;
// CMTrace timestamps are LOCAL in the log and the parser converts to UTC. Build lines from a
// local DateTime and compare against DateTime.UtcNow so the round-trip is timezone-correct.
private static string CmTraceLine(string message, DateTime localTime) =>
$"<![LOG[{message}]LOG]!><time=\"{localTime:HH:mm:ss.fffffff}\" date=\"{localTime:M-d-yyyy}\" " +
"component=\"AppEnforce\" context=\"\" type=\"1\" thread=\"1\" file=\"\">";
[Fact]
public void Fresh_cmtrace_install_line_is_counted()
{
var content = CmTraceLine("EnforcementState: Installing app X", DateTime.Now);
var fresh = StallProbeCollector.FilterFreshActiveInstalls(content, DateTime.UtcNow, FreshnessMinutes);
Assert.Single(fresh);
Assert.Contains("EnforcementState: Installing", fresh[0]);
}
[Fact]
public void Stale_cmtrace_install_line_is_excluded()
{
// Last install marker is an hour old → no longer evidence of progress.
var content = CmTraceLine("EnforcementState: Installing app X", DateTime.Now.AddMinutes(-60));
var fresh = StallProbeCollector.FilterFreshActiveInstalls(content, DateTime.UtcNow, FreshnessMinutes);
Assert.Empty(fresh);
}
[Fact]
public void Non_cmtrace_line_without_timestamp_stays_fresh()
{
// A raw line that does not parse as CMTrace has unknown age → kept (safe default).
var content = "some prefix EnforcementState: Downloading something";
var fresh = StallProbeCollector.FilterFreshActiveInstalls(content, DateTime.UtcNow, FreshnessMinutes);
Assert.Single(fresh);
}
[Fact]
public void Mixed_tail_keeps_only_fresh_lines()
{
var content = string.Join("\n", new[]
{
CmTraceLine("EnforcementState: Installing fresh-app", DateTime.Now),
CmTraceLine("EnforcementState: Installing stale-app", DateTime.Now.AddMinutes(-90)),
CmTraceLine("Nothing interesting here", DateTime.Now),
});
var fresh = StallProbeCollector.FilterFreshActiveInstalls(content, DateTime.UtcNow, FreshnessMinutes);
// Only the fresh install marker survives; the stale one and the non-matching line drop.
Assert.Single(fresh);
}
[Fact]
public void Empty_or_null_content_returns_empty()
{
Assert.Empty(StallProbeCollector.FilterFreshActiveInstalls(null, DateTime.UtcNow, FreshnessMinutes));
Assert.Empty(StallProbeCollector.FilterFreshActiveInstalls(string.Empty, DateTime.UtcNow, FreshnessMinutes));
}
[Fact]
public void Line_without_active_install_marker_is_not_counted()
{
var content = CmTraceLine("DetectionState: Detected app Y", DateTime.Now);
var fresh = StallProbeCollector.FilterFreshActiveInstalls(content, DateTime.UtcNow, FreshnessMinutes);
Assert.Empty(fresh);
}
}
}
@@ -1,4 +1,5 @@
using System;
using System.Collections.Generic;
using AutopilotMonitor.Agent.V2.Core.Logging;
using AutopilotMonitor.Agent.V2.Core.Orchestration;
using AutopilotMonitor.Agent.V2.Core.Tests.Harness;
@@ -52,6 +53,31 @@ namespace AutopilotMonitor.Agent.V2.Core.Tests.Orchestration
return (step, signal);
}
/// <summary>Reduces SessionAborted — a stage change (SessionStarted → Failed) that H1a always snapshots.</summary>
public (DecisionStep step, DecisionSignal signal) ReduceAbort(long ordinal = 0)
{
var signal = TestSignals.Raw(ordinal, DecisionSignalKind.SessionAborted, At);
var step = Engine.Reduce(InitialState, signal);
return (step, signal);
}
/// <summary>
/// Reduces an InformationalEvent pass-through (stage unchanged, only a telemetry
/// EmitEventTimelineEntry effect) from the processor's CURRENT state — the high-frequency
/// step class H1a skips for snapshots.
/// </summary>
public (DecisionStep step, DecisionSignal signal) ReduceInformational(DecisionStepProcessor sut, long ordinal)
{
var payload = new Dictionary<string, string>
{
[SignalPayloadKeys.EventType] = "download_progress",
[SignalPayloadKeys.Source] = "DeliveryOptimizationCollector",
};
var signal = TestSignals.Raw(ordinal, DecisionSignalKind.InformationalEvent, At, payload: payload);
var step = Engine.Reduce(sut.CurrentState, signal);
return (step, signal);
}
public void Dispose() => Tmp.Dispose();
}
@@ -94,7 +120,7 @@ namespace AutopilotMonitor.Agent.V2.Core.Tests.Orchestration
// ========================================================================= Happy path
[Fact]
public void ApplyStep_happy_path_appends_journal_runs_effects_saves_snapshot_forwards_state()
public void ApplyStep_happy_path_appends_journal_runs_effects_forwards_state()
{
using var rig = new Rig();
var sut = rig.Build();
@@ -107,13 +133,51 @@ namespace AutopilotMonitor.Agent.V2.Core.Tests.Orchestration
Assert.Equal(1, rig.Effects.CallCount);
Assert.Same(step.NewState, rig.Effects.Calls[0].State);
Assert.Equal(signal.OccurredAtUtc, rig.Effects.Calls[0].OccurredAtUtc);
Assert.Single(rig.Snapshot.Saved);
Assert.Same(step.NewState, rig.Snapshot.Saved[0]);
// H1a: SessionStarted is a pure pass-through (stage unchanged, no effects) → the
// snapshot cache is intentionally NOT written; recovery replays this trivially from
// the SignalLog. The authoritative writes (journal above) still happen per step.
Assert.Empty(rig.Snapshot.Saved);
Assert.Same(step.NewState, sut.CurrentState);
Assert.Equal(0, sut.ConsecutiveJournalFailureCount);
Assert.Empty(rig.Quarantine.Reasons);
}
// ========================================================================= Snapshot policy (H1a)
[Fact]
public void ApplyStep_snapshots_on_stage_change()
{
using var rig = new Rig();
var sut = rig.Build();
var (step, signal) = rig.ReduceAbort(); // SessionStarted -> Failed
sut.ApplyStep(step, signal);
Assert.Single(rig.Snapshot.Saved);
Assert.Same(step.NewState, rig.Snapshot.Saved[0]);
}
[Fact]
public void ApplyStep_skips_snapshot_for_informational_pass_throughs_until_floor()
{
using var rig = new Rig();
var sut = rig.Build();
// SnapshotPassThroughFloor consecutive pass-throughs → none snapshotted.
for (long i = 0; i < DecisionStepProcessor.SnapshotPassThroughFloor; i++)
{
var (step, signal) = rig.ReduceInformational(sut, i);
sut.ApplyStep(step, signal);
}
Assert.Equal(0, rig.Snapshot.SaveCallCount);
// The next pass-through crosses the floor → exactly one forced snapshot (bounds the
// worst-case restart replay tail), then the counter resets.
var (s2, sig2) = rig.ReduceInformational(sut, DecisionStepProcessor.SnapshotPassThroughFloor);
sut.ApplyStep(s2, sig2);
Assert.Equal(1, rig.Snapshot.SaveCallCount);
}
// ========================================================================= Journal failures
[Fact]
@@ -183,7 +247,8 @@ namespace AutopilotMonitor.Agent.V2.Core.Tests.Orchestration
using var rig = new Rig();
rig.Snapshot.ScriptThrow(new InvalidOperationException("snapshot-disk-full"));
var sut = rig.Build();
var (step, signal) = rig.ReduceSessionStarted();
// Use a stage-changing step so the snapshot actually runs (H1a skips pass-throughs).
var (step, signal) = rig.ReduceAbort();
sut.ApplyStep(step, signal); // does NOT throw
@@ -0,0 +1,84 @@
using System.Collections.Generic;
using AutopilotMonitor.Agent.V2.Core.Orchestration;
using AutopilotMonitor.DecisionCore.Engine;
using AutopilotMonitor.DecisionCore.Signals;
using Xunit;
namespace AutopilotMonitor.Agent.V2.Core.Tests.Orchestration
{
/// <summary>
/// Shared activity predicate for the two activity-gated hosts (PeriodicCollectorLifecycleHost
/// idle-stop + StallProbeHost idle clock). Reviews PERF-H3 (the 30 s classifier_tick must NOT
/// count as activity, or it defeats idle-stop) and MON-A2.
/// </summary>
public sealed class SignalActivityClassifierTests
{
[Fact]
public void DeadlineFired_is_never_activity()
{
Assert.False(SignalActivityClassifier.IsRealActivity(DecisionSignalKind.DeadlineFired, null));
// The classifier_tick is a DeadlineFired with a deadline-name payload — still not activity.
Assert.False(SignalActivityClassifier.IsRealActivity(
DecisionSignalKind.DeadlineFired,
new Dictionary<string, string> { [SignalPayloadKeys.Deadline] = "classifier_tick" }));
}
[Theory]
// Periodic self-emissions.
[InlineData("performance_snapshot")]
[InlineData("agent_metrics_snapshot")]
[InlineData("performance_collector_stopped")]
[InlineData("agent_metrics_collector_stopped")]
[InlineData("stall_probe_check")]
[InlineData("stall_probe_result")]
[InlineData("session_stalled")]
// Agent health / control / transport — not device/enrollment progress (P2).
[InlineData("collector_degraded")]
[InlineData("telemetry_upload_poisoned")]
[InlineData("telemetry_upload_blocked")]
[InlineData("state_quarantine_recovered")]
[InlineData("prior_run_died_with_state")]
[InlineData("previous_crash_detected")]
[InlineData("spool_pressure_detected")]
[InlineData("ingress_backpressure")]
[InlineData("agent_shutting_down")]
public void Non_activity_informational_events_are_not_activity(string eventType)
{
var payload = new Dictionary<string, string> { [SignalPayloadKeys.EventType] = eventType };
Assert.False(SignalActivityClassifier.IsRealActivity(DecisionSignalKind.InformationalEvent, payload));
}
[Fact]
public void Periodic_match_is_case_insensitive()
{
var payload = new Dictionary<string, string> { [SignalPayloadKeys.EventType] = "Performance_Snapshot" };
Assert.False(SignalActivityClassifier.IsRealActivity(DecisionSignalKind.InformationalEvent, payload));
}
[Fact]
public void Non_periodic_informational_event_is_activity()
{
var payload = new Dictionary<string, string> { [SignalPayloadKeys.EventType] = "esp_phase_changed" };
Assert.True(SignalActivityClassifier.IsRealActivity(DecisionSignalKind.InformationalEvent, payload));
}
[Fact]
public void Informational_event_without_payload_or_eventType_is_activity()
{
Assert.True(SignalActivityClassifier.IsRealActivity(DecisionSignalKind.InformationalEvent, null));
Assert.True(SignalActivityClassifier.IsRealActivity(
DecisionSignalKind.InformationalEvent, new Dictionary<string, string>()));
}
[Theory]
[InlineData(DecisionSignalKind.EspPhaseChanged)]
[InlineData(DecisionSignalKind.DesktopArrived)]
[InlineData(DecisionSignalKind.HelloResolved)]
[InlineData(DecisionSignalKind.AppInstallCompleted)]
[InlineData(DecisionSignalKind.SystemRebootObserved)]
public void Other_signal_kinds_are_activity(DecisionSignalKind kind)
{
Assert.True(SignalActivityClassifier.IsRealActivity(kind, null));
}
}
}
@@ -50,6 +50,24 @@ namespace AutopilotMonitor.Agent.V2.Core.Tests.Transport
return this;
}
public FakeBackendTelemetryUploader QueueUnauthorized(string reason = "fake-unauthorized")
{
_script.Enqueue(_ => UploadResult.Unauthorized(reason));
return this;
}
public FakeBackendTelemetryUploader QueueTooLarge(string reason = "fake-413")
{
_script.Enqueue(_ => UploadResult.TooLarge(reason));
return this;
}
public FakeBackendTelemetryUploader QueuePoison(string[] rejectedRowKeys, string reason = "fake-poison")
{
_script.Enqueue(_ => UploadResult.Poison(rejectedRowKeys, reason));
return this;
}
public FakeBackendTelemetryUploader QueueThrow(Exception ex)
{
_script.Enqueue(_ => throw ex);
@@ -127,20 +127,178 @@ namespace AutopilotMonitor.Agent.V2.Core.Tests.Transport
}
[Fact]
public async Task Permanent_failure_aborts_without_retrying()
public async Task Unknown_permanent_failure_RETAINS_batch_and_reports_blocked_once()
{
// P1: an unknown permanent 4xx (400 contract bug / 404 route / tenant-validation) must
// NOT discard data on a guess. The cursor stays put (batch retained), and a one-shot
// UploadBlocked event surfaces the stuck pipeline. No poison skip.
using var tmp = new TempDirectory();
var spool = new TelemetrySpool(tmp.Path, Clock());
spool.Enqueue(Draft("only"));
var uploader = new FakeBackendTelemetryUploader().QueuePermanent("bad-data");
var uploader = new FakeBackendTelemetryUploader().QueuePermanent("http 400: contract").QueuePermanent("http 400: contract");
using var sut = new TelemetryUploadOrchestrator(spool, uploader, Clock());
PoisonReport? poison = null;
int blockedCount = 0;
sut.BatchPoisoned += (_, p) => poison = p;
sut.UploadBlocked += (_, __) => blockedCount++;
var first = await sut.DrainAllAsync();
Assert.False(first.Success);
Assert.Equal(1, uploader.CallCount); // no retries for permanent
Assert.Equal(-1, sut.LastUploadedItemId); // RETAINED — cursor NOT advanced
Assert.Null(poison); // never discarded
Assert.Equal(1, blockedCount); // surfaced once
// Second drain still blocked, but the blocked event is one-shot per process.
var second = await sut.DrainAllAsync();
Assert.Equal(-1, sut.LastUploadedItemId);
Assert.Equal(1, blockedCount);
}
[Fact]
public async Task TooLarge_splits_batch_and_uploads_everything()
{
// P1: 413 is a size problem, not poison. Split & retry — never discard a multi-item batch.
using var tmp = new TempDirectory();
var spool = new TelemetrySpool(tmp.Path, Clock());
for (int i = 0; i < 4; i++) spool.Enqueue(Draft($"r{i}"));
// First attempt (batch of 4) → 413; orchestrator halves to 2 and the two halves upload.
var uploader = new FakeBackendTelemetryUploader().QueueTooLarge().QueueOk().QueueOk();
using var sut = new TelemetryUploadOrchestrator(spool, uploader, Clock(), batchSize: 4);
var result = await sut.DrainAllAsync();
Assert.True(result.Success);
Assert.Equal(4, result.UploadedItems); // nothing lost
Assert.Equal(3, sut.LastUploadedItemId);
Assert.Equal(3, uploader.CallCount); // 1 oversized + 2 half-batches
}
[Fact]
public async Task TooLarge_single_item_is_quarantined_as_oversize()
{
// The ONE locally-provable poison: a lone item the backend rejects with 413 even on its
// own can never be sent — quarantine just it (advance by one) and surface it.
using var tmp = new TempDirectory();
var spool = new TelemetrySpool(tmp.Path, Clock());
spool.Enqueue(Draft("huge"));
var uploader = new FakeBackendTelemetryUploader().QueueTooLarge("http 413");
using var sut = new TelemetryUploadOrchestrator(spool, uploader, Clock());
PoisonReport? poison = null;
sut.BatchPoisoned += (_, p) => poison = p;
var result = await sut.DrainAllAsync();
Assert.Equal(0, sut.LastUploadedItemId); // advanced past the single oversized item
Assert.NotNull(poison);
Assert.Equal(1, poison!.ItemCount);
Assert.Contains("oversize", poison.Reason);
Assert.Equal(PoisonKind.Oversize, poison.Kind); // P3
}
[Fact]
public async Task Explicit_item_poison_drops_named_items_uploads_survivors_and_advances()
{
// P1: an EXPLICIT item-level poison signal is the only authorized discard. The named
// RowKey is dropped; the rest of the batch is re-uploaded; the cursor advances past all.
using var tmp = new TempDirectory();
var spool = new TelemetrySpool(tmp.Path, Clock());
spool.Enqueue(Draft("r0"));
spool.Enqueue(Draft("r1"));
spool.Enqueue(Draft("r2"));
// First call (batch r0,r1,r2) → poison r1; survivors (r0,r2) re-upload → OK.
var uploader = new FakeBackendTelemetryUploader()
.QueuePoison(new[] { "r1" }, "schema_validation_failed")
.QueueOk();
using var sut = new TelemetryUploadOrchestrator(spool, uploader, Clock(), batchSize: 100);
PoisonReport? poison = null;
sut.BatchPoisoned += (_, p) => poison = p;
var result = await sut.DrainAllAsync();
Assert.Equal(2, result.UploadedItems); // r0 + r2 survived
Assert.Equal(2, sut.LastUploadedItemId); // cursor advanced past the whole batch
Assert.Equal(2, uploader.CallCount); // poison call + survivor re-upload
Assert.NotNull(poison);
Assert.Equal(1, poison!.ItemCount); // exactly one item dropped (r1)
Assert.Equal(PoisonKind.BackendRejected, poison.Kind); // P3 — not mislabeled "oversize"
}
[Fact]
public async Task Poison_survivor_reupload_honours_backend_control_signals()
{
// P1 fix: a DeviceBlocked (or kill/admin/actions) response on the SURVIVOR re-upload must
// be honoured exactly like the normal success path — previously it was silently dropped.
using var tmp = new TempDirectory();
var spool = new TelemetrySpool(tmp.Path, Clock());
spool.Enqueue(Draft("r0"));
spool.Enqueue(Draft("r1"));
// Poison r1; the survivor (r0) re-upload comes back with a DeviceBlocked quarantine.
var uploader = new FakeBackendTelemetryUploader()
.QueuePoison(new[] { "r1" }, "schema")
.QueueOkWithSignals(deviceBlocked: true);
using var sut = new TelemetryUploadOrchestrator(spool, uploader, Clock(), batchSize: 100);
UploadResult? signalled = null;
sut.ServerResponseReceived += (_, r) => signalled = r;
await sut.DrainAllAsync();
Assert.True(sut.IsPaused); // DeviceBlocked from the survivor upload took effect
Assert.NotNull(signalled); // and was surfaced via ServerResponseReceived
Assert.True(signalled!.DeviceBlocked);
}
[Fact]
public async Task Poison_rowkeys_not_in_batch_retains_everything()
{
// A mismatched / stale poison list must not cause data loss — nothing named is in the
// batch, so we retain (cursor stays) rather than discard or advance.
using var tmp = new TempDirectory();
var spool = new TelemetrySpool(tmp.Path, Clock());
spool.Enqueue(Draft("r0"));
spool.Enqueue(Draft("r1"));
var uploader = new FakeBackendTelemetryUploader().QueuePoison(new[] { "not-in-batch" }, "stale");
using var sut = new TelemetryUploadOrchestrator(spool, uploader, Clock(), batchSize: 100);
PoisonReport? poison = null;
sut.BatchPoisoned += (_, p) => poison = p;
var result = await sut.DrainAllAsync();
Assert.False(result.Success);
Assert.Equal(1, uploader.CallCount); // no retries for permanent
Assert.Equal(-1, sut.LastUploadedItemId);
Assert.Equal(-1, sut.LastUploadedItemId); // RETAINED — nothing advanced or dropped
Assert.Null(poison);
Assert.Equal(1, uploader.CallCount); // no survivor re-upload (whole batch retained)
}
[Fact]
public async Task Auth_permanent_failure_retains_cursor_and_is_not_blocked_or_poisoned()
{
// 401/403: cursor stays (retained), no poison, and NOT counted as a "blocked" pipeline —
// the uploader already drove AuthFailureTracker → shutdown, which owns the outcome.
using var tmp = new TempDirectory();
var spool = new TelemetrySpool(tmp.Path, Clock());
spool.Enqueue(Draft("only"));
var uploader = new FakeBackendTelemetryUploader().QueueUnauthorized("unauthorized: http 401");
using var sut = new TelemetryUploadOrchestrator(spool, uploader, Clock());
PoisonReport? poison = null;
int blockedCount = 0;
sut.BatchPoisoned += (_, p) => poison = p;
sut.UploadBlocked += (_, __) => blockedCount++;
var result = await sut.DrainAllAsync();
Assert.False(result.Success);
Assert.Equal(1, uploader.CallCount); // no retries
Assert.Equal(-1, sut.LastUploadedItemId); // cursor NOT advanced
Assert.Null(poison);
Assert.Equal(0, blockedCount); // auth path owns shutdown, not a "block"
}
[Fact]
@@ -16,21 +16,51 @@ namespace AutopilotMonitor.Agent.V2.Core.Monitoring.Enrollment.Ime
/// </summary>
public partial class ImeLogTracker
{
// M2: precompiled matchers for LogFilePatterns so CheckLogFilesAsync can filter a single
// directory enumeration in-memory. Globs use '?' (single char) in the patterns; '*' is
// supported too for forward-compatibility.
private static readonly Regex[] LogFilePatternRegexes = BuildLogFilePatternRegexes();
private static Regex[] BuildLogFilePatternRegexes()
{
var result = new Regex[LogFilePatterns.Length];
for (var i = 0; i < LogFilePatterns.Length; i++)
{
var escaped = Regex.Escape(LogFilePatterns[i]).Replace("\\?", ".").Replace("\\*", ".*");
result[i] = new Regex("^" + escaped + "$", RegexOptions.IgnoreCase | RegexOptions.Compiled);
}
return result;
}
private static bool MatchesLogFilePattern(string fileName)
{
foreach (var rx in LogFilePatternRegexes)
{
if (rx.IsMatch(fileName)) return true;
}
return false;
}
private async Task CheckLogFilesAsync(CancellationToken token)
{
if (!Directory.Exists(_logFolder))
return;
// Get all matching log files, sorted by name (archived files come before current)
// Get all matching log files, sorted by name (archived files come before current).
// M2: a SINGLE directory enumeration filtered in-memory against the patterns, instead
// of one Directory.GetFiles(pattern) per LogFilePattern every 100 ms poll (~90 folder
// enumerations/s → 1). The in-memory regex match is anchored, so it is also immune to
// the Win32 "*.ext" 8.3 search-pattern quirk.
var files = new List<string>();
foreach (var pattern in LogFilePatterns)
try
{
try
foreach (var filePath in Directory.EnumerateFiles(_logFolder))
{
files.AddRange(Directory.GetFiles(_logFolder, pattern));
if (MatchesLogFilePattern(Path.GetFileName(filePath)))
files.Add(filePath);
}
catch (DirectoryNotFoundException) { }
}
catch (DirectoryNotFoundException) { }
files.Sort(StringComparer.OrdinalIgnoreCase);
@@ -46,10 +76,14 @@ namespace AutopilotMonitor.Agent.V2.Core.Monitoring.Enrollment.Ime
var startPos = _positionTracker.GetSafePosition(filePath, fileInfo.Length);
if (startPos >= fileInfo.Length)
{
_logger.Trace($"ImeLogTracker: {Path.GetFileName(filePath)} — no new data (pos={startPos}, size={fileInfo.Length})");
// M2: guard the interpolated string so it isn't built every 100 ms tick
// (per file) when Trace is off — which is the production default (Info).
if (_logger.LogLevel >= AgentLogLevel.Trace)
_logger.Trace($"ImeLogTracker: {Path.GetFileName(filePath)} — no new data (pos={startPos}, size={fileInfo.Length})");
continue;
}
_logger.Trace($"ImeLogTracker: reading {Path.GetFileName(filePath)} from pos {startPos} (size={fileInfo.Length}, delta={fileInfo.Length - startPos})");
if (_logger.LogLevel >= AgentLogLevel.Trace)
_logger.Trace($"ImeLogTracker: reading {Path.GetFileName(filePath)} from pos {startPos} (size={fileInfo.Length}, delta={fileInfo.Length - startPos})");
using (var stream = new FileStream(filePath, FileMode.Open, FileAccess.Read, FileShare.ReadWrite))
{
@@ -81,6 +81,16 @@ namespace AutopilotMonitor.Agent.V2.Core.Monitoring.Enrollment.Ime
private readonly ImeTrackerStatePersistence _statePersistence;
private bool _stateDirty;
// M1: the state file is a read-position bookmark + package-state cache, NOT a source of
// truth (the IME logs themselves are written by Windows and always on disk). Saving it on
// every 100 ms poll while IME streams progress lines hammers the disk during the most
// contended phase of enrollment. Throttle to at most once per this interval; a save that
// is throttled away leaves _stateDirty set so the next eligible cycle (or the unconditional
// final save on shutdown) persists it. Worst case on restart: ~this-many-seconds of already
// processed log lines are re-read and re-emitted (deduped downstream) — no data loss.
private static readonly TimeSpan StateSaveMinInterval = TimeSpan.FromSeconds(2);
private DateTime _lastStateSaveUtc = DateTime.MinValue;
// Standard GUID capture pattern used as {GUID} placeholder in patterns
private const string GuidPattern = @"(?<id>[a-z0-9]{8}-[a-z0-9]{4}-[a-z0-9]{4}-[a-z0-9]{4}-[a-z0-9]{12})";
@@ -382,11 +392,15 @@ namespace AutopilotMonitor.Agent.V2.Core.Monitoring.Enrollment.Ime
try { await Task.Delay(1000, token); } catch (OperationCanceledException) { break; }
}
// Persist state after each polling cycle so agent restart continues from here
if (_stateDirty)
// Persist state after each polling cycle so agent restart continues from here.
// M1: throttled to at most once per StateSaveMinInterval — a throttled-away save
// keeps _stateDirty set for the next eligible cycle; the final save on shutdown
// (below) is unconditional so nothing pending is lost on a clean exit.
if (_stateDirty && (DateTime.UtcNow - _lastStateSaveUtc) >= StateSaveMinInterval)
{
SaveState();
_stateDirty = false;
_lastStateSaveUtc = DateTime.UtcNow;
}
try { await Task.Delay(_pollingIntervalMs, token); } catch (OperationCanceledException) { break; }
@@ -63,9 +63,18 @@ namespace AutopilotMonitor.Agent.V2.Core.Monitoring.Enrollment.Ime
{
Directory.CreateDirectory(_stateDirectory);
var tempPath = _stateFilePath + ".tmp";
File.WriteAllText(tempPath, JsonConvert.SerializeObject(state, Formatting.Indented));
File.Copy(tempPath, _stateFilePath, overwrite: true);
try { File.Delete(tempPath); } catch { }
// M1: Formatting.None (was Indented — this state is machine-only, ~halves the
// bytes for 30+ packages) and an atomic File.Replace/Move instead of File.Copy
// (which read+wrote the whole file again AND was non-atomic).
File.WriteAllText(tempPath, JsonConvert.SerializeObject(state));
if (File.Exists(_stateFilePath))
{
File.Replace(tempPath, _stateFilePath, destinationBackupFileName: null, ignoreMetadataErrors: true);
}
else
{
File.Move(tempPath, _stateFilePath);
}
}
catch (Exception ex)
{
@@ -0,0 +1,74 @@
using System;
using System.Collections.Generic;
using AutopilotMonitor.Agent.V2.Core.Orchestration;
using AutopilotMonitor.Shared;
using AutopilotMonitor.Shared.Models;
namespace AutopilotMonitor.Agent.V2.Core.Monitoring.Enrollment.SystemSignals
{
/// <summary>
/// Emits a single <c>collector_degraded</c> Warning event when a collector/watcher fails to
/// arm or operate. Review MON-D1: kernel event-log/registry watcher failures were previously
/// local-log only, so a dead Shell-Core / ModernDeployment / Hello watcher looked identical on
/// the backend to a genuine no-signal enrollment until the 5 h maintenance sweep. This surfaces
/// the failure as telemetry without violating the no-heartbeat rule — it is one-shot,
/// state-change-only (fired on the failure, not periodically).
/// </summary>
internal static class CollectorDegradationReporter
{
/// <summary>
/// Best-effort emit of a <c>collector_degraded</c> Warning. Never throws — it runs from
/// failure paths where an additional exception would be worse than a missing event.
/// </summary>
/// <param name="post">Telemetry post target (no-op if null).</param>
/// <param name="sessionId">Session correlation id.</param>
/// <param name="tenantId">Tenant correlation id.</param>
/// <param name="collectorName">The collector/watcher that degraded (also the event Source).</param>
/// <param name="reason">Short machine-readable reason (e.g. <c>watcher_arm_failed</c>).</param>
/// <param name="ex">Optional underlying exception — its type + message are attached.</param>
public static void Report(
InformationalEventPost post,
string sessionId,
string tenantId,
string collectorName,
string reason,
Exception ex = null)
{
if (post == null) return;
try
{
var data = new Dictionary<string, object>
{
{ "collector", collectorName },
{ "reason", reason },
};
if (ex != null)
{
data["error"] = ex.Message;
data["errorType"] = ex.GetType().Name;
}
var message = ex != null
? $"Collector '{collectorName}' degraded: {reason} ({ex.GetType().Name}: {ex.Message})"
: $"Collector '{collectorName}' degraded: {reason}";
post.Emit(new EnrollmentEvent
{
SessionId = sessionId,
TenantId = tenantId,
EventType = Constants.EventTypes.CollectorDegraded,
Severity = EventSeverity.Warning,
Source = collectorName,
Phase = EnrollmentPhase.Unknown,
Message = message,
Data = data,
});
}
catch
{
// Best-effort observability — never throw from a failure path.
}
}
}
}
@@ -31,6 +31,7 @@ namespace AutopilotMonitor.Agent.V2.Core.Monitoring.Enrollment.SystemSignals
{
private readonly AgentLogger _logger;
private Timer _pollingTimer;
private int _pollInProgress; // Interlocked reentrancy guard for PollForDesktop (ARCH-F8)
private bool _desktopArrived;
private bool _excludedUserTraced; // Emit trace event only once for excluded-user skips
private const int PollingIntervalSeconds = 30;
@@ -160,6 +161,14 @@ namespace AutopilotMonitor.Agent.V2.Core.Monitoring.Enrollment.SystemSignals
if (_desktopArrived)
return;
// Reentrancy guard (review ARCH-F8): WMI GetOwner can run long enough that the 30 s
// timer queues a second callback before the first returns. Without this, two
// overlapping polls could both pass the _desktopArrived check and double-fire
// RealUserOwnerObserved → RealmJoin HKU-watcher arming. Skip if a poll is in flight;
// the finally below always clears the flag (covers the mid-method success return).
if (Interlocked.CompareExchange(ref _pollInProgress, 1, 0) != 0)
return;
_pollCount++;
try
@@ -264,6 +273,10 @@ namespace AutopilotMonitor.Agent.V2.Core.Monitoring.Enrollment.SystemSignals
// count carries the failure mode to the backend.
EmitFirstPollIfNeeded(explorerProcessCount: 0);
}
finally
{
Interlocked.Exchange(ref _pollInProgress, 0);
}
}
/// <summary>
@@ -467,6 +467,9 @@ namespace AutopilotMonitor.Agent.V2.Core.Monitoring.Enrollment.SystemSignals
catch (Exception ex)
{
_logger.Error("Failed to start Hello event log watcher", ex);
// MON-D1: a dead UDR watcher silently drops Hello provisioning evidence.
CollectorDegradationReporter.Report(_post, _sessionId, _tenantId,
collectorName: "HelloTracker", reason: "udr_watcher_arm_failed", ex: ex);
}
}
@@ -746,6 +749,9 @@ namespace AutopilotMonitor.Agent.V2.Core.Monitoring.Enrollment.SystemSignals
catch (Exception ex)
{
_logger.Error("Failed to start HelloForBusiness event log watcher", ex);
// MON-D1: a dead HelloForBusiness watcher silently drops Hello completion evidence.
CollectorDegradationReporter.Report(_post, _sessionId, _tenantId,
collectorName: "HelloTracker", reason: "hellobusiness_watcher_arm_failed", ex: ex);
}
}
@@ -143,6 +143,9 @@ namespace AutopilotMonitor.Agent.V2.Core.Monitoring.Enrollment.SystemSignals
catch (Exception ex)
{
_logger.Error($"Failed to start ModernDeployment watcher for {channelName}", ex);
// MON-D1: surface a dead ModernDeployment channel watcher as one-shot telemetry.
CollectorDegradationReporter.Report(_post, _sessionId, _tenantId,
collectorName: "ModernDeploymentTracker", reason: $"watcher_arm_failed:{channelName}", ex: ex);
return null;
}
}
@@ -109,6 +109,11 @@ namespace AutopilotMonitor.Agent.V2.Core.Monitoring.Enrollment.SystemSignals
catch (Exception ex)
{
_logger.Error("Failed to start Shell-Core event log watcher", ex);
// MON-D1: a dead Shell-Core watcher means the session never observes ESP exit /
// WhiteGlove success — indistinguishable on the backend from a real no-signal
// enrollment. Surface it as one-shot telemetry.
CollectorDegradationReporter.Report(_post, _sessionId, _tenantId,
collectorName: "ShellCoreTracker", reason: "watcher_arm_failed", ex: ex);
}
}
@@ -7,6 +7,7 @@ using System.Text;
using System.Text.RegularExpressions;
using System.Threading;
using AutopilotMonitor.Agent.V2.Core.Logging;
using AutopilotMonitor.Agent.V2.Core.Monitoring.Enrollment.Ime;
using AutopilotMonitor.Agent.V2.Core.Orchestration;
using AutopilotMonitor.Shared;
using AutopilotMonitor.Shared.Models;
@@ -55,6 +56,11 @@ namespace AutopilotMonitor.Agent.V2.Core.Monitoring.Enrollment.SystemSignals
private readonly HashSet<int> _firedProbeIndices = new HashSet<int>();
private bool _sessionStalledFired;
// P2 — monotonic generation, bumped by ResetProbes. RunProbe runs OUTSIDE _stateLock; if a
// real-activity reset lands while a probe is in flight, the post-probe "mark fired" must be
// dropped (its generation is stale) so the next idle period can fire that probe again.
private int _probeGeneration;
public event EventHandler<string> EspFailureDetected;
// Per-source hard timeouts (ms)
@@ -67,6 +73,13 @@ namespace AutopilotMonitor.Agent.V2.Core.Monitoring.Enrollment.SystemSignals
private const int MaxRawSamples = 5;
private const int MaxActiveInstalls = 10;
// Active-install age window (review MON-B8). An AppWorkload "EnforcementState: Installing"
// line older than this is no longer evidence of progress — it lets a wedged install whose
// last line lingers in the 200 KB tail keep suppressing session_stalled forever. Lines
// with a parsed CMTrace timestamp older than this are excluded; lines whose age we cannot
// determine stay counted, so a long-but-active installer is never mis-flagged as stalled.
private const int ActiveInstallFreshnessMinutes = 15;
// Terminal ESP failure EventIDs in ModernDeployment-Diagnostics-Provider channels.
// Conservative list — we prefer to NOT auto-terminate on unknown IDs and let the user decide.
//
@@ -162,6 +175,9 @@ namespace AutopilotMonitor.Agent.V2.Core.Monitoring.Enrollment.SystemSignals
{
lock (_stateLock)
{
// Always bump the generation, even when no probes are currently marked fired — a
// probe may be running right now (outside the lock) and must not re-mark itself.
_probeGeneration++;
if (_firedProbeIndices.Count > 0)
{
_logger.Trace($"StallProbeCollector: resetting {_firedProbeIndices.Count} fired probes");
@@ -186,9 +202,11 @@ namespace AutopilotMonitor.Agent.V2.Core.Monitoring.Enrollment.SystemSignals
int probeIndex = i + 1; // 1-based for user-facing output
bool alreadyFired;
int genAtStart;
lock (_stateLock)
{
alreadyFired = _firedProbeIndices.Contains(probeIndex);
genAtStart = _probeGeneration;
}
if (alreadyFired)
continue;
@@ -197,7 +215,10 @@ namespace AutopilotMonitor.Agent.V2.Core.Monitoring.Enrollment.SystemSignals
lock (_stateLock)
{
_firedProbeIndices.Add(probeIndex);
// P2: only mark fired if no ResetProbes (real activity) landed while this probe
// ran — otherwise the next genuine idle period would wrongly skip this index.
if (_probeGeneration == genAtStart)
_firedProbeIndices.Add(probeIndex);
}
}
}
@@ -484,11 +505,15 @@ namespace AutopilotMonitor.Agent.V2.Core.Monitoring.Enrollment.SystemSignals
if (firstNewline > 0 && firstNewline < content.Length - 1)
content = content.Substring(firstNewline + 1);
// Positive-signal scan: active installations recently logged.
foreach (Match m in ActiveInstallRegex.Matches(content))
// Positive-signal scan: active installations recently logged. Age-filtered so a
// wedged install whose last "Installing" line lingers in the tail stops counting
// as progress once it is stale (review MON-B8).
foreach (var marker in FilterFreshActiveInstalls(content, DateTime.UtcNow, ActiveInstallFreshnessMinutes))
{
if (result.ActiveInstalls.Count < MaxActiveInstalls && !result.ActiveInstalls.Contains(m.Value))
result.ActiveInstalls.Add(m.Value);
if (result.ActiveInstalls.Count >= MaxActiveInstalls)
break;
if (!result.ActiveInstalls.Contains(marker))
result.ActiveInstalls.Add(marker);
}
// Negative-signal scan: failures.
@@ -516,6 +541,41 @@ namespace AutopilotMonitor.Agent.V2.Core.Monitoring.Enrollment.SystemSignals
}
}
/// <summary>
/// Extracts the active-install markers (<c>EnforcementState: Installing/Downloading/InProgress</c>)
/// from an AppWorkload.log tail, dropping any whose CMTrace timestamp is provably older than
/// <paramref name="freshnessMinutes"/>. Conservative by design: a line is excluded ONLY when
/// it parses as CMTrace AND its timestamp is older than the cutoff. Lines that do not parse,
/// or whose timestamp cannot be determined, are treated as fresh — so a long-but-genuinely-
/// active installer is never mis-classified as stalled (review MON-B8). Internal for testing.
/// </summary>
internal static List<string> FilterFreshActiveInstalls(string content, DateTime nowUtc, int freshnessMinutes)
{
var fresh = new List<string>();
if (string.IsNullOrEmpty(content))
return fresh;
var cutoffUtc = nowUtc - TimeSpan.FromMinutes(freshnessMinutes);
foreach (var rawLine in content.Split('\n'))
{
var scanText = rawLine;
if (CmTraceLogParser.TryParseLine(rawLine, out var entry))
{
// Provably stale → not progress. (TryParseLine falls back to now on an
// unparseable timestamp, which keeps the line fresh — the safe default.)
if (entry.Timestamp < cutoffUtc)
continue;
scanText = entry.Message;
}
var m = ActiveInstallRegex.Match(scanText);
if (m.Success)
fresh.Add(m.Value);
}
return fresh;
}
private void EmitProbeResultEvent(ProbeResult result, EventSeverity severity)
{
var data = BuildDataJson(result);
@@ -49,40 +49,63 @@ Start-Sleep -Seconds 2
Wait-Process -Id {currentProcessId} -ErrorAction SilentlyContinue -Timeout 30
Start-Sleep -Seconds 1
# Remove Scheduled Task
try {{
Stop-ScheduledTask -TaskName '{_configuration.ScheduledTaskName}' -ErrorAction SilentlyContinue
Unregister-ScheduledTask -TaskName '{_configuration.ScheduledTaskName}' -Confirm:$false -ErrorAction SilentlyContinue
}} catch {{ }}
# Remove Scheduled Task — verify removal BEFORE deleting files (LIFE-F6). An orphan task firing
# against a deleted exe spams Task Scheduler errors every boot forever and, once the files (incl.
# the enrollment-complete marker) are gone, has no cleanup-retry path. So confirm the task is gone
# first; if it cannot be removed, SKIP the file deletion and leave everything intact for a clean
# cleanup retry on the next boot (the marker survives, so bootstrap re-runs cleanup).
$taskName = '{_configuration.ScheduledTaskName}'
$taskGone = $false
for ($i = 1; $i -le 5; $i++) {{
try {{ Stop-ScheduledTask -TaskName $taskName -ErrorAction SilentlyContinue }} catch {{ }}
try {{ Unregister-ScheduledTask -TaskName $taskName -Confirm:$false -ErrorAction SilentlyContinue }} catch {{ }}
$still = $null
try {{ $still = Get-ScheduledTask -TaskName $taskName -ErrorAction SilentlyContinue }} catch {{ }}
if (-not $still) {{ $taskGone = $true; break }}
Start-Sleep -Seconds 1
}}
if (-not $taskGone) {{
# Fallback: schtasks.exe /Delete does not depend on the ScheduledTasks PowerShell module.
try {{ & schtasks.exe /Delete /TN $taskName /F 2>$null | Out-Null }} catch {{ }}
try {{ $taskGone = -not (Get-ScheduledTask -TaskName $taskName -ErrorAction SilentlyContinue) }} catch {{ }}
}}
{(keepLogs ? $@"
# Delete everything except Logs directory
Get-ChildItem -Path '{agentBasePath}' -Exclude 'Logs' | ForEach-Object {{
$dest = $_.FullName + '.del'
try {{ Rename-Item -Path $_.FullName -NewName $dest -Force -ErrorAction SilentlyContinue }} catch {{ }}
Remove-Item -Path $dest -Recurse -Force -ErrorAction SilentlyContinue
}}
" : $@"
# Rename folder then delete. Retry a few times to let the OS release handles.
$renamedPath = '{agentBasePath}.del'
$renamed = $false
for ($i = 1; $i -le 10; $i++) {{
try {{
Rename-Item -Path '{agentBasePath}' -NewName $renamedPath -Force -ErrorAction Stop
$renamed = $true
break
}} catch {{
Start-Sleep -Seconds 2
if ($taskGone) {{
# Delete everything except Logs directory
Get-ChildItem -Path '{agentBasePath}' -Exclude 'Logs' | ForEach-Object {{
$dest = $_.FullName + '.del'
try {{ Rename-Item -Path $_.FullName -NewName $dest -Force -ErrorAction SilentlyContinue }} catch {{ }}
Remove-Item -Path $dest -Recurse -Force -ErrorAction SilentlyContinue
}}
}}
if ($renamed) {{
Remove-Item -Path $renamedPath -Recurse -Force -ErrorAction SilentlyContinue
}} else {{
Remove-Item -Path '{agentBasePath}' -Recurse -Force -ErrorAction SilentlyContinue
" : $@"
if ($taskGone) {{
# Rename folder then delete. Retry a few times to let the OS release handles.
$renamedPath = '{agentBasePath}.del'
$renamed = $false
for ($i = 1; $i -le 10; $i++) {{
try {{
Rename-Item -Path '{agentBasePath}' -NewName $renamedPath -Force -ErrorAction Stop
$renamed = $true
break
}} catch {{
Start-Sleep -Seconds 2
}}
}}
if ($renamed) {{
Remove-Item -Path $renamedPath -Recurse -Force -ErrorAction SilentlyContinue
}} else {{
Remove-Item -Path '{agentBasePath}' -Recurse -Force -ErrorAction SilentlyContinue
}}
}}
")}
{(_configuration.RebootOnComplete ? @"
Restart-Computer -Force -Delay 10 -Comment 'Autopilot enrollment completed - Autopilot Monitor is rebooting'
# shutdown.exe (NOT Restart-Computer): Restart-Computer has no -Comment parameter and -Delay is
# only honoured with -Wait, so the previous invocation failed parameter binding and the device
# silently never rebooted. shutdown.exe /r /t 10 mirrors the standalone reboot path and reboots
# after a 10 s delay (lets this script finish + the agent exit). Review LIFE-F2.
shutdown.exe /r /t 10 /c 'Autopilot enrollment completed - Autopilot Monitor is rebooting'
" : "")}
Remove-Item -Path $scriptPath -Force -ErrorAction SilentlyContinue
";
@@ -46,6 +46,16 @@ namespace AutopilotMonitor.Agent.V2.Core.Orchestration
/// <summary>Default — 3 Failures hintereinander → Quarantine.</summary>
public const int DefaultQuarantineThreshold = 3;
/// <summary>
/// H1a (Wave 2) — the snapshot is a recovery CACHE, not the source of truth (SignalLog +
/// Journal are, both still fsync'd per step). Pure InformationalEvent pass-throughs (Stage
/// unchanged + only a telemetry <see cref="DecisionEffectKind.EmitEventTimelineEntry"/>
/// effect) skip the snapshot — they are replayed from the SignalLog as deterministic state
/// reconstruction on the next start. This floor forces a snapshot at least every N such
/// steps so the worst-case restart replay tail stays bounded.
/// </summary>
public const int SnapshotPassThroughFloor = 100;
private readonly IJournalWriter _journal;
private readonly IEffectRunner _effectRunner;
private readonly ISnapshotPersistence _snapshot;
@@ -59,6 +69,7 @@ namespace AutopilotMonitor.Agent.V2.Core.Orchestration
private int _consecutiveJournalFailures;
private bool _quarantineTriggered;
private bool _terminalNotified;
private int _passThroughStepsSinceSnapshot; // H1a — pass-through steps skipped since last snapshot
public DecisionStepProcessor(
DecisionState initialState,
@@ -186,16 +197,30 @@ namespace AutopilotMonitor.Agent.V2.Core.Orchestration
$"for signal ordinal={signal.SessionSignalOrdinal}.");
}
// 3) Snapshot — best-effort. Journal ist die Wahrheit, Snapshot ist Cache.
try
// 3) Snapshot — best-effort recovery CACHE (Journal ist die Wahrheit). H1a: only
// cache on a meaningful step (Stage transition / non-telemetry effect) or once
// every SnapshotPassThroughFloor pass-through steps. Pure InformationalEvent
// pass-throughs are skipped — replayed from the SignalLog (pure state
// reconstruction, no effects) on the next start, so nothing is lost.
if (ShouldSnapshot(step, previousStage))
{
_snapshot.Save(step.NewState);
try
{
_snapshot.Save(step.NewState);
_passThroughStepsSinceSnapshot = 0;
}
catch (Exception ex)
{
_logger.Warning(
$"DecisionStepProcessor: snapshot save failed (transient, not escalated) " +
$"for signal ordinal={signal.SessionSignalOrdinal}: {ex.Message}");
// Floor counter intentionally NOT reset — a failed save leaves the cache
// stale, so the next step should attempt the snapshot again.
}
}
catch (Exception ex)
else
{
_logger.Warning(
$"DecisionStepProcessor: snapshot save failed (transient, not escalated) " +
$"for signal ordinal={signal.SessionSignalOrdinal}: {ex.Message}");
_passThroughStepsSinceSnapshot++;
}
}
@@ -241,6 +266,29 @@ namespace AutopilotMonitor.Agent.V2.Core.Orchestration
return effectResult;
}
/// <summary>
/// H1a snapshot policy. The snapshot is a recovery cache; skipping a step is always
/// correctness-safe (the SignalLog tail is replayed deterministically on the next start,
/// running no effects). We cache when the step is "meaningful" so the restart replay tail
/// stays short: a Stage transition (covers every terminal stage + the SessionMustAbort
/// replacement), or any effect other than the pure telemetry projection
/// (<see cref="DecisionEffectKind.EmitEventTimelineEntry"/>). Pure InformationalEvent
/// pass-throughs (download_progress, IME/perf snapshots, …) are the high-frequency case we
/// skip; <see cref="SnapshotPassThroughFloor"/> bounds the worst-case replay length.
/// </summary>
private bool ShouldSnapshot(DecisionStep step, SessionStage previousStage)
{
if (step.NewState.Stage != previousStage) return true;
foreach (var effect in step.Effects)
{
if (effect.Kind != DecisionEffectKind.EmitEventTimelineEntry)
return true;
}
return _passThroughStepsSinceSnapshot >= SnapshotPassThroughFloor;
}
private void TryTriggerQuarantine(string reason)
{
try
@@ -387,9 +387,9 @@ namespace AutopilotMonitor.Agent.V2.Core.Orchestration
// signal before the watcher starts (avoids missing the wake for an install already in flight).
if (officeHost != null) hosts.Add(officeHost);
// M4.6.δ — Gather-rules runtime executor. Runs the backend-defined rules whose
// Trigger is "startup" once the agent is up; signal / event / periodic triggers
// remain supported inside the executor itself.
// M4.6.δ — Gather-rules runtime executor. Runs "startup" rules once the agent is up
// and "interval" rules on their timers; "phase_change" / "on_event" rules are driven
// by the host's SignalIngress.SignalPosted subscription (MON-A1).
if (_remoteConfig.GatherRules != null && _remoteConfig.GatherRules.Count > 0)
{
hosts.Add(new GatherRuleExecutorHost(
@@ -134,6 +134,20 @@ namespace AutopilotMonitor.Agent.V2.Core.Orchestration
private bool _quarantineRequested;
private string? _quarantineReason;
// TRACE-H2: marker file written by TriggerQuarantine (mid-run, when the journal can no
// longer be appended) and honoured on the NEXT Start — a running process must not pull the
// state files out from under itself. When the marker is found at startup, all reducer-state
// segments + snapshot are quarantined and the session starts from a clean Initial state.
private const string QuarantineRequestedMarkerFile = "quarantine-requested.marker";
private bool _priorRunQuarantined;
private string? _priorRunQuarantineReason;
// P2: set when a prior-run quarantine was REQUESTED (marker present) but did NOT fully
// succeed this start. The marker demands a clean reset, so we must NOT fall back to normal
// snapshot/log recovery (which could reload the very state we were told to discard) — this
// forces the run to seed from a fresh Initial state instead.
private bool _forceCleanStart;
// Recovery flags (populated during Start()).
// V1-symmetric Part-2 hint (analog to V1 MonitoringService._isWhiteGlovePart2):
// set when Start() detects a persisted WhiteGloveSealed snapshot and archives the
@@ -317,6 +331,50 @@ namespace AutopilotMonitor.Agent.V2.Core.Orchestration
var journalPath = Path.Combine(_stateDirectory, "journal.jsonl");
var eventSequencePath = Path.Combine(_stateDirectory, "event-sequence.json");
// -1) TRACE-H2 — honour a prior run's quarantine request BEFORE the writers scan the
// (suspect) segment files. A prior run hit the consecutive-journal-failure threshold
// and could no longer persist consistently; it left the marker so THIS run resets
// cleanly. Quarantining here (not mid-run) preserves the "a running process never
// pulls its own state files out from under itself" invariant. event-sequence is
// quarantined too — a clean restart re-seeds it from 0 (backend dedups on RowKey).
var quarantineMarkerPath = Path.Combine(_stateDirectory, QuarantineRequestedMarkerFile);
if (File.Exists(quarantineMarkerPath))
{
var quarantineReason = TryReadMarkerReason(quarantineMarkerPath);
_logger.Error(
"EnrollmentOrchestrator: prior-run quarantine marker found — quarantining all state " +
$"segments + snapshot before recovery. Reason: {quarantineReason}");
try
{
SegmentQuarantine.QuarantineAll(
_stateDirectory, "quarantine-requested:" + quarantineReason, () => _clock.UtcNow);
new SnapshotPersistence(snapshotPath, () => _clock.UtcNow)
.Quarantine("quarantine-requested:" + quarantineReason);
// P2: delete the marker ONLY after the quarantine fully succeeded. If QuarantineAll
// or the snapshot quarantine threw (ACL / lock / disk), the marker is LEFT so the
// next start retries fail-closed instead of proceeding on possibly-still-corrupt
// segments with the retry hint lost.
try { File.Delete(quarantineMarkerPath); }
catch (Exception ex) { _logger.Warning($"EnrollmentOrchestrator: quarantine marker delete failed: {ex.Message}"); }
_priorRunQuarantined = true;
_priorRunQuarantineReason = quarantineReason;
}
catch (Exception ex)
{
// Marker intentionally retained — do NOT delete on failure (P2). AND force a
// clean Initial seed for THIS run: a partial failure (e.g. segments moved but the
// snapshot quarantine threw) must not let normal recovery reload the discarded
// snapshot. Fail-closed to fresh state; the retained marker drives a proper retry
// next start.
_forceCleanStart = true;
_logger.Error(
"EnrollmentOrchestrator: prior-run quarantine cleanup FAILED — marker retained for "
+ "next-start retry; forcing a clean Initial seed for this run.", ex);
}
}
// 0) WhiteGlove Part-2 resume detection. V1-symmetric Archive-and-Reset
// pattern: peek the persisted snapshot via the lock-free static reader; if
// the prior run sealed Part-1, move the reducer-state segment files
@@ -405,7 +463,18 @@ namespace AutopilotMonitor.Agent.V2.Core.Orchestration
IReadOnlyList<DecisionSignal> signalsToReplay = Array.Empty<DecisionSignal>();
string branchTag;
if (loadedState == null && snapshotFileExistsPreLoad)
if (_forceCleanStart)
{
// P2 — a requested prior-run quarantine did not fully succeed. Fail-closed: ignore
// any (possibly-still-present) snapshot and SignalLog and seed from a fresh Initial
// state. seed/signalsToReplay are already CreateInitial/empty above. The retained
// marker drives a proper full quarantine + retry on the next start.
_logger.Warning(
"EnrollmentOrchestrator: forcing clean Initial seed (prior-run quarantine incomplete) — "
+ "skipping snapshot/log recovery for this run.");
branchTag = "forced-clean-after-failed-quarantine";
}
else if (loadedState == null && snapshotFileExistsPreLoad)
{
// Branch (a) — Snapshot corrupt. Quarantine the snapshot and attempt log replay.
_logger.Error(
@@ -509,6 +578,12 @@ namespace AutopilotMonitor.Agent.V2.Core.Orchestration
_spool = new TelemetrySpool(_transportDirectory, _clock, _logger);
_transport = new TelemetryUploadOrchestrator(_spool, _uploader, _clock, batchSize: _uploadBatchSize);
// TRACE-H1 — surface skipped oversized items + blocked-pipeline conditions as timeline
// events. Raised after the drain guard releases, so posting back through the ingress
// here cannot deadlock the drain.
_transport.BatchPoisoned += OnBatchPoisoned;
_transport.UploadBlocked += OnUploadBlocked;
// Late-bind the spool reference into the component factory so peripheral
// collectors (AgentSelfMetricsCollector via PeriodicCollectorLifecycleHost) can
// read PendingItemCount / SpoolFileSizeBytes for the agent_metrics_snapshot
@@ -722,6 +797,34 @@ namespace AutopilotMonitor.Agent.V2.Core.Orchestration
// downstream UI splitters can render a "User Enrollment Part 2"
// block. Posted AFTER the onIngressReady hook so agent_started lands
// first (matches V1 ordering).
// 13b-Q) TRACE-H2 — announce that this start honoured a prior run's quarantine request,
// so the backend can distinguish a corruption-recovery from a normal restart.
// Posted after onIngressReady (agent_started lands first), Warning severity.
if (_priorRunQuarantined)
{
try
{
var quarantinePost = new InformationalEventPost(_ingress, _clock);
quarantinePost.Emit(
eventType: Constants.EventTypes.StateQuarantineRecovered,
source: "Agent",
message: "Prior run could not persist consistently (consecutive journal-append failures); "
+ "state segments + snapshot quarantined and session restarted from a clean state.",
severity: Shared.Models.EventSeverity.Warning,
immediateUpload: true,
data: new Dictionary<string, string>(StringComparer.Ordinal)
{
["reason"] = _priorRunQuarantineReason ?? string.Empty,
},
sourceOrigin: "EnrollmentOrchestrator",
evidenceSummary: "Prior-run quarantine marker honoured on startup.");
}
catch (Exception ex)
{
_logger.Error("EnrollmentOrchestrator: failed to post state_quarantine_recovered.", ex);
}
}
if (_isWhiteGlovePart2)
{
try
@@ -1140,7 +1243,112 @@ namespace AutopilotMonitor.Agent.V2.Core.Orchestration
_quarantineRequested = true;
_quarantineReason = reason ?? string.Empty;
_logger.Error($"EnrollmentOrchestrator: quarantine requested — {reason}");
// Actual segment-quarantine happens on next Start (M4.4.5.f), not mid-run.
// TRACE-H2: persist a marker so the NEXT Start performs the actual segment-quarantine —
// a running process must not pull its own state files out from under itself. Best-effort:
// the journal failure that triggered this is often disk-full / ACL on the very same state
// dir, so the marker write may also fail; if it does we're no worse off than the previous
// log-only behaviour (the BootTrigger / max-lifetime restart still re-runs recovery).
try
{
var markerPath = Path.Combine(_stateDirectory, QuarantineRequestedMarkerFile);
var body = _clock.UtcNow.ToString("o", System.Globalization.CultureInfo.InvariantCulture)
+ " " + (reason ?? string.Empty);
File.WriteAllText(markerPath, body, System.Text.Encoding.UTF8);
}
catch (Exception ex)
{
_logger.Error($"EnrollmentOrchestrator: failed to persist quarantine marker: {ex.Message}");
}
}
/// <summary>Reads the reason text from a quarantine marker; returns a fallback on any error.</summary>
private static string TryReadMarkerReason(string markerPath)
{
try { return File.ReadAllText(markerPath, System.Text.Encoding.UTF8); }
catch { return "unknown (marker unreadable)"; }
}
/// <summary>
/// TRACE-H1/P1 — turns a quarantined oversized item into a <c>telemetry_upload_poisoned</c>
/// Warning so the drop is visible on the backend timeline (the item stays in spool.jsonl for
/// the diag ZIP). Only fires for the one locally-provable poison case: a lone item the backend
/// rejects with 413 even on its own. Posted through the ingress like other lifecycle events.
/// </summary>
private void OnBatchPoisoned(object? sender, PoisonReport poison)
{
var ingress = _ingress;
if (ingress == null || poison == null) return;
try
{
// P3 — message reflects WHY the item was dropped so forensics don't read an explicit
// backend schema/RowKey rejection as an "oversized" item.
var message = poison.Kind == PoisonKind.Oversize
? $"Quarantined {poison.ItemCount} oversized telemetry item(s) the backend cannot "
+ "accept at any batch size; item remains on disk for diagnostics."
: $"Dropped {poison.ItemCount} telemetry item(s) the backend explicitly rejected as "
+ "poison (per response signal); items remain on disk for diagnostics.";
new InformationalEventPost(ingress, _clock).Emit(
eventType: Constants.EventTypes.TelemetryUploadPoisoned,
source: "TelemetryUploadOrchestrator",
message: message,
severity: Shared.Models.EventSeverity.Warning,
immediateUpload: true,
data: new Dictionary<string, string>(StringComparer.Ordinal)
{
["kind"] = poison.Kind.ToString(),
["itemCount"] = poison.ItemCount.ToString(System.Globalization.CultureInfo.InvariantCulture),
["throughItemId"] = poison.ThroughItemId.ToString(System.Globalization.CultureInfo.InvariantCulture),
["reason"] = poison.Reason ?? string.Empty,
},
sourceOrigin: "EnrollmentOrchestrator",
evidenceSummary: poison.Kind == PoisonKind.Oversize
? "Oversized telemetry item quarantined to keep the upload pipeline flowing."
: "Backend-rejected (poison) telemetry items dropped to keep the upload pipeline flowing.");
}
catch (Exception ex)
{
_logger.Error("EnrollmentOrchestrator: failed to post telemetry_upload_poisoned.", ex);
}
}
/// <summary>
/// P1 — records a blocked drain (retained unknown permanent 4xx) as a one-shot
/// <c>telemetry_upload_blocked</c> Warning. NO data is discarded. This is primarily a LOCAL
/// diagnostic (agent log + diagnostics ZIP): the event is queued on the same spool behind the
/// very batch that is blocking, so it does NOT reach the backend while the block persists — it
/// uploads (with the retained batch) only once the cause is resolved. There is no out-of-band
/// bypass by design.
/// </summary>
private void OnUploadBlocked(object? sender, string reason)
{
// Local-first: the agent log + diagnostics ZIP are the reliable channel while blocked.
_logger.Warning($"EnrollmentTerminationHandler/Upload: drain blocked by permanent backend rejection — {reason}");
var ingress = _ingress;
if (ingress == null) return;
try
{
new InformationalEventPost(ingress, _clock).Emit(
eventType: Constants.EventTypes.TelemetryUploadBlocked,
source: "TelemetryUploadOrchestrator",
message: "Telemetry upload blocked by a permanent backend rejection; data is retained "
+ "on disk (agent log / diagnostics ZIP). This marker is queued behind the "
+ "blocking batch and reaches the backend only once the block clears.",
severity: Shared.Models.EventSeverity.Warning,
immediateUpload: true,
data: new Dictionary<string, string>(StringComparer.Ordinal)
{
["reason"] = reason ?? string.Empty,
},
sourceOrigin: "EnrollmentOrchestrator",
evidenceSummary: "Upload pipeline blocked on a retained permanent rejection (local diagnostic; no data discarded).");
}
catch (Exception ex)
{
_logger.Error("EnrollmentOrchestrator: failed to post telemetry_upload_blocked.", ex);
}
}
// ---------------------------------------------------------------- Test helpers
@@ -4,6 +4,7 @@ using System.Collections.Generic;
using System.Threading;
using AutopilotMonitor.Agent.V2.Core.Logging;
using AutopilotMonitor.DecisionCore.Engine;
using AutopilotMonitor.DecisionCore.Signals;
using AutopilotMonitor.Shared.Models;
namespace AutopilotMonitor.Agent.V2.Core.Orchestration
@@ -18,6 +19,18 @@ namespace AutopilotMonitor.Agent.V2.Core.Orchestration
private readonly bool _unrestrictedMode;
private int _disposed;
// MON-A1: drive phase_change / on_event gather triggers from the central signal stream.
// V1 fired these from MonitoringService on every EnrollmentEvent (OnPhaseChanged(evt.Phase)
// on phase change + OnEvent(evt.EventType) per event); the V2 host previously only called
// UpdateRules, so 5 of 10 shipped rules (incl. the only enabled one, dsregcmd at
// FinalizingSetup) silently never fired. In V2 every event is an InformationalEvent signal
// carrying eventType + (optional) phase in its payload. Null on test fakes / non-SignalIngress
// sinks — then signal-triggers degrade off (startup + interval rules still run).
private readonly SignalIngress? _observableIngress;
private Action<DecisionSignalKind, IReadOnlyDictionary<string, string>?>? _signalPostedHandler;
private readonly object _sync = new object();
private string? _lastPhaseName;
public GatherRuleExecutorHost(
string sessionId,
string tenantId,
@@ -45,6 +58,7 @@ namespace AutopilotMonitor.Agent.V2.Core.Orchestration
var post = new InformationalEventPost(ingress, clock);
_executor = new Monitoring.Telemetry.Gather.GatherRuleExecutor(
sessionId, tenantId, evt => post.Emit(evt), logger, imeLogPathOverride);
_observableIngress = ingress as SignalIngress;
}
public void Start()
@@ -54,18 +68,73 @@ namespace AutopilotMonitor.Agent.V2.Core.Orchestration
// rule sees the elevated policy when AllowList checks would otherwise reject it.
_executor.UnrestrictedMode = _unrestrictedMode;
_executor.UpdateRules(_rules);
// MON-A1: observe the central signal stream so phase_change / on_event rules fire.
if (_observableIngress != null && _signalPostedHandler == null)
{
_signalPostedHandler = OnSignalPosted;
_observableIngress.SignalPosted += _signalPostedHandler;
}
_logger.Info(
$"GatherRuleExecutorHost: started with {_rules.Count} rule(s), unrestrictedMode={_unrestrictedMode}.");
$"GatherRuleExecutorHost: started with {_rules.Count} rule(s), unrestrictedMode={_unrestrictedMode}, signalTriggers={(_observableIngress != null)}.");
}
/// <summary>
/// Translates posted signals into the executor's phase_change / on_event triggers (MON-A1).
/// Phase: fire <see cref="Monitoring.Telemetry.Gather.GatherRuleExecutor.OnPhaseChanged"/> when
/// the observed phase changes and parses to an <see cref="EnrollmentPhase"/> (raw collector
/// phase strings that aren't enum names are ignored — they carry no gather-rule meaning).
/// Event: fire <see cref="Monitoring.Telemetry.Gather.GatherRuleExecutor.OnEvent"/> for every
/// InformationalEvent's eventType. The executor dispatches rule execution on the ThreadPool,
/// so this stays off the signal-posting hot path; it also dedups phase rules per (rule, phase).
/// </summary>
private void OnSignalPosted(DecisionSignalKind kind, IReadOnlyDictionary<string, string>? payload)
{
if (payload == null) return;
try
{
lock (_sync)
{
if (payload.TryGetValue(SignalPayloadKeys.EspPhase, out var phaseName)
&& !string.IsNullOrEmpty(phaseName)
&& !string.Equals(phaseName, _lastPhaseName, StringComparison.OrdinalIgnoreCase)
&& Enum.TryParse<EnrollmentPhase>(phaseName, ignoreCase: true, out var phase))
{
_lastPhaseName = phaseName;
_executor.OnPhaseChanged(phase);
}
if (kind == DecisionSignalKind.InformationalEvent
&& payload.TryGetValue(SignalPayloadKeys.EventType, out var eventType)
&& !string.IsNullOrEmpty(eventType))
{
_executor.OnEvent(eventType);
}
}
}
catch (Exception ex)
{
_logger.Verbose($"GatherRuleExecutorHost: signal-trigger dispatch failed: {ex.Message}");
}
}
public void Stop()
{
// GatherRuleExecutor is IDisposable; no explicit Stop. Rely on Dispose.
if (_observableIngress != null && _signalPostedHandler != null)
{
try { _observableIngress.SignalPosted -= _signalPostedHandler; }
catch { /* best-effort unsubscribe during shutdown */ }
_signalPostedHandler = null;
}
// GatherRuleExecutor is IDisposable; no explicit Stop beyond unsubscribe. Rely on Dispose.
}
public void Dispose()
{
if (Interlocked.Exchange(ref _disposed, 1) == 1) return;
Stop();
try { _executor.Dispose(); } catch { }
}
}
@@ -25,13 +25,15 @@ namespace AutopilotMonitor.Agent.V2.Core.Orchestration
/// <see cref="SignalIngress"/>. Activity observation lives centrally on
/// <see cref="SignalIngress.SignalPosted"/> (Codex Finding 4) so this host can see signals
/// from <b>every</b> source — Program.cs lifecycle, IME, ESP, DeviceInfo, Analyzer,
/// Gather, etc. — not just the subset that flows through its own post. The handler peeks
/// every posted signal's <c>eventType</c> payload and bumps <c>_lastRealEventTimeUtc</c>
/// unless the event belongs to a well-known periodic type
/// (<c>performance_snapshot</c>, <c>agent_metrics_snapshot</c>,
/// Gather, etc. — not just the subset that flows through its own post. The handler delegates
/// the "is this real activity?" decision to <see cref="SignalActivityClassifier"/> (shared with
/// <see cref="StallProbeHost"/>), which excludes internal scheduler ticks
/// (<see cref="DecisionSignalKind.DeadlineFired"/>, notably the 30 s <c>classifier_tick</c>) and
/// the periodic event-type denylist (<c>performance_snapshot</c>, <c>agent_metrics_snapshot</c>,
/// <c>performance_collector_stopped</c>, <c>agent_metrics_collector_stopped</c>,
/// <c>stall_probe_*</c>, <c>session_stalled</c>). Events emitted by the managed collectors
/// themselves therefore never reset their own idle window.
/// themselves — and the classifier_tick that previously kept them alive forever — therefore
/// never reset the idle window.
/// </para>
/// </summary>
internal sealed class PeriodicCollectorLifecycleHost : ICollectorHost
@@ -39,16 +41,6 @@ namespace AutopilotMonitor.Agent.V2.Core.Orchestration
public string Name => "PeriodicCollectorLifecycleHost";
private static readonly TimeSpan IdleTickInterval = TimeSpan.FromSeconds(60);
private static readonly HashSet<string> PeriodicEventTypes = new HashSet<string>(StringComparer.OrdinalIgnoreCase)
{
"performance_snapshot",
"agent_metrics_snapshot",
SharedConstants.EventTypes.PerformanceCollectorStopped,
SharedConstants.EventTypes.AgentMetricsCollectorStopped,
SharedConstants.EventTypes.StallProbeCheck,
SharedConstants.EventTypes.StallProbeResult,
SharedConstants.EventTypes.SessionStalled,
};
private readonly string _sessionId;
private readonly string _tenantId;
@@ -125,22 +117,13 @@ namespace AutopilotMonitor.Agent.V2.Core.Orchestration
/// </summary>
private void OnSignalPosted(DecisionSignalKind kind, IReadOnlyDictionary<string, string>? payload)
{
// InformationalEvents carry eventType in payload — filter those against the
// periodic-denylist so snapshots never mark the session as "active".
if (kind == DecisionSignalKind.InformationalEvent)
{
if (payload != null
&& payload.TryGetValue(SignalPayloadKeys.EventType, out var eventType)
&& !string.IsNullOrEmpty(eventType)
&& PeriodicEventTypes.Contains(eventType))
{
return;
}
}
// Shared classifier filters out periodic self-emissions and internal scheduler ticks
// (DeadlineFired / classifier_tick). Everything else — ESP phase, IME session, hello,
// desktop arrival, classifier verdict, session-lifecycle, … — is real activity that
// should keep the collectors running.
if (!SignalActivityClassifier.IsRealActivity(kind, payload))
return;
// Every other signal kind — ESP phase, IME session, hello, desktop arrival,
// deadline fired, classifier verdict, session-lifecycle, … — is real activity
// that should keep the collectors running.
OnRealEvent();
}
@@ -1,13 +1,30 @@
#nullable enable
using System;
using System.Collections.Generic;
using System.Threading;
using AutopilotMonitor.Agent.V2.Core.Logging;
using AutopilotMonitor.Agent.V2.Core.Monitoring.Enrollment.SystemSignals;
using AutopilotMonitor.Agent.V2.Core.SignalAdapters;
using AutopilotMonitor.DecisionCore.Engine;
using AutopilotMonitor.DecisionCore.Signals;
namespace AutopilotMonitor.Agent.V2.Core.Orchestration
{
/// <summary>
/// Drives the <see cref="StallProbeCollector"/> from a 60-s tick. The collector fires its deep
/// probes at idle thresholds (2/15/30/60/180 min), so the host must feed it a true
/// <i>idle</i> duration — minutes since the last real enrollment activity — not agent uptime.
/// <para>
/// Activity is observed centrally on <see cref="SignalIngress.SignalPosted"/> and classified by
/// <see cref="SignalActivityClassifier"/> (shared with <see cref="PeriodicCollectorLifecycleHost"/>),
/// so internal scheduler ticks (the 30 s <c>classifier_tick</c>) and the collector's own
/// <c>stall_probe_*</c> emissions never reset the clock. On real activity the per-probe fire-once
/// state is reset (<see cref="StallProbeCollector.ResetProbes"/>) so a stall that begins <i>after</i>
/// a burst of activity can still trigger probes. Without this the host fed agent uptime as
/// "idle minutes" — probes fired on a fixed schedule on healthy sessions and could never re-fire
/// for a genuine late stall (review MON-A2).
/// </para>
/// </summary>
internal sealed class StallProbeHost : ICollectorHost
{
public string Name => "StallProbeCollector";
@@ -18,9 +35,18 @@ namespace AutopilotMonitor.Agent.V2.Core.Orchestration
private readonly StallProbeCollectorAdapter _adapter;
private readonly AgentLogger _logger;
private Timer? _tickTimer;
private readonly DateTime _startedAtUtc;
private int _disposed;
// Activity clock. Concrete SignalIngress (when available) lets us observe activity from ALL
// sources via SignalPosted. Null on test fakes / non-SignalIngress sinks — then the clock
// is never advanced and idle measures time since Start (the original uptime-based behaviour,
// preserved so existing collector/adapter tests are unaffected).
private readonly SignalIngress? _observableIngress;
private Action<DecisionSignalKind, IReadOnlyDictionary<string, string>?>? _signalPostedHandler;
private readonly object _activityLock = new object();
private DateTime _lastRealEventTimeUtc;
public StallProbeHost(
string sessionId,
string tenantId,
@@ -49,11 +75,20 @@ namespace AutopilotMonitor.Agent.V2.Core.Orchestration
harmlessModernDeploymentEventIds: harmlessModernDeploymentEventIds);
_adapter = new StallProbeCollectorAdapter(_collector, ingress, clock);
_startedAtUtc = DateTime.UtcNow;
_observableIngress = ingress as SignalIngress;
_lastRealEventTimeUtc = DateTime.UtcNow;
}
public void Start()
{
// Observe cross-source activity before the tick timer starts so the idle clock cannot be
// reset by an early signal we'd otherwise miss. No-op on test fakes (_observableIngress null).
if (_observableIngress != null && _signalPostedHandler == null)
{
_signalPostedHandler = OnSignalPosted;
_observableIngress.SignalPosted += _signalPostedHandler;
}
// The collector has no timer of its own — we drive it with a 60-s idle tick.
_tickTimer = new Timer(
_ => SafeTick(),
@@ -67,6 +102,13 @@ namespace AutopilotMonitor.Agent.V2.Core.Orchestration
{
try
{
if (_observableIngress != null && _signalPostedHandler != null)
{
try { _observableIngress.SignalPosted -= _signalPostedHandler; }
catch { /* best-effort unsubscribe during shutdown */ }
_signalPostedHandler = null;
}
_tickTimer?.Dispose();
_tickTimer = null;
}
@@ -80,11 +122,31 @@ namespace AutopilotMonitor.Agent.V2.Core.Orchestration
try { _adapter.Dispose(); } catch { }
}
/// <summary>
/// Subscriber for <see cref="SignalIngress.SignalPosted"/>. On real enrollment activity the
/// idle clock is reset and the per-probe fire-once state cleared so probes can fire again on
/// the next stall. Periodic self-emissions and internal scheduler ticks are filtered out by
/// <see cref="SignalActivityClassifier"/>.
/// </summary>
private void OnSignalPosted(DecisionSignalKind kind, IReadOnlyDictionary<string, string>? payload)
{
if (!SignalActivityClassifier.IsRealActivity(kind, payload))
return;
lock (_activityLock)
{
_lastRealEventTimeUtc = DateTime.UtcNow;
}
_collector.ResetProbes();
}
private void SafeTick()
{
try
{
var idleMinutes = (DateTime.UtcNow - _startedAtUtc).TotalMinutes;
DateTime lastReal;
lock (_activityLock) { lastReal = _lastRealEventTimeUtc; }
var idleMinutes = (DateTime.UtcNow - lastReal).TotalMinutes;
_collector.CheckAndRunProbes(idleMinutes);
}
catch (Exception ex)
@@ -0,0 +1,97 @@
#nullable enable
using System;
using System.Collections.Generic;
using AutopilotMonitor.DecisionCore.Engine;
using AutopilotMonitor.DecisionCore.Signals;
using SharedConstants = AutopilotMonitor.Shared.Constants;
namespace AutopilotMonitor.Agent.V2.Core.Orchestration
{
/// <summary>
/// Single source of truth for "is this posted signal real enrollment activity?" — i.e. the
/// kind of device/enrollment progress that an activity-gated collector should treat as
/// keeping the session alive.
/// <para>
/// Shared by <see cref="PeriodicCollectorLifecycleHost"/> (idle-stop of the Performance /
/// AgentSelfMetrics collectors) and <see cref="StallProbeHost"/> (idle clock + probe reset)
/// so the two never drift apart. Previously each host carried its own denylist; the
/// <see cref="PeriodicCollectorLifecycleHost"/> copy also failed to exclude internal scheduler
/// ticks, which let the 30 s <c>classifier_tick</c> reset the idle clock on every session and
/// silently defeat idle-stop (review PERF-H3), and <see cref="StallProbeHost"/> measured agent
/// uptime instead of idleness (review MON-A2).
/// </para>
/// </summary>
internal static class SignalActivityClassifier
{
/// <summary>
/// Event types that are NOT real device/enrollment progress and must therefore NOT reset
/// the idle clocks. Two families:
/// <list type="bullet">
/// <item><b>Periodic self-emissions</b> by the activity-gated collectors themselves —
/// otherwise a collector's own snapshot would keep its idle window alive forever.</item>
/// <item><b>Agent health / control / transport</b> events — these report the agent's own
/// state (degraded collector, upload poisoned/blocked, quarantine recovery, prior
/// crash, spool/ingress back-pressure, shutdown). They are not enrollment progress;
/// counting them would push out stall detection and keep the periodic collectors
/// running on an otherwise-idle device (P2 review).</item>
/// </list>
/// </summary>
private static readonly HashSet<string> NonActivityEventTypes = new HashSet<string>(StringComparer.OrdinalIgnoreCase)
{
// Periodic self-emissions.
"performance_snapshot",
"agent_metrics_snapshot",
SharedConstants.EventTypes.PerformanceCollectorStopped,
SharedConstants.EventTypes.AgentMetricsCollectorStopped,
SharedConstants.EventTypes.StallProbeCheck,
SharedConstants.EventTypes.StallProbeResult,
SharedConstants.EventTypes.SessionStalled,
// Agent health / control / transport — not device/enrollment progress.
SharedConstants.EventTypes.CollectorDegraded,
SharedConstants.EventTypes.TelemetryUploadPoisoned,
SharedConstants.EventTypes.TelemetryUploadBlocked,
SharedConstants.EventTypes.StateQuarantineRecovered,
SharedConstants.EventTypes.PriorRunDiedWithState,
SharedConstants.EventTypes.PreviousCrashDetected,
SharedConstants.EventTypes.SpoolPressureDetected,
SharedConstants.EventTypes.IngressBackpressure,
SharedConstants.EventTypes.AgentShuttingDown,
};
/// <summary>
/// Returns <c>true</c> when the posted signal represents real enrollment activity.
/// </summary>
/// <remarks>
/// Excludes two classes of "non-activity":
/// <list type="number">
/// <item><see cref="DecisionSignalKind.DeadlineFired"/> — internal scheduler ticks
/// (notably the 30 s <c>classifier_tick</c>). These are the agent's own timers, not
/// device activity; any real <i>consequence</i> of a deadline (a state transition)
/// flows back through the ingress as its own signal and DOES count, so excluding the
/// raw tick is safe and necessary.</item>
/// <item><see cref="DecisionSignalKind.InformationalEvent"/> whose <c>eventType</c> is in
/// the periodic denylist — the snapshots emitted by the activity-gated collectors
/// themselves.</item>
/// </list>
/// Everything else — ESP phase, IME session, Hello, desktop arrival, classifier verdict,
/// session lifecycle, and all non-periodic informational events — is real activity.
/// </remarks>
public static bool IsRealActivity(DecisionSignalKind kind, IReadOnlyDictionary<string, string>? payload)
{
if (kind == DecisionSignalKind.DeadlineFired)
return false;
if (kind == DecisionSignalKind.InformationalEvent
&& payload != null
&& payload.TryGetValue(SignalPayloadKeys.EventType, out var eventType)
&& !string.IsNullOrEmpty(eventType)
&& NonActivityEventTypes.Contains(eventType))
{
return false;
}
return true;
}
}
}
@@ -772,6 +772,16 @@ namespace AutopilotMonitor.Agent.V2.Core.Termination
try
{
// LIFE-F3: drain the spool BEFORE launching the self-deleting cleanup script. The
// script renames+deletes the agent tree (and may force-delete unlocked files) on its
// own timer, which can otherwise race the final telemetry drain that
// TerminationPipeline.Run/orchestrator.Stop performs later — losing the terminal
// enrollment_complete / app_tracking_summary / diagnostics_* events that were emitted
// 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();
var service = _cleanupServiceFactory();
service.ExecuteSelfDestruct();
_logger.Info("EnrollmentTerminationHandler: CleanupService.ExecuteSelfDestruct() invoked (fire-and-forget).");
@@ -194,15 +194,29 @@ namespace AutopilotMonitor.Agent.V2.Core.Transport.Telemetry
// Feed the central tracker so the agent can shut down after MaxAuthFailures
// instead of retrying a permanently-revoked certificate forever.
_authFailureTracker?.RecordFailure(statusCode, "agent/telemetry");
return UploadResult.Permanent($"unauthorized: {shortReason}");
return UploadResult.Unauthorized($"unauthorized: {shortReason}");
case HttpStatusCode.RequestTimeout: // 408
case (HttpStatusCode)429: // TooManyRequests — enum available net48 SP1+, cast for safety
return UploadResult.Transient(shortReason);
case HttpStatusCode.RequestEntityTooLarge: // 413 — the data is fine, only the batch
return UploadResult.TooLarge(shortReason); // size is wrong → split & retry (P1).
}
if (statusCode >= 500 && statusCode <= 599) return UploadResult.Transient(shortReason);
if (statusCode >= 400 && statusCode <= 499) return UploadResult.Permanent(shortReason);
if (statusCode >= 400 && statusCode <= 499)
{
// P1: an explicit item-level poison signal in the 4xx body is the ONLY thing that
// authorises a discard ({ "poison": true, "rejectedRowKeys": [...] }). Absent that,
// an unknown permanent 4xx (400 contract bug / 404 route mismatch / tenant-validation)
// means only THIS request was rejected — we RETAIN the batch (cursor stays) and let
// the drain block + report rather than discard good data on a guessed status code.
var poison = await TryReadPoisonSignalAsync(response).ConfigureAwait(false);
if (poison != null) return poison;
return UploadResult.Permanent(shortReason);
}
// Unexpected range (1xx, 3xx). Treat as transient — orchestrator retries and logs.
return UploadResult.Transient(shortReason);
@@ -269,6 +283,52 @@ namespace AutopilotMonitor.Agent.V2.Core.Transport.Telemetry
: UploadResult.Ok();
}
/// <summary>
/// P1 — parse an EXPLICIT item-level poison signal from a 4xx response body:
/// <c>{ "poison": true, "rejectedRowKeys": ["rk1", "rk2"], "reason": "..." }</c>.
/// Returns <c>null</c> (→ caller falls back to retain) unless <c>poison == true</c> AND a
/// non-empty <c>rejectedRowKeys</c> array is present — we never discard a whole batch on an
/// underspecified signal, and we never infer poison from the status code alone.
/// </summary>
private static async Task<UploadResult?> TryReadPoisonSignalAsync(HttpResponseMessage response)
{
string body;
try { body = await response.Content.ReadAsStringAsync().ConfigureAwait(false); }
catch { return null; }
if (string.IsNullOrWhiteSpace(body)) return null;
JObject? root;
try { root = JObject.Parse(body); }
catch { return null; }
if (root == null) return null;
if ((TryBool(root, "poison", "Poison") ?? false) == false) return null;
var rowKeys = TryStringArray(root, "rejectedRowKeys", "RejectedRowKeys");
if (rowKeys == null || rowKeys.Count == 0) return null; // item-level only — no rowkeys → retain
var reason = TryString(root, "reason", "Reason");
return UploadResult.Poison(rowKeys, string.IsNullOrEmpty(reason) ? "backend_poison" : reason!);
}
private static IReadOnlyList<string>? TryStringArray(JObject root, params string[] names)
{
foreach (var n in names)
{
if (root.TryGetValue(n, StringComparison.OrdinalIgnoreCase, out var tok) && tok is JArray arr)
{
var list = new List<string>(arr.Count);
foreach (var e in arr)
{
var s = e?.ToString();
if (!string.IsNullOrEmpty(s)) list.Add(s!);
}
return list;
}
}
return null;
}
private static bool? TryBool(JObject root, params string[] names)
{
foreach (var n in names)
@@ -0,0 +1,40 @@
#nullable enable
namespace AutopilotMonitor.Agent.V2.Core.Transport.Telemetry
{
/// <summary>
/// TRACE-H1 — describes a poison batch the drain skipped (a non-transient, non-auth upload
/// failure). Carried by <see cref="TelemetryUploadOrchestrator.BatchPoisoned"/> so the
/// orchestrator can surface a <c>telemetry_upload_poisoned</c> timeline event.
/// </summary>
/// <summary>How the dropped item(s) were determined to be un-sendable.</summary>
public enum PoisonKind
{
/// <summary>A lone item the backend rejected with 413 even on its own (locally-provable).</summary>
Oversize,
/// <summary>The backend explicitly named these RowKeys as poison in a 4xx body (P1).</summary>
BackendRejected,
}
public sealed class PoisonReport
{
public PoisonReport(int itemCount, long throughItemId, string? reason, PoisonKind kind)
{
ItemCount = itemCount;
ThroughItemId = throughItemId;
Reason = reason;
Kind = kind;
}
/// <summary>How the drop was decided (P3 — keeps forensics from mislabeling explicit poison as "oversized").</summary>
public PoisonKind Kind { get; }
/// <summary>Number of spool items in the skipped batch.</summary>
public int ItemCount { get; }
/// <summary>The <c>TelemetryItemId</c> the cursor was advanced to (inclusive) to skip the batch.</summary>
public long ThroughItemId { get; }
/// <summary>The backend rejection reason (e.g. "http 413: ..."), if known.</summary>
public string? Reason { get; }
}
}
@@ -19,7 +19,7 @@ namespace AutopilotMonitor.Agent.V2.Core.Transport.Telemetry
public static string Serialize(TelemetryItem item)
{
if (item == null) throw new ArgumentNullException(nameof(item));
return JsonConvert.SerializeObject(item, DecisionCoreJsonSettings.Create());
return JsonConvert.SerializeObject(item, DecisionCoreJsonSettings.Shared);
}
public static TelemetryItem Deserialize(string line)
@@ -28,7 +28,7 @@ namespace AutopilotMonitor.Agent.V2.Core.Transport.Telemetry
try
{
var result = JsonConvert.DeserializeObject<TelemetryItem>(line, DecisionCoreJsonSettings.Create());
var result = JsonConvert.DeserializeObject<TelemetryItem>(line, DecisionCoreJsonSettings.Shared);
if (result == null)
{
throw new JsonSerializationException($"TelemetryItem deserialization produced null: {line}");
@@ -36,6 +36,15 @@ namespace AutopilotMonitor.Agent.V2.Core.Transport.Telemetry
// M4.6.ε — backend-to-agent control signals.
private DateTime? _pausedUntilUtc;
// P1 — current effective batch size. Shrinks (never grows) when the backend returns 413,
// so an oversized batch is split down instead of discarded. Only mutated under the drain
// guard. Floor 1.
private int _effectiveBatchSize;
// P1 — one-shot guard so a retained (non-transient, non-auth) permanent block reports
// telemetry_upload_blocked exactly once per process rather than on every 30 s drain.
private bool _uploadBlockedReported;
public TelemetryUploadOrchestrator(
ITelemetrySpool spool,
IBackendTelemetryUploader uploader,
@@ -49,6 +58,7 @@ namespace AutopilotMonitor.Agent.V2.Core.Transport.Telemetry
if (batchSize <= 0) throw new ArgumentOutOfRangeException(nameof(batchSize), "BatchSize must be positive.");
_batchSize = batchSize;
_effectiveBatchSize = batchSize;
_retryBackoffs = retryBackoffs ?? DefaultRetryBackoffs;
}
@@ -69,6 +79,24 @@ namespace AutopilotMonitor.Agent.V2.Core.Transport.Telemetry
/// </summary>
public event EventHandler<UploadResult>? ServerResponseReceived;
/// <summary>
/// TRACE-H1 — raised (after the drain guard releases) when a single oversized item was
/// quarantined to keep the pipeline flowing (the only locally-provable poison: a lone item
/// the backend rejects with 413 even on its own). The orchestrator turns this into a
/// <c>telemetry_upload_poisoned</c> timeline event so the drop is visible on the backend.
/// </summary>
public event EventHandler<PoisonReport>? BatchPoisoned;
/// <summary>
/// P1 — raised once per process (after the drain guard releases) when the drain is blocked by
/// a retained non-transient, non-auth failure (unknown permanent 4xx: 400 contract bug, 404
/// route mismatch, tenant-validation …). The batch is NOT discarded. This is a LOCAL diagnostic
/// (agent log + diagnostics ZIP): any timeline event the handler posts is queued behind the
/// blocking batch on the same spool, so it does not reach the backend until the block clears —
/// there is no out-of-band bypass by design. Carries the backend error reason.
/// </summary>
public event EventHandler<string>? UploadBlocked;
/// <summary>
/// <c>true</c> when the last successful upload returned <see cref="UploadResult.DeviceBlocked"/>
/// and the paused-until window has not expired. Drain is a no-op in this state.
@@ -94,6 +122,8 @@ namespace AutopilotMonitor.Agent.V2.Core.Transport.Telemetry
// onTerminateRequested → orchestrator.Stop → DrainAllAsync, which blocks on the
// guard and deadlocks the final drain (lost agent_shutting_down). Codex Finding 1.
List<UploadResult>? pendingSignals = null;
List<PoisonReport>? pendingPoison = null;
string? pendingBlockedReason = null;
DrainResult result;
try
@@ -102,6 +132,17 @@ namespace AutopilotMonitor.Agent.V2.Core.Transport.Telemetry
int failedBatches = 0;
string? lastError = null;
// P1 — shared success-side control-signal handling. Applied to BOTH the normal
// success path and the poison-survivor re-upload, so a DeviceBlocked / DeviceKill /
// AdminAction / Actions response on the survivor upload is honoured identically and
// never silently dropped. Returns true if the backend just quarantined the device.
bool CollectControlSignals(UploadResult o)
{
ApplyControlSignals(o); // mutates _pausedUntilUtc synchronously (next IsPaused sees it)
if (CarriesSignal(o)) (pendingSignals ??= new List<UploadResult>()).Add(o);
return o.DeviceBlocked;
}
while (!cancellationToken.IsCancellationRequested)
{
// M4.6.ε — honour a backend-issued DeviceBlocked quarantine. The block is
@@ -112,7 +153,7 @@ namespace AutopilotMonitor.Agent.V2.Core.Transport.Telemetry
break;
}
var batch = _spool.Peek(_batchSize);
var batch = _spool.Peek(_effectiveBatchSize);
if (batch.Count == 0) break;
var outcome = await TryUploadWithRetryAsync(batch, cancellationToken).ConfigureAwait(false);
@@ -122,25 +163,110 @@ namespace AutopilotMonitor.Agent.V2.Core.Transport.Telemetry
_spool.MarkUploaded(batch[batch.Count - 1].TelemetryItemId);
uploaded += batch.Count;
// ApplyControlSignals mutates _pausedUntilUtc synchronously so the next
// iteration's IsPaused check sees the block. Event dispatch is deferred.
ApplyControlSignals(outcome);
if (CarriesSignal(outcome))
{
(pendingSignals ??= new List<UploadResult>()).Add(outcome);
}
// Stop draining immediately if the backend just quarantined the device —
// the next loop iteration would see IsPaused and break anyway, but there
// is no point in even Peeking another batch.
if (outcome.DeviceBlocked) break;
if (CollectControlSignals(outcome)) break;
}
else if (outcome.IsPoison)
{
// P1 — explicit item-level poison. The backend NAMED specific RowKeys as
// permanently un-ingestable. Drop ONLY those; re-upload the rest of the batch;
// then advance the cursor past the whole batch. We never discard an item the
// backend did not explicitly name.
var poisonSet = new HashSet<string>(
outcome.PoisonRowKeys ?? (IReadOnlyList<string>)Array.Empty<string>(), StringComparer.Ordinal);
var survivors = new List<TelemetryItem>(batch.Count);
foreach (var item in batch)
{
if (!poisonSet.Contains(item.RowKey)) survivors.Add(item);
}
var dropped = batch.Count - survivors.Count;
if (dropped == 0)
{
// None of the named RowKeys are in this batch — a mismatched signal.
// RETAIN (do not advance) so a stale/misaddressed poison list can't cause
// data loss; retry next drain.
failedBatches++;
lastError = outcome.ErrorReason;
break;
}
bool deviceBlockedBySurvivors = false;
if (survivors.Count > 0)
{
var survivorOutcome = await TryUploadWithRetryAsync(survivors, cancellationToken).ConfigureAwait(false);
if (!survivorOutcome.Success)
{
// Could not upload the good items — RETAIN the whole batch (cursor
// stays) and retry next drain rather than risk losing survivors.
// (Nested poison re-filters on the next cycle.)
failedBatches++;
lastError = survivorOutcome.ErrorReason;
break;
}
uploaded += survivors.Count;
// P1 fix: honour any control signals the survivor re-upload carried
// (DeviceBlocked / DeviceKill / AdminAction / Actions) — same as the
// normal success path; previously these were silently dropped.
deviceBlockedBySurvivors = CollectControlSignals(survivorOutcome);
}
// Survivors uploaded (or none) — advance past the whole batch; the named items
// are intentionally dropped per the explicit backend signal. Surface the drop.
_spool.MarkUploaded(batch[batch.Count - 1].TelemetryItemId);
(pendingPoison ??= new List<PoisonReport>()).Add(
new PoisonReport(dropped, batch[batch.Count - 1].TelemetryItemId, outcome.ErrorReason,
PoisonKind.BackendRejected));
// If the survivor upload just quarantined the device, stop draining (mirror the
// normal success path) instead of continuing past the block.
if (deviceBlockedBySurvivors) break;
continue;
}
else if (outcome.RequiresSplit)
{
// P1: 413 payload-too-large. The data is fine — only the batch size is wrong,
// so we SPLIT instead of discarding. Halve the effective batch size and retry
// (shrink persists for the session so we don't re-hit the limit every cycle).
if (batch.Count > 1)
{
_effectiveBatchSize = Math.Max(1, batch.Count / 2);
continue; // re-Peek smaller, same drain cycle
}
// A LONE item that is still too large is the one case the agent can locally
// prove is permanently un-sendable (no size fits) — quarantine just that item,
// advance the cursor by exactly one, and surface it. Everything else flows.
failedBatches++;
lastError = outcome.ErrorReason;
var oversizeId = batch[0].TelemetryItemId;
_spool.MarkUploaded(oversizeId);
(pendingPoison ??= new List<PoisonReport>()).Add(
new PoisonReport(1, oversizeId, "oversize: " + outcome.ErrorReason, PoisonKind.Oversize));
break;
}
else
{
// Stop drain — cursor stays put, re-Peek next drain will return the same
// batch (plus anything newly enqueued), backend will dedup on PK/RK.
// P1 — RETAIN. Transient (retry next drain), auth-permanent (uploader already
// drove AuthFailureTracker → shutdown), OR an unknown permanent 4xx (400
// contract bug / 404 route mismatch / tenant-validation …). We do NOT advance
// the cursor on ANY of these: blocking is recoverable (backend/route fix, diag
// ZIP), discarding good telemetry on a guessed "poison" is not. A genuine
// per-item poison is skipped only on an EXPLICIT backend poison signal (not yet
// wired). Re-Peek next drain returns the same batch; backend dedups on PK/RK.
failedBatches++;
lastError = outcome.ErrorReason;
// One-shot loud marker so a stuck non-transient, non-auth block is visible on
// the backend instead of only in the agent log (transient = normal retry;
// auth = the session is already shutting down via the auth path).
if (!outcome.IsTransient && !outcome.IsAuthFailure && !_uploadBlockedReported)
{
_uploadBlockedReported = true;
pendingBlockedReason = outcome.ErrorReason;
}
break;
}
}
@@ -160,6 +286,21 @@ namespace AutopilotMonitor.Agent.V2.Core.Transport.Telemetry
}
}
if (pendingPoison != null)
{
foreach (var poison in pendingPoison)
{
try { BatchPoisoned?.Invoke(this, poison); }
catch { /* handler must not abort the drain result; cursor already advanced */ }
}
}
if (pendingBlockedReason != null)
{
try { UploadBlocked?.Invoke(this, pendingBlockedReason); }
catch { /* handler must not abort the drain result */ }
}
return result;
}
@@ -26,7 +26,11 @@ namespace AutopilotMonitor.Agent.V2.Core.Transport.Telemetry
DateTime? unblockAt,
bool deviceKillSignal,
string? adminAction,
IReadOnlyList<ServerAction>? actions)
IReadOnlyList<ServerAction>? actions,
bool isAuthFailure = false,
bool requiresSplit = false,
bool isPoison = false,
IReadOnlyList<string>? poisonRowKeys = null)
{
Success = success;
ErrorReason = errorReason;
@@ -36,6 +40,10 @@ namespace AutopilotMonitor.Agent.V2.Core.Transport.Telemetry
DeviceKillSignal = deviceKillSignal;
AdminAction = adminAction;
Actions = actions;
IsAuthFailure = isAuthFailure;
RequiresSplit = requiresSplit;
IsPoison = isPoison;
PoisonRowKeys = poisonRowKeys;
}
public bool Success { get; }
@@ -46,6 +54,33 @@ namespace AutopilotMonitor.Agent.V2.Core.Transport.Telemetry
/// <summary>Bei Success ignoriert. Bei Fehler: true → Retry sinnvoll, false → dauerhafter Fehler (kein Retry).</summary>
public bool IsTransient { get; }
/// <summary>
/// TRACE-H1: a 401/403 auth-permanent failure. The uploader already drove
/// <c>AuthFailureTracker</c> → shutdown; the orchestrator retains the batch (cursor stays)
/// and lets the auth path end the session.
/// </summary>
public bool IsAuthFailure { get; }
/// <summary>
/// TRACE-H1 (P1): a 413 "payload too large". NOT poison — the data is fine, only the batch
/// size is wrong. The orchestrator splits/shrinks the batch and retries (never discards a
/// multi-item batch); only a lone item that is still too large is quarantined, because that
/// is the single case the agent can locally prove is permanently un-sendable.
/// </summary>
public bool RequiresSplit { get; }
/// <summary>
/// P1: the backend EXPLICITLY declared specific items permanently un-ingestable via a 4xx
/// response body (<c>{ "poison": true, "rejectedRowKeys": [...] }</c>). This is the ONLY
/// signal that authorises a discard — a bare status code never infers poison. The orchestrator
/// drops exactly the named <see cref="PoisonRowKeys"/>, re-uploads the rest of the batch, then
/// advances the cursor.
/// </summary>
public bool IsPoison { get; }
/// <summary>The RowKeys the backend named as poison (item-level). Non-empty when <see cref="IsPoison"/>.</summary>
public IReadOnlyList<string>? PoisonRowKeys { get; }
/// <summary>
/// Non-terminal quarantine signalled by the backend. The agent should stop draining
/// uploads until <see cref="UnblockAt"/> (if set) elapses — the session stays alive so
@@ -100,5 +135,35 @@ namespace AutopilotMonitor.Agent.V2.Core.Transport.Telemetry
public static UploadResult Permanent(string reason) =>
new UploadResult(false, reason ?? throw new ArgumentNullException(nameof(reason)), isTransient: false,
deviceBlocked: false, unblockAt: null, deviceKillSignal: false, adminAction: null, actions: null);
/// <summary>
/// 401/403 auth-permanent failure. Like <see cref="Permanent"/> (no retry, batch retained)
/// but flagged <see cref="IsAuthFailure"/> so the drain knows the auth-failure path
/// (AuthFailureTracker → shutdown) owns the outcome.
/// </summary>
public static UploadResult Unauthorized(string reason) =>
new UploadResult(false, reason ?? throw new ArgumentNullException(nameof(reason)), isTransient: false,
deviceBlocked: false, unblockAt: null, deviceKillSignal: false, adminAction: null, actions: null,
isAuthFailure: true);
/// <summary>
/// 413 "payload too large". Non-retryable as-is, but the data is fine — flagged
/// <see cref="RequiresSplit"/> so the drain shrinks the batch and retries instead of
/// discarding (P1: never lose good telemetry to a size error).
/// </summary>
public static UploadResult TooLarge(string reason) =>
new UploadResult(false, reason ?? throw new ArgumentNullException(nameof(reason)), isTransient: false,
deviceBlocked: false, unblockAt: null, deviceKillSignal: false, adminAction: null, actions: null,
requiresSplit: true);
/// <summary>
/// Explicit item-level backend poison signal (P1). <paramref name="rejectedRowKeys"/> are the
/// items the backend declared permanently un-ingestable; the orchestrator drops exactly those
/// and re-uploads the rest of the batch.
/// </summary>
public static UploadResult Poison(IReadOnlyList<string> rejectedRowKeys, string reason) =>
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)));
}
}
@@ -1 +1 @@
1015
1036
@@ -1 +1 @@
1433
1439
@@ -157,6 +157,8 @@ namespace AutopilotMonitor.DecisionCore.Tests
// Backward-compat: snapshots from before this field existed deserialize with
// AgentBootUtc=null. The helper must fall back to signal.OccurredAtUtc so legacy
// sessions reproduce their pre-fix deadline values exactly (no replay correction).
// Vehicle: the reactively-armed ClassifierTick (Wave 2 / H2 — SessionStarted no
// longer arms it; a primary WhiteGlove signal does, via the same EffectiveDeadlineBase).
var engine = new DecisionEngine();
var legacy = new DecisionState(
sessionId: "sess-legacy",
@@ -180,9 +182,10 @@ namespace AutopilotMonitor.DecisionCore.Tests
agentBootUtc: null);
var step = engine.Reduce(legacy,
MakeSignal(0, DecisionSignalKind.SessionStarted, LogPast));
MakeSignal(0, DecisionSignalKind.WhiteGloveSealingPatternDetected, LogPast));
var tick = Assert.Single(step.NewState.Deadlines);
Assert.Equal(DeadlineNames.ClassifierTick, tick.Name);
Assert.Null(step.NewState.AgentBootUtc);
Assert.Equal(LogPast.AddSeconds(30), tick.DueAtUtc);
}
@@ -1,4 +1,5 @@
using System;
using System.Linq;
using AutopilotMonitor.DecisionCore.Engine;
using AutopilotMonitor.DecisionCore.Signals;
using AutopilotMonitor.DecisionCore.State;
@@ -7,72 +8,64 @@ using Xunit;
namespace AutopilotMonitor.DecisionCore.Tests
{
/// <summary>
/// Plan §4.x M4.4.4 — <c>ClassifierTick</c> is armed on <c>SessionStarted</c> so the
/// WhiteGlove classifier re-evaluates periodically from the start of every session,
/// not only reactively on the first WG-relevant signal (which never fires for non-WG
/// paths and closed the M3 re-trigger-lücke).
/// Wave 2 / H2 — the <c>ClassifierTick</c> is armed REACTIVELY on the first primary WhiteGlove
/// signal (shellcore_wg_success / sealing_pattern), NOT eagerly at <c>SessionStarted</c>.
/// <para>
/// Safety argument: the tick drives only <see cref="WhiteGloveSealingClassifier"/>, which can
/// reach <see cref="HypothesisLevel.Confirmed"/> (the sole level that seals, threshold 70) only
/// with a primary signal — the secondary-only ceiling is device_only(15)+system_reboot(15)=30.
/// Both primaries arm the tick via <c>AttachWhiteGloveClassifierEffects</c> and it re-arms until
/// terminal, so late secondary facts are still caught. Eager arming at SessionStarted only
/// produced ~720 no-op ticks/session on the ~95% of enrollments that never see a WG signal.
/// </para>
/// </summary>
public sealed class ClassifierTickFromSessionStartTests
{
private static readonly DateTime SessionStartUtc = new DateTime(2026, 4, 20, 10, 0, 0, DateTimeKind.Utc);
// Pass SessionStartUtc as the agent-boot anchor so the EffectiveDeadlineBase guard
// doesn't floor signal-time-based deadlines at the test runner's wall-clock now
// (which would always be later than the deterministic SessionStartUtc constant).
private static DecisionState FreshState() =>
DecisionState.CreateInitial("s", "t", SessionStartUtc);
[Fact]
public void SessionStarted_from_fresh_state_arms_ClassifierTick_deadline()
public void SessionStarted_does_NOT_arm_ClassifierTick()
{
// H2 core: SessionStarted is a pure lifecycle anchor — no periodic classifier tick.
var engine = new DecisionEngine();
var state = FreshState();
var step = engine.Reduce(state, MakeSessionStarted(ordinal: 0));
var step = engine.Reduce(FreshState(), MakeSessionStarted(ordinal: 0));
Assert.True(step.Transition.Taken);
Assert.Single(step.NewState.Deadlines);
Assert.Equal(DeadlineNames.ClassifierTick, step.NewState.Deadlines[0].Name);
Assert.Equal(SessionStage.SessionStarted, step.NewState.Stage);
Assert.Empty(step.NewState.Deadlines);
Assert.Empty(step.Effects);
}
[Fact]
public void Armed_ClassifierTick_dueAt_equals_signalOccurredAtUtc_plus_30s()
{
// Plan §2.6 — deadline dueAt is deterministic from the triggering signal's
// OccurredAtUtc, not from clock.UtcNow. Replay must reproduce it.
var engine = new DecisionEngine();
var state = FreshState();
var step = engine.Reduce(state, MakeSessionStarted(ordinal: 0));
var tick = step.NewState.Deadlines[0];
Assert.Equal(SessionStartUtc.AddSeconds(30), tick.DueAtUtc);
Assert.Equal(DecisionSignalKind.DeadlineFired, tick.FiresSignalKind);
Assert.NotNull(tick.FiresPayload);
Assert.Equal(DeadlineNames.ClassifierTick, tick.FiresPayload![SignalPayloadKeys.Deadline]);
}
[Fact]
public void SessionStarted_emits_ScheduleDeadline_effect_for_ClassifierTick()
public void First_primary_WhiteGlove_signal_arms_ClassifierTick_reactively()
{
var engine = new DecisionEngine();
var state = FreshState();
var state = engine.Reduce(FreshState(), MakeSessionStarted(ordinal: 0)).NewState;
Assert.Empty(state.Deadlines); // precondition: nothing armed yet
var step = engine.Reduce(state, MakeSessionStarted(ordinal: 0));
var wgAt = SessionStartUtc.AddMinutes(5);
var step = engine.Reduce(state, MakeSealingPatternDetected(ordinal: 1, occurredAtUtc: wgAt));
Assert.Single(step.Effects);
var effect = step.Effects[0];
Assert.Equal(DecisionEffectKind.ScheduleDeadline, effect.Kind);
Assert.NotNull(effect.Deadline);
Assert.Equal(DeadlineNames.ClassifierTick, effect.Deadline!.Name);
Assert.Equal(step.NewState.Deadlines[0].DueAtUtc, effect.Deadline.DueAtUtc);
// Deadline now armed, dueAt deterministic from the WG signal time (+30s).
var tick = Assert.Single(step.NewState.Deadlines);
Assert.Equal(DeadlineNames.ClassifierTick, tick.Name);
Assert.Equal(wgAt.AddSeconds(30), tick.DueAtUtc);
// The effects carry a ScheduleDeadline for the tick (plus the RunClassifier effect).
Assert.Contains(step.Effects, e =>
e.Kind == DecisionEffectKind.ScheduleDeadline
&& e.Deadline != null
&& e.Deadline.Name == DeadlineNames.ClassifierTick);
}
[Fact]
public void DeadEnd_SessionStarted_on_active_state_does_NOT_arm_ClassifierTick()
{
// Plan §2.5 — dead-end branch (replay of truncated log, session already mid-flight)
// must not reset state; specifically, no new deadline.
// Dead-end branch (replay of truncated log, session already mid-flight) must not
// reset state; specifically, no new deadline / effect.
var engine = new DecisionEngine();
var active = FreshState()
.ToBuilder()
@@ -89,29 +82,26 @@ namespace AutopilotMonitor.DecisionCore.Tests
}
[Fact]
public void ClassifierTick_from_session_start_persists_in_snapshot_and_fires_via_DeadlineFired()
public void Reactively_armed_ClassifierTick_fires_and_rearms()
{
// End-to-end: SessionStarted arms the tick; firing the deadline runs classifier +
// re-arms. Wiring depends only on the generic DeadlineFired dispatcher + the
// existing HandleClassifierTickDeadlineFired handler from M3.3+.
// End-to-end: a WG signal arms the tick; firing the deadline runs the classifier +
// re-arms (so late secondary facts are still picked up on genuine WG sessions).
var engine = new DecisionEngine();
var state = FreshState();
var state = engine.Reduce(FreshState(), MakeSessionStarted(ordinal: 0)).NewState;
var startStep = engine.Reduce(state, MakeSessionStarted(ordinal: 0));
Assert.Single(startStep.NewState.Deadlines);
var wgAt = SessionStartUtc.AddMinutes(5);
state = engine.Reduce(state, MakeSealingPatternDetected(ordinal: 1, occurredAtUtc: wgAt)).NewState;
Assert.Single(state.Deadlines);
var firedSignal = MakeDeadlineFired(
ordinal: 1,
occurredAtUtc: SessionStartUtc.AddSeconds(30),
deadlineName: DeadlineNames.ClassifierTick);
var firedStep = engine.Reduce(startStep.NewState, firedSignal);
var firedStep = engine.Reduce(state, MakeDeadlineFired(
ordinal: 2,
occurredAtUtc: wgAt.AddSeconds(30),
deadlineName: DeadlineNames.ClassifierTick));
Assert.True(firedStep.Transition.Taken);
// After fire: the deadline is cancelled + re-armed (same name, new dueAt).
Assert.Single(firedStep.NewState.Deadlines);
Assert.Equal(DeadlineNames.ClassifierTick, firedStep.NewState.Deadlines[0].Name);
Assert.Equal(SessionStartUtc.AddSeconds(60), firedStep.NewState.Deadlines[0].DueAtUtc);
var tick = Assert.Single(firedStep.NewState.Deadlines);
Assert.Equal(DeadlineNames.ClassifierTick, tick.Name);
Assert.Equal(wgAt.AddSeconds(60), tick.DueAtUtc); // re-armed +30s from the fire time
}
private static DecisionSignal MakeSessionStarted(long ordinal) =>
@@ -124,6 +114,16 @@ namespace AutopilotMonitor.DecisionCore.Tests
sourceOrigin: "test",
evidence: new Evidence(EvidenceKind.Synthetic, "session:started", "test"));
private static DecisionSignal MakeSealingPatternDetected(long ordinal, DateTime occurredAtUtc) =>
new DecisionSignal(
sessionSignalOrdinal: ordinal,
sessionTraceOrdinal: ordinal,
kind: DecisionSignalKind.WhiteGloveSealingPatternDetected,
kindSchemaVersion: 1,
occurredAtUtc: occurredAtUtc,
sourceOrigin: "test",
evidence: new Evidence(EvidenceKind.Synthetic, "wg:sealing-pattern", "test"));
private static DecisionSignal MakeDeadlineFired(long ordinal, DateTime occurredAtUtc, string deadlineName) =>
new DecisionSignal(
sessionSignalOrdinal: ordinal,
@@ -193,6 +193,46 @@ namespace AutopilotMonitor.DecisionCore.Tests
Assert.Equal("abc123", v.InputHash);
}
// ============================================================ admin preemption
[Theory]
// Review TRACE-M1: AdminPreemption terminal events previously carried parameters only and
// no audit trail, so an admin-overridden session could not be post-mortemed from backend
// telemetry like every other terminal outcome. Both outcomes must now carry the trail.
[InlineData("Succeeded", "enrollment_complete", nameof(SessionStage.Completed))]
[InlineData("Failed", "enrollment_failed", nameof(SessionStage.Failed))]
public void AdminPreemption_terminal_event_carries_audit_trail(
string adminOutcome, string expectedEventType, string expectedStage)
{
var engine = new DecisionEngine();
var state = DecisionState.CreateInitial("sess-admin", "tenant-admin");
state = engine.Reduce(state, MakeSignal(0, DecisionSignalKind.SessionStarted, T0)).NewState;
var step = engine.Reduce(state, MakeSignal(
1,
DecisionSignalKind.AdminPreemptionDetected,
T0.AddMinutes(1),
payload: new Dictionary<string, string> { ["adminOutcome"] = adminOutcome },
sourceOrigin: "register_session_response"));
var effect = SingleTimelineEffect(step, expectedEventType);
Assert.Equal(adminOutcome, effect.Parameters!["adminAction"]);
var trail = Assert.IsType<Dictionary<string, object>>(effect.TypedPayload);
Assert.Equal("DecisionEngine", trail["decisionSource"]);
Assert.Equal($"AdminPreemption:{adminOutcome}", trail["trigger"]);
Assert.Equal(expectedStage, trail["sessionStage"]);
Assert.IsType<List<string>>(trail["signalsSeen"]);
Assert.True(trail.ContainsKey("signalTimestamps"));
Assert.True(trail.ContainsKey("scenario"));
// Failure carries a reason; the success path does not synthesize one.
if (adminOutcome == "Failed")
Assert.Contains("administrator", (string)trail["reason"]);
else
Assert.False(trail.ContainsKey("reason"));
}
// ============================================================ whiteglove_complete
[Fact]
@@ -90,10 +90,10 @@ namespace AutopilotMonitor.DecisionCore.Tests
Assert.True(step.Transition.Taken);
Assert.Equal("SessionStarted", step.Transition.Trigger);
Assert.Null(step.Transition.DeadEndReason);
// M4.4.4: SessionStarted arms the ClassifierTick deadline up-front.
Assert.Single(step.Effects);
Assert.Equal(DecisionEffectKind.ScheduleDeadline, step.Effects[0].Kind);
Assert.Equal(DeadlineNames.ClassifierTick, step.Effects[0].Deadline!.Name);
// Wave 2 / H2: SessionStarted is a pure lifecycle anchor — it no longer arms the
// ClassifierTick (that is now reactive on the first primary WhiteGlove signal).
Assert.Empty(step.Effects);
Assert.Empty(step.NewState.Deadlines);
}
[Fact]
@@ -136,6 +136,10 @@ export const EVENT_TYPES_CATALOG = {
'stall_probe_result',
'session_stalled',
'spool_pressure_detected',
'collector_degraded',
'state_quarantine_recovered',
'telemetry_upload_poisoned',
'telemetry_upload_blocked',
],
metrics_events: ['performance_snapshot', 'agent_metrics_snapshot', 'ingress_backpressure'],
security_events: [
@@ -50,23 +50,23 @@ namespace AutopilotMonitor.DecisionCore.Engine
Array.Empty<DecisionEffect>());
}
// Arm ClassifierTick up-front so the White-Glove classifier re-evaluates
// periodically from the very start of the session — Plan §4.x M4.4.4.
// Floor the base at AgentBootUtc to keep the tick anchored to "now" even when
// SessionStarted itself ever carries a replayed historical timestamp.
var classifierTick = BuildClassifierTickDeadline(EffectiveDeadlineBase(state, signal));
// V2 race-fix (10c8e0bf debrief, 2026-04-26): SessionStarted is now a pure
// lifecycle anchor (stage / classifier-tick deadline / step bookkeeping).
// The registry-derived profile facts (enrollmentType + isHybridJoin) flow
// through the dedicated DecisionSignalKind.EnrollmentFactsObserved signal,
// which is stage-agnostic and therefore immune to the V2-Tracker /
// Backend-register-session race that previously dropped the JoinMode update.
// H2 (Wave 2): the ClassifierTick is NO LONGER armed here. It drives only the
// WhiteGloveSealingClassifier, which can reach Confirmed — the sole level that seals
// (HighThreshold=70) — only with a PRIMARY WG signal: shellcore_wg_success (+80) or
// sealing_pattern (+40). The secondary-only ceiling is device_only(+15) +
// system_reboot(+15) = 30 < 70, so before a primary signal the periodic tick can NEVER
// change the outcome. Both primary signals arm the tick reactively via
// AttachWhiteGloveClassifierEffects and it re-arms itself until terminal, so late
// secondary facts are still caught on genuine WG sessions. Arming from SessionStarted
// previously produced ~720 no-op ticks/session on the ~95% of enrollments that never
// see a WG signal — pure SignalLog + Journal + snapshot write-amplification with zero
// decision value. Profile facts (enrollmentType + isHybridJoin) flow through the
// dedicated EnrollmentFactsObserved signal; SessionStarted stays a pure lifecycle
// anchor (stage / step bookkeeping).
var builder = state.ToBuilder()
.WithStage(SessionStage.SessionStarted)
.WithStepIndex(state.StepIndex + 1)
.WithLastAppliedSignalOrdinal(signal.SessionSignalOrdinal)
.AddDeadline(classifierTick);
.WithLastAppliedSignalOrdinal(signal.SessionSignalOrdinal);
var newState = builder.Build();
@@ -77,12 +77,7 @@ namespace AutopilotMonitor.DecisionCore.Engine
nextStepIndex: newState.StepIndex,
trigger: nameof(DecisionSignalKind.SessionStarted));
var effects = new DecisionEffect[]
{
new DecisionEffect(DecisionEffectKind.ScheduleDeadline, deadline: classifierTick),
};
return new DecisionStep(newState, transition, effects);
return new DecisionStep(newState, transition, Array.Empty<DecisionEffect>());
}
/// <summary>
@@ -294,6 +289,7 @@ namespace AutopilotMonitor.DecisionCore.Engine
effects.Add(BuildPhaseTransitionEffect(EnrollmentPhase.FinalizingSetup));
effects.Add(BuildPhaseTransitionEffect(EnrollmentPhase.Complete));
}
var adminReason = $"Session {adminOutcome.ToLowerInvariant()} by administrator (detected on register-session).";
effects.Add(new DecisionEffect(
DecisionEffectKind.EmitEventTimelineEntry,
parameters: new Dictionary<string, string>
@@ -301,8 +297,17 @@ namespace AutopilotMonitor.DecisionCore.Engine
["eventType"] = eventType,
["adminAction"] = adminOutcome,
["source"] = signal.SourceOrigin ?? "register_session_response",
["reason"] = $"Session {adminOutcome.ToLowerInvariant()} by administrator (detected on register-session).",
}));
["reason"] = adminReason,
},
// Parity with the 7 other terminal/state-changing sites (review TRACE-M1): without
// the audit trail an admin-preempted session's terminal event carried no
// signalsSeen / signalTimestamps / scenario census, so it could not be
// post-mortemed from backend telemetry like every other terminal outcome.
typedPayload: DecisionAuditTrailBuilder.Build(
postState: newState,
decidedStage: toStage,
trigger: $"AdminPreemption:{adminOutcome}",
failureReason: succeeded ? null : adminReason)));
return new DecisionStep(newState, transition, effects.ToArray());
}
@@ -24,9 +24,19 @@ namespace AutopilotMonitor.DecisionCore.Serialization
public static class DecisionCoreJsonSettings
{
/// <summary>
/// Create a fresh settings instance. Not a static singleton — Newtonsoft mutates
/// internal state during serialization (binder caches, etc.), so handing out
/// shared instances across threads is a well-known footgun.
/// Shared settings instance for the hot (de)serialization paths (M5). Safe to reuse across
/// threads: <see cref="JsonConvert"/> builds a fresh <c>JsonSerializer</c> per call from the
/// settings (no shared serializer-level mutable state), and every registered converter
/// (<see cref="UnknownFallbackEnumConverter{T}"/>) is stateless. Reusing this avoids
/// allocating a settings object + 12 converters on every signal / transition / snapshot
/// (de)serialize. <b>Do not mutate</b> (e.g. add converters) — call <see cref="Create"/>
/// for a private, mutable copy.
/// </summary>
public static JsonSerializerSettings Shared { get; } = Create();
/// <summary>
/// Create a fresh settings instance. Used by callers that need a private, mutable copy
/// (e.g. tests); the production serializers use <see cref="Shared"/>.
/// </summary>
public static JsonSerializerSettings Create()
{
@@ -21,14 +21,14 @@ namespace AutopilotMonitor.DecisionCore.Serialization
public static string Serialize(DecisionSignal signal)
{
if (signal == null) throw new ArgumentNullException(nameof(signal));
return JsonConvert.SerializeObject(signal, DecisionCoreJsonSettings.Create());
return JsonConvert.SerializeObject(signal, DecisionCoreJsonSettings.Shared);
}
public static DecisionSignal Deserialize(string line)
{
if (line == null) throw new ArgumentNullException(nameof(line));
var settings = DecisionCoreJsonSettings.Create();
var settings = DecisionCoreJsonSettings.Shared;
var obj = JsonConvert.DeserializeObject<JObject>(line, settings)
?? throw new JsonReaderException($"Failed to parse line as JSON object: {line}");
@@ -20,7 +20,7 @@ namespace AutopilotMonitor.DecisionCore.Serialization
public static string Serialize(DecisionState state)
{
if (state == null) throw new ArgumentNullException(nameof(state));
return JsonConvert.SerializeObject(state, DecisionCoreJsonSettings.Create());
return JsonConvert.SerializeObject(state, DecisionCoreJsonSettings.Shared);
}
public static DecisionState Deserialize(string payload)
@@ -29,7 +29,7 @@ namespace AutopilotMonitor.DecisionCore.Serialization
try
{
var result = JsonConvert.DeserializeObject<DecisionState>(payload, DecisionCoreJsonSettings.Create());
var result = JsonConvert.DeserializeObject<DecisionState>(payload, DecisionCoreJsonSettings.Shared);
if (result == null)
{
throw new JsonSerializationException("State deserialization produced null.");
@@ -19,7 +19,7 @@ namespace AutopilotMonitor.DecisionCore.Serialization
public static string Serialize(DecisionTransition transition)
{
if (transition == null) throw new ArgumentNullException(nameof(transition));
return JsonConvert.SerializeObject(transition, DecisionCoreJsonSettings.Create());
return JsonConvert.SerializeObject(transition, DecisionCoreJsonSettings.Shared);
}
public static DecisionTransition Deserialize(string line)
@@ -28,7 +28,7 @@ namespace AutopilotMonitor.DecisionCore.Serialization
try
{
var result = JsonConvert.DeserializeObject<DecisionTransition>(line, DecisionCoreJsonSettings.Create());
var result = JsonConvert.DeserializeObject<DecisionTransition>(line, DecisionCoreJsonSettings.Shared);
if (result == null)
{
throw new JsonSerializationException($"Transition deserialization produced null for line: {line}");
@@ -287,6 +287,10 @@ namespace AutopilotMonitor.Shared
public const string PerformanceCollectorStopped = "performance_collector_stopped"; // Idle-stop anchor — fired after 15 min idle by PerformanceCollector. PR1.
public const string AgentMetricsCollectorStopped = "agent_metrics_collector_stopped"; // Idle-stop anchor — fired after 15 min idle by AgentMetricsCollector. PR1.
public const string PriorRunDiedWithState = "prior_run_died_with_state"; // Death-Rattle (Plan §B) — emitted on next run if previous exit was unclean. PR1.
public const string CollectorDegraded = "collector_degraded"; // One-shot Warning when a collector/watcher fails to arm (e.g. EventLogWatcher start error). Review MON-D1: without it a dead kernel watcher is indistinguishable from a genuine no-signal session.
public const string StateQuarantineRecovered = "state_quarantine_recovered"; // Warning emitted on the start that honours a prior run's quarantine marker (3 consecutive journal-append failures). Review TRACE-H2: without it a corruption-recovery looks like a normal restart on the backend.
public const string TelemetryUploadPoisoned = "telemetry_upload_poisoned"; // Warning when the drain quarantined a single oversized item (413 even on its own — the only locally-provable poison) to keep the pipeline flowing. Review TRACE-H1/P1.
public const string TelemetryUploadBlocked = "telemetry_upload_blocked"; // Warning (one-shot) when the drain is blocked by a retained unknown permanent 4xx (400 contract bug / 404 route mismatch / tenant-validation). Data is NOT discarded. LOCAL diagnostic (agent log + diag ZIP): this marker is queued behind the blocking batch and only reaches the backend after the block clears — no out-of-band bypass by design.
// IME log tracker — app / device / script telemetry (V2 single-rail plan §5.9)
public const string EspPhaseChanged = "esp_phase_changed";