feat(agent): V2 WhiteGlove Part 1 graceful-exit hardening

Three coordinated fixes so a WhiteGlove Part-1 reseal-reboot does not race
the agent's termination path and leave the timeline missing
agent_shutting_down / whiteglove_part1_complete events plus a stale
previousExit=reboot_kill on the next run.

- Real spool-empty drain: EnrollmentTerminationHandler.DrainSpool polls
  PendingItemCount + IngressPendingSignalCount instead of blind Task.Delay,
  so it returns the moment the backend actually ack'd the lifecycle events
  (or hits the bounded fallback timeout).

- Early clean-exit.marker write: marker is written before _signalShutdown
  returns control, in addition to the existing AppDomain.ProcessExit
  fallback. Wins the race against admin-triggered shutdown.exe.

- Off-worker termination dispatch: Terminated event from
  OnDecisionTerminalStage now goes through Task.Run so the SignalIngress
  worker is freed to process events the handler itself just posted. Adds
  PendingSignalCount with race-free posted-not-yet-completed semantics
  (Interlocked.Increment in Post BEFORE channel.Add, decrement in
  ProcessItem finally + add-throw paths). Two-phase drain (ingress queue
  -> spool) replaces the previous spool-only wait.

- Fast-path inline classifier on WhiteGloveShellCoreSuccess: emits
  whiteglove_complete in a single reduce step with the full
  DecisionAuditTrailBuilder TypedPayload (decisionSource, trigger,
  signalsSeen, classifier verdict, classifierInputs, scenario), preserving
  byte-stable forensics across fast and slow paths.

Tests:
- DecisionCore: fast-path audit-trail completeness, fast-path single-step
  transition, slow-path fallback when AccountSetup activity excludes WG,
  scenario-replay run/skip counts.
- V2 Core: Termination two-phase drain (immediate / poll-until-zero /
  timeout-fallback / blind-delay-fallback / accessor-throw safe), early
  clean-exit-marker ordering vs signalShutdown, off-worker
  Terminated-dispatch verified by Thread.Name + handler-posted-signal
  drain, SignalIngress PendingSignalCount idle/in-flight/post-stop and
  deterministic race-leak regression via internal test seam.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
Oliver Kieselbach
2026-04-30 18:05:37 +02:00
parent 9f3ee5eedb
commit ba0fb7ae0c
15 changed files with 1183 additions and 33 deletions
@@ -0,0 +1,129 @@
using System;
using System.Collections.Generic;
using System.Threading;
using AutopilotMonitor.Agent.V2.Core.Orchestration;
using AutopilotMonitor.Agent.V2.Core.Tests.Harness;
using AutopilotMonitor.DecisionCore.Classifiers;
using AutopilotMonitor.DecisionCore.Engine;
using AutopilotMonitor.DecisionCore.Signals;
using Xunit;
namespace AutopilotMonitor.Agent.V2.Core.Tests.Orchestration
{
/// <summary>
/// Codex review follow-up (Finding 2, 2026-04-30) — the <see cref="EnrollmentOrchestrator.Terminated"/>
/// event must NOT fire on the SignalIngress worker thread. The previous synchronous
/// dispatch was a deadlock-by-design for any handler that:
/// <list type="bullet">
/// <item>Posts new lifecycle events back to the ingress (the worker can only process
/// them after the handler returns), AND</item>
/// <item>Then waits for those events to drain (the ingress can only drain on the
/// worker thread, which the handler is blocking).</item>
/// </list>
/// The orchestrator now wraps the synchronous <c>Terminated?.Invoke</c> in a
/// <see cref="System.Threading.Tasks.Task.Run"/> so the worker is freed immediately and
/// the handler runs on a thread-pool thread that CAN wait for ingress drain.
/// </summary>
[Collection("SerialThreading")]
public sealed class EnrollmentOrchestratorTerminatedDispatchTests
{
private static DateTime At => new DateTime(2026, 4, 30, 10, 0, 0, DateTimeKind.Utc);
[Fact]
public void DecisionTerminalStage_dispatches_Terminated_off_ingress_worker_thread()
{
using var rig = new EnrollmentOrchestratorRig(At);
// Production wires WhiteGloveSealingClassifier — the test needs it too because
// we drive the engine to a terminal stage via the WG fast-path.
rig.Classifiers.Add(new WhiteGloveSealingClassifier());
var sut = rig.Build();
string? handlerThreadName = null;
using var fired = new ManualResetEventSlim(false);
sut.Terminated += (_, _) =>
{
handlerThreadName = Thread.CurrentThread.Name;
fired.Set();
};
sut.Start();
try
{
// Drive the engine to WhiteGloveSealed via the strong WG signal — fast-path
// transitions in a single reduce step, OnDecisionTerminalStage fires.
sut.IngressSink.Post(
kind: DecisionSignalKind.WhiteGloveShellCoreSuccess,
occurredAtUtc: At.AddMinutes(1),
sourceOrigin: "ShellCoreTracker",
evidence: new Evidence(EvidenceKind.Raw, "ShellCore-62407", "WhiteGlove_Success"));
Assert.True(fired.Wait(TimeSpan.FromSeconds(5)),
"Terminated event did not fire within 5s.");
// The SignalIngress worker thread is named "SignalIngress.Worker". The
// handler must NOT have observed itself running there — Task.Run sends it
// to a thread-pool worker, which has no name (null) or "ThreadPool*" name.
Assert.NotEqual("SignalIngress.Worker", handlerThreadName);
}
finally { sut.Stop(); }
}
[Fact]
public void Off_worker_dispatch_lets_handler_post_signals_that_get_processed()
{
using var rig = new EnrollmentOrchestratorRig(At);
rig.Classifiers.Add(new WhiteGloveSealingClassifier());
var sut = rig.Build();
using var handlerEnteredFromTask = new ManualResetEventSlim(false);
using var handlerCompletedDrain = new ManualResetEventSlim(false);
sut.Terminated += (_, _) =>
{
handlerEnteredFromTask.Set();
// Post a follow-up signal from inside the handler. With synchronous
// dispatch (the OLD behaviour) this would just enqueue and never get
// processed because the worker is blocked on us. With off-worker dispatch
// the worker is free, processes the signal, and PendingSignalCount goes
// back to 0.
sut.IngressSink.Post(
kind: DecisionSignalKind.InformationalEvent,
occurredAtUtc: At.AddMinutes(2),
sourceOrigin: "TerminationHandlerTest",
evidence: new Evidence(EvidenceKind.Synthetic, "lifecycle-event", "from handler"),
payload: new Dictionary<string, string>
{
[SignalPayloadKeys.EventType] = "test_lifecycle_event",
[SignalPayloadKeys.Source] = "test",
});
// Wait until the worker actually drained our just-posted signal — must
// settle within a generous bound; if dispatch were synchronous this would
// time out and the test would fail loudly.
var deadline = DateTime.UtcNow + TimeSpan.FromSeconds(5);
while (sut.IngressPendingSignalCount > 0 && DateTime.UtcNow < deadline)
{
Thread.Sleep(20);
}
handlerCompletedDrain.Set();
};
sut.Start();
try
{
sut.IngressSink.Post(
kind: DecisionSignalKind.WhiteGloveShellCoreSuccess,
occurredAtUtc: At.AddMinutes(1),
sourceOrigin: "ShellCoreTracker",
evidence: new Evidence(EvidenceKind.Raw, "ShellCore-62407", "WhiteGlove_Success"));
Assert.True(handlerEnteredFromTask.Wait(TimeSpan.FromSeconds(5)),
"Terminated handler did not start.");
Assert.True(handlerCompletedDrain.Wait(TimeSpan.FromSeconds(10)),
"Handler-posted signal did not drain — worker is likely blocked, regression of off-worker dispatch.");
Assert.Equal(0L, sut.IngressPendingSignalCount);
}
finally { sut.Stop(); }
}
}
}
@@ -783,5 +783,137 @@ namespace AutopilotMonitor.Agent.V2.Core.Tests.Orchestration
Assert.True(traces[i] > traces[i - 1]);
}
}
// ============================================================ Codex Finding 2 —
// PendingSignalCount: posted-but-not-yet-fully-processed counter used by the
// termination-handler off-worker drain.
[Fact]
public void PendingSignalCount_is_zero_when_idle()
{
using var rig = new Rig();
var ing = rig.Build();
ing.Start();
try
{
Assert.Equal(0L, ing.PendingSignalCount);
}
finally { ing.Stop(); }
}
[Fact]
public void PendingSignalCount_returns_to_zero_after_post_is_processed()
{
using var rig = new Rig();
var ing = rig.Build();
ing.Start();
try
{
ing.Post(DecisionSignalKind.SessionStarted, At, "Collector", RawEvidence());
// The worker is async — give it room to fully process and decrement.
Assert.True(WaitFor(() => ing.PendingSignalCount == 0L),
"PendingSignalCount did not return to 0 after post was processed.");
}
finally { ing.Stop(); }
}
[Fact]
public void PendingSignalCount_observes_in_flight_item()
{
using var rig = new Rig();
// Block ApplyStep on a gate so we can observe a non-zero pending count while
// the worker is mid-process.
var gate = new ManualResetEventSlim(false);
rig.Processor.BlockHandle = gate;
var ing = rig.Build();
ing.Start();
try
{
ing.Post(DecisionSignalKind.SessionStarted, At, "Collector", RawEvidence());
// The worker has either dequeued and is waiting on gate, or is about to —
// both states are observable as PendingSignalCount >= 1 because the Post
// increment happens before the channel.Add publishes the item.
Assert.True(WaitFor(() => ing.PendingSignalCount >= 1L),
"PendingSignalCount stayed at 0 even though an item was in flight.");
}
finally
{
gate.Set();
ing.Stop();
}
}
[Fact]
public void PendingSignalCount_decremented_on_post_after_stop_failure()
{
using var rig = new Rig();
var ing = rig.Build();
ing.Start();
ing.Stop();
// Post after Stop throws — the increment in Post must NOT leak. Verify the
// counter is still at 0 after the throw.
Assert.Throws<InvalidOperationException>(() =>
ing.Post(DecisionSignalKind.SessionStarted, At, "Collector", RawEvidence()));
Assert.Equal(0L, ing.PendingSignalCount);
}
[Fact]
public void PendingSignalCount_does_not_leak_when_TryAdd_throws_after_completed_race()
{
// Codex follow-up (Low, 2026-04-30) — Post() has a narrow race window:
// 1. Top-of-Post check sees _stopRequested==0 && !IsAddingCompleted (live)
// 2. Stop runs concurrently: _stopRequested=1, then channel.CompleteAdding()
// 3. Post continues to Interlocked.Increment(_unprocessedCount)
// 4. Post calls _channel.TryAdd(item) → InvalidOperationException (channel
// is now completed)
//
// Without the symmetric catch on TryAdd the just-incremented counter would
// leak. The window is too narrow to hit stochastically, so we reproduce it
// deterministically via the internal _testHookBetweenPostStopCheckAndAdd
// seam: the hook fires AFTER the post-stop check but BEFORE TryAdd, so we
// can call CompleteAdding on the private channel inside it — exactly the
// ordering the production race produces.
using var rig = new Rig();
var ing = rig.Build();
ing.Start();
// Grab the private _channel via reflection — we'll invoke CompleteAdding
// from inside the test hook to simulate Stop running in the race window.
var channelField = typeof(SignalIngress).GetField(
"_channel",
System.Reflection.BindingFlags.Instance | System.Reflection.BindingFlags.NonPublic);
Assert.NotNull(channelField);
var channel = channelField!.GetValue(ing);
Assert.NotNull(channel);
var completeAdding = channel!.GetType().GetMethod("CompleteAdding");
Assert.NotNull(completeAdding);
// Wire the hook — fires once, between the post-stop check and TryAdd. By
// calling CompleteAdding inside it, the next TryAdd will throw exactly the
// production-race InvalidOperationException.
int hookFired = 0;
ing._testHookBetweenPostStopCheckAndAdd = () =>
{
if (Interlocked.Exchange(ref hookFired, 1) == 0)
{
completeAdding!.Invoke(channel, Array.Empty<object>());
}
};
Assert.Throws<InvalidOperationException>(() =>
ing.Post(DecisionSignalKind.SessionStarted, At, "Collector", RawEvidence()));
// Hook ran exactly once and the catch reset the counter. Without the catch
// (regression), this assert would observe 1 — proving the test actually
// exercises the bug.
Assert.Equal(1, hookFired);
Assert.Equal(0L, ing.PendingSignalCount);
// Don't call Stop() — the channel is already completed; Stop's internal
// CompleteAdding would throw. Dispose handles cleanup via the Rig.
}
}
}
@@ -23,6 +23,13 @@ using Xunit;
namespace AutopilotMonitor.Agent.V2.Core.Tests.Termination
{
// Codex Finding 2 follow-up — the new DrainSpool_* tests use Stopwatch-bound
// assertions on Task.Delay timings that flake under full-suite parallel load (the
// Stopwatch sees thread-pool contention as drift beyond the assertion margins).
// SerialThreading serialises this class against the other timing-sensitive ones so
// CI runs deterministically. Cost is small — most tests in the class are sub-second
// and just live alongside the timing-sensitive ones.
[Collection("SerialThreading")]
public sealed class EnrollmentTerminationHandlerTests
{
private static DateTime StartUtc => new DateTime(2026, 4, 21, 10, 0, 0, DateTimeKind.Utc);
@@ -128,6 +135,17 @@ namespace AutopilotMonitor.Agent.V2.Core.Tests.Termination
public IReadOnlyDictionary<string, AppInstallTiming>? AppTimingsOverride { get; set; }
// Option 1+2 (WG Part 1 graceful-exit hardening, 2026-04-30) test hooks.
// Tests that exercise the new behaviour set these directly; tests that don't
// care leave them null (handler falls back to the legacy blind-delay drain
// and skips the early marker write).
public Func<int>? PendingItemCountAccessor;
public Func<long>? IngressPendingSignalCountAccessor;
public Action? WriteCleanExitMarkerHook;
public int CleanExitMarkerWrites;
public List<string> TerminationActionLog { get; } = new List<string>();
public TimeSpan SpoolDrainPeriodOverride { get; set; } = TimeSpan.Zero;
public EnrollmentTerminationHandler Build(AgentConfiguration? config = null) =>
BuildCore(config, agentVersion: null);
@@ -152,7 +170,11 @@ namespace AutopilotMonitor.Agent.V2.Core.Tests.Termination
LastDiagnosticsSuffix = suffix;
return Task.FromResult(DiagnosticsResult ?? new DiagnosticsUploadResult { BlobName = "blob" });
},
signalShutdown: () => Interlocked.Increment(ref ShutdownSignalled),
signalShutdown: () =>
{
Interlocked.Increment(ref ShutdownSignalled);
TerminationActionLog.Add("signalShutdown");
},
analyzerManager: null,
post: Post,
sessionPersistence: SessionPersistence,
@@ -164,9 +186,16 @@ namespace AutopilotMonitor.Agent.V2.Core.Tests.Termination
// Zero out the timing ceremony for tests — production paths are covered by
// the dedicated V1-parity tests below which opt back in via their own Rig.
lateEventGracePeriod: TimeSpan.Zero,
spoolDrainPeriod: TimeSpan.Zero,
spoolDrainPeriod: SpoolDrainPeriodOverride,
appTimingsAccessor: () => AppTimingsOverride ?? new Dictionary<string, AppInstallTiming>(),
agentVersion: agentVersion);
agentVersion: agentVersion,
pendingItemCountAccessor: PendingItemCountAccessor,
writeCleanExitMarker: WriteCleanExitMarkerHook ?? new Action(() =>
{
Interlocked.Increment(ref CleanExitMarkerWrites);
TerminationActionLog.Add("writeCleanExitMarker");
}),
ingressPendingSignalCountAccessor: IngressPendingSignalCountAccessor);
}
public void Dispose() => Tmp.Dispose();
@@ -568,5 +597,238 @@ namespace AutopilotMonitor.Agent.V2.Core.Tests.Termination
Assert.NotNull(payload);
Assert.Equal("true", payload![SignalPayloadKeys.ImmediateUpload]);
}
// ============================================================= Option 1+2 — WG Part 1
// graceful-exit hardening (2026-04-30). Active spool-empty drain + early clean-exit
// marker write before _signalShutdown returns control to the main thread.
// DrainSpool is invoked on the WG Part 1 + standalone-reboot paths. We exercise it
// through the WG Part 1 path here (Stage=WhiteGloveSealed → DrainSpool runs in the
// graceful-exit branch).
[Fact]
public void DrainSpool_returns_immediately_when_pending_count_is_zero()
{
using var rig = new Rig();
rig.State = new DecisionStateBuilder(DecisionState.CreateInitial("S1", "T1")) { Stage = SessionStage.WhiteGloveSealed }.Build();
// Long bounded timeout — if the active drain is broken the test would hit this.
rig.SpoolDrainPeriodOverride = TimeSpan.FromSeconds(5);
rig.PendingItemCountAccessor = () => 0;
var sw = System.Diagnostics.Stopwatch.StartNew();
rig.Build().Handle(sender: null!,
Args(EnrollmentTerminationReason.DecisionTerminalStage, EnrollmentTerminationOutcome.Succeeded, SessionStage.WhiteGloveSealed));
sw.Stop();
Assert.True(sw.Elapsed < TimeSpan.FromSeconds(2),
$"DrainSpool should exit immediately on pending=0 but took {sw.ElapsedMilliseconds}ms.");
}
[Fact]
public void DrainSpool_polls_until_pending_drops_to_zero()
{
using var rig = new Rig();
rig.State = new DecisionStateBuilder(DecisionState.CreateInitial("S1", "T1")) { Stage = SessionStage.WhiteGloveSealed }.Build();
rig.SpoolDrainPeriodOverride = TimeSpan.FromSeconds(5);
// Simulate the spool draining ~250ms after the accessor is first observed.
var startedAt = DateTime.UtcNow;
rig.PendingItemCountAccessor = () =>
(DateTime.UtcNow - startedAt) > TimeSpan.FromMilliseconds(250) ? 0 : 3;
var sw = System.Diagnostics.Stopwatch.StartNew();
rig.Build().Handle(sender: null!,
Args(EnrollmentTerminationReason.DecisionTerminalStage, EnrollmentTerminationOutcome.Succeeded, SessionStage.WhiteGloveSealed));
sw.Stop();
Assert.True(sw.Elapsed >= TimeSpan.FromMilliseconds(200),
$"Drain returned too early (elapsed={sw.ElapsedMilliseconds}ms).");
Assert.True(sw.Elapsed < TimeSpan.FromSeconds(2),
$"Drain did not exit early after spool drained (elapsed={sw.ElapsedMilliseconds}ms).");
}
[Fact]
public void DrainSpool_falls_back_to_timeout_when_pending_never_drains()
{
using var rig = new Rig();
rig.State = new DecisionStateBuilder(DecisionState.CreateInitial("S1", "T1")) { Stage = SessionStage.WhiteGloveSealed }.Build();
// Tight timeout to keep the test fast.
rig.SpoolDrainPeriodOverride = TimeSpan.FromMilliseconds(400);
rig.PendingItemCountAccessor = () => 5; // Always pending — drain never satisfies.
var sw = System.Diagnostics.Stopwatch.StartNew();
rig.Build().Handle(sender: null!,
Args(EnrollmentTerminationReason.DecisionTerminalStage, EnrollmentTerminationOutcome.Succeeded, SessionStage.WhiteGloveSealed));
sw.Stop();
Assert.True(sw.Elapsed >= TimeSpan.FromMilliseconds(350),
$"Drain returned before bounded timeout (elapsed={sw.ElapsedMilliseconds}ms).");
}
[Fact]
public void DrainSpool_falls_back_to_blind_delay_when_no_accessor_is_wired()
{
using var rig = new Rig();
rig.State = new DecisionStateBuilder(DecisionState.CreateInitial("S1", "T1")) { Stage = SessionStage.WhiteGloveSealed }.Build();
rig.SpoolDrainPeriodOverride = TimeSpan.FromMilliseconds(300);
rig.PendingItemCountAccessor = null; // Legacy / null path.
var sw = System.Diagnostics.Stopwatch.StartNew();
rig.Build().Handle(sender: null!,
Args(EnrollmentTerminationReason.DecisionTerminalStage, EnrollmentTerminationOutcome.Succeeded, SessionStage.WhiteGloveSealed));
sw.Stop();
// Without an accessor the drain just sleeps the full bounded period — verifies
// the V1-parity fallback path stays intact.
Assert.True(sw.Elapsed >= TimeSpan.FromMilliseconds(280),
$"Blind delay returned too early (elapsed={sw.ElapsedMilliseconds}ms).");
}
[Fact]
public void WhiteGlovePart1_writes_clean_exit_marker_before_signaling_shutdown()
{
using var rig = new Rig();
rig.State = new DecisionStateBuilder(DecisionState.CreateInitial("S1", "T1")) { Stage = SessionStage.WhiteGloveSealed }.Build();
rig.Build().Handle(sender: null!,
Args(EnrollmentTerminationReason.DecisionTerminalStage, EnrollmentTerminationOutcome.Succeeded, SessionStage.WhiteGloveSealed));
Assert.Equal(1, rig.CleanExitMarkerWrites);
// Order matters — marker must be on disk BEFORE the main thread is released to
// exit. Otherwise an admin reseal-reboot can race the AppDomain.ProcessExit
// fallback handler and we'd be classified as reboot_kill on the next boot.
var markerIdx = rig.TerminationActionLog.IndexOf("writeCleanExitMarker");
var shutdownIdx = rig.TerminationActionLog.IndexOf("signalShutdown");
Assert.True(markerIdx >= 0, "writeCleanExitMarker hook must be invoked on WG Part 1 exit.");
Assert.True(shutdownIdx > markerIdx,
$"writeCleanExitMarker must precede signalShutdown (got marker@{markerIdx}, shutdown@{shutdownIdx}).");
}
[Fact]
public void Standard_terminal_path_writes_clean_exit_marker_too()
{
using var rig = new Rig();
rig.State = new DecisionStateBuilder(DecisionState.CreateInitial("S1", "T1")) { Stage = SessionStage.Completed }.Build();
rig.Build().Handle(sender: null!,
Args(EnrollmentTerminationReason.DecisionTerminalStage, EnrollmentTerminationOutcome.Succeeded, SessionStage.Completed));
Assert.Equal(1, rig.CleanExitMarkerWrites);
var markerIdx = rig.TerminationActionLog.IndexOf("writeCleanExitMarker");
var shutdownIdx = rig.TerminationActionLog.IndexOf("signalShutdown");
Assert.True(markerIdx >= 0 && shutdownIdx > markerIdx);
}
[Fact]
public void CleanExitMarker_writer_failure_does_not_break_termination()
{
using var rig = new Rig();
rig.State = new DecisionStateBuilder(DecisionState.CreateInitial("S1", "T1")) { Stage = SessionStage.WhiteGloveSealed }.Build();
// Replace the default with a throwing hook to verify the handler swallows + logs
// and still reaches signalShutdown.
rig.WriteCleanExitMarkerHook = () => throw new System.IO.IOException("disk full");
rig.Build().Handle(sender: null!,
Args(EnrollmentTerminationReason.DecisionTerminalStage, EnrollmentTerminationOutcome.Succeeded, SessionStage.WhiteGloveSealed));
Assert.Equal(1, rig.ShutdownSignalled);
}
// ============================================================== Codex Finding 2 —
// off-worker dispatch + two-phase drain (ingress queue → spool). The drain MUST
// wait for the ingress to settle BEFORE polling spool-empty, otherwise events the
// handler itself just posted are still in the channel when DrainSpool returns.
[Fact]
public void DrainSpool_waits_for_ingress_to_drain_before_polling_spool()
{
using var rig = new Rig();
rig.State = new DecisionStateBuilder(DecisionState.CreateInitial("S1", "T1")) { Stage = SessionStage.WhiteGloveSealed }.Build();
rig.SpoolDrainPeriodOverride = TimeSpan.FromSeconds(5);
// Simulate the ingress holding the lifecycle events for ~250ms before the
// worker processes them; spool stays empty (events haven't been spooled yet).
// Once ingress drains, spool is also empty (no late uploads to wait for).
int spoolPollsObserved = 0;
var ingressIdleAt = DateTime.UtcNow + TimeSpan.FromMilliseconds(250);
rig.IngressPendingSignalCountAccessor = () =>
DateTime.UtcNow >= ingressIdleAt ? 0L : 4L;
rig.PendingItemCountAccessor = () => { spoolPollsObserved++; return 0; };
var sw = System.Diagnostics.Stopwatch.StartNew();
rig.Build().Handle(sender: null!,
Args(EnrollmentTerminationReason.DecisionTerminalStage, EnrollmentTerminationOutcome.Succeeded, SessionStage.WhiteGloveSealed));
sw.Stop();
// The drain MUST have waited until the ingress drained (>=200ms) — anything
// shorter means we polled spool-empty before the handler-posted events hit it.
Assert.True(sw.Elapsed >= TimeSpan.FromMilliseconds(200),
$"Drain returned before ingress drained (elapsed={sw.ElapsedMilliseconds}ms).");
Assert.True(sw.Elapsed < TimeSpan.FromSeconds(2),
$"Drain did not exit early after ingress + spool both went idle (elapsed={sw.ElapsedMilliseconds}ms).");
// Spool must have been polled at least once after ingress idle — proves Phase B ran.
Assert.True(spoolPollsObserved >= 1, "Spool poll did not run after ingress drain.");
}
[Fact]
public void DrainSpool_two_phase_budget_is_shared_across_ingress_and_spool()
{
using var rig = new Rig();
rig.State = new DecisionStateBuilder(DecisionState.CreateInitial("S1", "T1")) { Stage = SessionStage.WhiteGloveSealed }.Build();
// Tight shared budget — the ingress phase eats most of it, the spool phase
// gets whatever's left and times out cleanly.
rig.SpoolDrainPeriodOverride = TimeSpan.FromMilliseconds(400);
rig.IngressPendingSignalCountAccessor = () => 7L; // never drains
rig.PendingItemCountAccessor = () => 3; // never drains either
var sw = System.Diagnostics.Stopwatch.StartNew();
rig.Build().Handle(sender: null!,
Args(EnrollmentTerminationReason.DecisionTerminalStage, EnrollmentTerminationOutcome.Succeeded, SessionStage.WhiteGloveSealed));
sw.Stop();
// Total drain MUST be bounded by the shared budget — the spool phase doesn't
// get its OWN 400ms on top of the ingress phase.
Assert.True(sw.Elapsed >= TimeSpan.FromMilliseconds(350),
$"Drain returned too early (elapsed={sw.ElapsedMilliseconds}ms).");
Assert.True(sw.Elapsed < TimeSpan.FromMilliseconds(900),
$"Drain exceeded shared budget (elapsed={sw.ElapsedMilliseconds}ms) — phases must share, not stack.");
}
[Fact]
public void DrainSpool_skips_ingress_phase_when_only_spool_accessor_wired()
{
using var rig = new Rig();
rig.State = new DecisionStateBuilder(DecisionState.CreateInitial("S1", "T1")) { Stage = SessionStage.WhiteGloveSealed }.Build();
rig.SpoolDrainPeriodOverride = TimeSpan.FromSeconds(5);
// No ingress accessor — handler skips Phase A and just polls spool.
rig.IngressPendingSignalCountAccessor = null;
rig.PendingItemCountAccessor = () => 0;
var sw = System.Diagnostics.Stopwatch.StartNew();
rig.Build().Handle(sender: null!,
Args(EnrollmentTerminationReason.DecisionTerminalStage, EnrollmentTerminationOutcome.Succeeded, SessionStage.WhiteGloveSealed));
sw.Stop();
Assert.True(sw.Elapsed < TimeSpan.FromSeconds(1),
$"Spool-only drain should return promptly (elapsed={sw.ElapsedMilliseconds}ms).");
}
[Fact]
public void DrainSpool_ingress_accessor_failure_does_not_break_termination()
{
using var rig = new Rig();
rig.State = new DecisionStateBuilder(DecisionState.CreateInitial("S1", "T1")) { Stage = SessionStage.WhiteGloveSealed }.Build();
rig.SpoolDrainPeriodOverride = TimeSpan.FromMilliseconds(200);
rig.IngressPendingSignalCountAccessor = () => throw new InvalidOperationException("orchestrator stopped");
rig.PendingItemCountAccessor = () => 0;
// Must not throw out of Handle, must still call signalShutdown.
rig.Build().Handle(sender: null!,
Args(EnrollmentTerminationReason.DecisionTerminalStage, EnrollmentTerminationOutcome.Succeeded, SessionStage.WhiteGloveSealed));
Assert.Equal(1, rig.ShutdownSignalled);
}
}
}
@@ -256,6 +256,28 @@ namespace AutopilotMonitor.Agent.V2.Core.Orchestration
public TelemetryUploadOrchestrator Transport =>
_transport ?? throw new InvalidOperationException("Orchestrator not started.");
/// <summary>
/// Number of telemetry items currently queued in the spool that have not yet been
/// acknowledged as uploaded. Returns <c>0</c> before <see cref="Start"/> has wired the
/// spool so callers can poll this safely from the termination path without guarding
/// the orchestrator lifecycle. Used by <c>EnrollmentTerminationHandler.DrainSpool</c>
/// to short-circuit the bounded wait once the spool is actually empty (Option 1 of
/// the WG Part 1 graceful-exit hardening, 2026-04-30).
/// </summary>
public int PendingItemCount => _spool?.PendingItemCount ?? 0;
/// <summary>
/// Number of signals accepted by <c>SignalIngress.Post</c> that have not yet
/// finished reduce + effect-run. Codex Finding 2 (2026-04-30): the termination
/// handler — now dispatched off the ingress worker — uses this to wait until the
/// lifecycle events it just posted (<c>agent_shutting_down</c>,
/// <c>whiteglove_part1_complete</c>, analyzer events) have actually been
/// processed by the worker (and therefore reached the spool) BEFORE polling
/// <see cref="PendingItemCount"/> for spool-empty. Returns <c>0</c> before
/// <see cref="Start"/> wires the ingress.
/// </summary>
public long IngressPendingSignalCount => _ingress?.PendingSignalCount ?? 0L;
// ---------------------------------------------------------------- Lifecycle
/// <summary>
@@ -852,6 +874,17 @@ namespace AutopilotMonitor.Agent.V2.Core.Orchestration
/// outcome derived from the stage (Completed/Part2→Succeeded, Failed→Failed,
/// WhiteGloveSealed→Succeeded but with <see cref="SessionStageExtensions.IsPauseBeforePart2"/>
/// signalled to callers via the stage name — callers decide whether to self-destruct).
/// <para>
/// <b>Codex Finding 2 (2026-04-30)</b> — the <see cref="Terminated"/> handler is
/// dispatched on a thread-pool task instead of synchronously, because this callback
/// fires on the SignalIngress worker thread. A synchronous handler that posts
/// lifecycle events back to the ingress (e.g. <c>agent_shutting_down</c>,
/// <c>whiteglove_part1_complete</c>) would enqueue items the worker can only
/// process AFTER the handler returns — making any in-handler "wait for ingress to
/// drain" impossible (deadlock) and any "wait for spool to drain" trivially
/// returning before the just-posted events are even reduced. Off-worker dispatch
/// frees the worker to keep processing while the handler waits.
/// </para>
/// </summary>
private void OnDecisionTerminalStage(DecisionState terminalState)
{
@@ -870,7 +903,7 @@ namespace AutopilotMonitor.Agent.V2.Core.Orchestration
};
_logger.Info(
$"EnrollmentOrchestrator: decision terminal stage reached (stage={terminalState.Stage}, outcome={outcome}) — firing Terminated(DecisionTerminalStage).");
$"EnrollmentOrchestrator: decision terminal stage reached (stage={terminalState.Stage}, outcome={outcome}) — dispatching Terminated(DecisionTerminalStage) off worker.");
var terminatedArgs = new EnrollmentTerminatedEventArgs(
reason: EnrollmentTerminationReason.DecisionTerminalStage,
@@ -881,8 +914,16 @@ namespace AutopilotMonitor.Agent.V2.Core.Orchestration
? "WhiteGlove Part 1 sealed — session will resume on Part 2 post-reboot; self-destruct suppressed."
: null);
try { Terminated?.Invoke(this, terminatedArgs); }
catch (Exception ex) { _logger.Error("EnrollmentOrchestrator: Terminated handler threw.", ex); }
// Codex Finding 2 — off-worker dispatch. Task.Run unblocks the ingress worker
// immediately; the worker keeps draining the channel (including events the
// termination handler is about to post). The handler's _signalShutdown still
// fires from its finally block on whatever exception path, so an unhandled
// throw inside the handler does NOT prevent the agent from shutting down.
_ = System.Threading.Tasks.Task.Run(() =>
{
try { Terminated?.Invoke(this, terminatedArgs); }
catch (Exception ex) { _logger.Error("EnrollmentOrchestrator: Terminated handler threw on background task.", ex); }
});
}
/// <summary>
@@ -55,6 +55,23 @@ namespace AutopilotMonitor.Agent.V2.Core.Orchestration
private readonly AgentLogger? _logger;
private long _processedCount;
private long _stopRequestedAtTicks;
// Codex review follow-up (Finding 2, 2026-04-30): items posted but not yet fully
// processed by the worker. Incremented in <see cref="Post"/> *before* the channel
// accepts the item (any add-failure path decrements again), decremented in the
// <see cref="ProcessItem"/> finally so every successful enqueue eventually
// accounts for itself. A foreign thread polling <see cref="PendingSignalCount"/>
// is guaranteed: a steady read of 0 means all items have finished both reduce and
// effect — there is no race window where a dequeued-but-not-yet-incremented item
// would be invisible. Used by the termination drain to wait for ingress idle
// before draining the spool.
private long _unprocessedCount;
// Codex follow-up (Low, 2026-04-30) — internal test seam, called between the
// post-stop check and channel.TryAdd inside <see cref="Post"/>. Lets a unit test
// deterministically reproduce the Stop-in-the-gap race that would otherwise
// require a stress run with non-trivial false-negative odds. Always null in
// production; only set from <c>SignalIngressTests</c>.
internal Action? _testHookBetweenPostStopCheckAndAdd;
private readonly BlockingCollection<IngressItem> _channel;
private readonly Dictionary<string, DateTime> _lastBackPressureEmittedUtc =
@@ -108,6 +125,16 @@ namespace AutopilotMonitor.Agent.V2.Core.Orchestration
/// <summary>Aktuelle Queue-Länge. Nur für Observability/Tests — enthält Race-Fenster.</summary>
public int ApproximateQueueLength => _channel.Count;
/// <summary>
/// Total signals accepted by <see cref="Post"/> that have not yet finished
/// <see cref="ProcessItem"/>. Used by the termination drain (which runs off the
/// worker thread) to wait for "ingress fully idle" before draining the telemetry
/// spool — otherwise events the termination handler itself just posted would
/// still be sitting in the channel when the spool-drain check fires and the
/// handler would exit early. Codex Finding 2 (2026-04-30).
/// </summary>
public long PendingSignalCount => Interlocked.Read(ref _unprocessedCount);
/// <summary>
/// Fires synchronously from <see cref="Post"/> after a signal has been accepted into
/// the channel (for both the fast-path <see cref="BlockingCollection{T}.TryAdd"/> and
@@ -216,8 +243,37 @@ namespace AutopilotMonitor.Agent.V2.Core.Orchestration
var item = new IngressItem(kind, kindSchemaVersion, occurredAtUtc, sourceOrigin, evidence, payload, typedPayload);
// Fast path: non-blocking enqueue.
if (_channel.TryAdd(item))
// Codex Finding 2 (2026-04-30): bump the pending-counter BEFORE handing the item
// to the channel. Decrement in three places: (a) when TryAdd throws because
// CompleteAdding raced between the post-stop check and the enqueue, (b) when
// the slow-path Add throws for the same reason, and (c) at the end of
// ProcessItem's finally for items that successfully made it to the worker. The
// increment-before-Add ordering is what eliminates the otherwise-unavoidable
// race where a foreign poller reads PendingSignalCount==0 between
// channel.Take and the worker's "now in flight" bookkeeping.
Interlocked.Increment(ref _unprocessedCount);
// Test hook (Codex Low follow-up, 2026-04-30) — invoked between the post-stop
// check and the channel.TryAdd call so a deterministic test can simulate the
// production race where Stop runs in this exact gap. Always null in production.
_testHookBetweenPostStopCheckAndAdd?.Invoke();
// Fast path: non-blocking enqueue. Codex follow-up (Low, 2026-04-30) — TryAdd
// throws InvalidOperationException (NOT returns false) if CompleteAdding raced
// between our post-stop check above and the enqueue here. Without this catch
// the just-incremented counter would leak. Mirror the slow-path Add handler.
bool added;
try
{
added = _channel.TryAdd(item);
}
catch (InvalidOperationException)
{
Interlocked.Decrement(ref _unprocessedCount);
throw new InvalidOperationException("SignalIngress has been stopped.");
}
if (added)
{
RaiseSignalPosted(kind, payload);
return;
@@ -233,6 +289,7 @@ namespace AutopilotMonitor.Agent.V2.Core.Orchestration
catch (InvalidOperationException)
{
// CompleteAdding wurde zwischen TryAdd und Add gerufen.
Interlocked.Decrement(ref _unprocessedCount);
throw new InvalidOperationException("SignalIngress has been stopped.");
}
@@ -310,6 +367,22 @@ namespace AutopilotMonitor.Agent.V2.Core.Orchestration
}
private void ProcessItem(IngressItem item)
{
// Codex Finding 2 (2026-04-30): match the Interlocked.Increment that Post()
// performed against this same item. Multiple early-return paths below — using
// try/finally guarantees the counter always settles, even on
// DecisionSignal-ctor-rejection or SignalLog.Append-failure paths.
try
{
ProcessItemInner(item);
}
finally
{
Interlocked.Decrement(ref _unprocessedCount);
}
}
private void ProcessItemInner(IngressItem item)
{
// Single-writer: _nextSignalOrdinal wird ausschließlich hier verändert.
long signalOrdinal = _nextSignalOrdinal;
@@ -50,6 +50,11 @@ namespace AutopilotMonitor.Agent.V2.Core.Termination
{
private static readonly TimeSpan DefaultLateEventGrace = TimeSpan.FromMilliseconds(2000);
private static readonly TimeSpan DefaultSpoolDrain = TimeSpan.FromMilliseconds(10000); // V1 parity: up to 20 × 500ms drains before shutdown.exe
// Option 1 (WG Part 1 graceful-exit hardening, 2026-04-30): poll cadence for the
// active spool-empty drain. 50ms is short enough that we exit the bounded wait
// within ~one cadence after the last upload is acknowledged, but long enough that
// an empty spool only costs a single sleep call.
private static readonly TimeSpan SpoolDrainPollInterval = TimeSpan.FromMilliseconds(50);
private readonly AgentConfiguration _configuration;
private readonly AgentLogger _logger;
@@ -79,6 +84,28 @@ namespace AutopilotMonitor.Agent.V2.Core.Termination
private readonly TimeSpan _lateEventGracePeriod;
private readonly TimeSpan _spoolDrainPeriod;
private readonly string _agentVersion;
// Option 1 (WG Part 1 graceful-exit hardening, 2026-04-30): when wired, the
// termination handler polls this accessor every <see cref="SpoolDrainPollInterval"/>
// during <see cref="DrainSpool"/> and exits the wait as soon as it returns 0
// (spool fully acknowledged by the backend). Falls back to the bounded
// <see cref="_spoolDrainPeriod"/> timeout if the spool never drains. Null = legacy
// blind-delay behaviour (kept for tests + safety on the off-chance the orchestrator
// wiring fails to provide it).
private readonly Func<int> _pendingItemCountAccessor;
// Codex Finding 2 (2026-04-30): the termination handler is now dispatched off the
// ingress worker, so it can wait for the worker to actually process the lifecycle
// events the handler posts (agent_shutting_down, whiteglove_part1_complete,
// analyzer events) BEFORE polling the spool — without this the spool-empty check
// would trivially fire on already-uploaded items while the just-posted events are
// still in the ingress channel. Polled in <see cref="DrainSpool"/> as the first
// wait step. Null = legacy behaviour (no ingress-drain wait, just spool-drain).
private readonly Func<long> _ingressPendingSignalCountAccessor;
// Option 2 (same hardening): writes the <c>clean-exit.marker</c> file directly,
// before <see cref="_signalShutdown"/> hands control to the main thread. This wins
// the race against an admin-triggered reseal-reboot — without this hook the marker
// is only written by the AppDomain.ProcessExit handler, which Windows can pre-empt.
// Null = no early write (tests / parity with the original blind-exit path).
private readonly Action _writeCleanExitMarker;
private int _handled;
@@ -102,7 +129,10 @@ namespace AutopilotMonitor.Agent.V2.Core.Termination
Func<IReadOnlyDictionary<string, AppInstallTiming>> appTimingsAccessor = null,
string agentVersion = null,
Action stopPeripheralCollectors = null,
Func<int> ignoredCountAccessor = null)
Func<int> ignoredCountAccessor = null,
Func<int> pendingItemCountAccessor = null,
Action writeCleanExitMarker = null,
Func<long> ingressPendingSignalCountAccessor = null)
{
_configuration = configuration ?? throw new ArgumentNullException(nameof(configuration));
_logger = logger ?? throw new ArgumentNullException(nameof(logger));
@@ -128,6 +158,9 @@ namespace AutopilotMonitor.Agent.V2.Core.Termination
_lateEventGracePeriod = lateEventGracePeriod ?? DefaultLateEventGrace;
_spoolDrainPeriod = spoolDrainPeriod ?? DefaultSpoolDrain;
_agentVersion = agentVersion ?? string.Empty;
_pendingItemCountAccessor = pendingItemCountAccessor;
_writeCleanExitMarker = writeCleanExitMarker;
_ingressPendingSignalCountAccessor = ingressPendingSignalCountAccessor;
}
/// <summary>
@@ -197,6 +230,13 @@ namespace AutopilotMonitor.Agent.V2.Core.Termination
DrainSpool();
TrySaveWhiteGloveComplete();
// Option 2 (WG Part 1 graceful-exit hardening, 2026-04-30): write the
// clean-exit marker BEFORE _signalShutdown returns control to the main
// thread. The AppDomain.ProcessExit handler still writes it as a second
// line of defence, but admin-triggered reseal-reboots routinely pre-empt
// ProcessExit and leave the next run mis-classifying its predecessor as
// reboot_kill. The two writes are idempotent (overwrite-with-timestamp).
TryWriteCleanExitMarker();
return;
}
@@ -213,6 +253,10 @@ namespace AutopilotMonitor.Agent.V2.Core.Termination
WriteEnrollmentCompleteMarker(args);
RunStandaloneRebootIfRequested();
RunSelfDestructIfAppropriate(args);
// Option 2 (same hardening) — covers the standard Completed / Failed terminal
// path too. Cleanup PowerShell does not touch this marker, so writing it before
// ExecuteSelfDestruct is harmless.
TryWriteCleanExitMarker();
}
catch (Exception ex)
{
@@ -594,6 +638,28 @@ namespace AutopilotMonitor.Agent.V2.Core.Termination
catch (Exception ex) { _logger.Warning($"EnrollmentTerminationHandler: SaveWhiteGloveComplete threw: {ex.Message}"); }
}
/// <summary>
/// Option 2 helper — writes the <c>clean-exit.marker</c> early via the host-supplied
/// hook so the next agent run classifies <c>previousExit=clean</c> instead of
/// <c>reboot_kill</c>, even when Windows kills the process between
/// <see cref="_signalShutdown"/> and the legacy <c>AppDomain.ProcessExit</c> handler
/// (typical race in admin-triggered reseal-reboots). No-op when the host did not
/// wire the accessor (legacy call sites + tests).
/// </summary>
private void TryWriteCleanExitMarker()
{
if (_writeCleanExitMarker == null) return;
try
{
_writeCleanExitMarker();
_logger.Debug("EnrollmentTerminationHandler: clean-exit marker written (early).");
}
catch (Exception ex)
{
_logger.Warning($"EnrollmentTerminationHandler: early clean-exit marker write threw: {ex.Message}");
}
}
private void DelayLateEventGrace()
{
if (_lateEventGracePeriod <= TimeSpan.Zero) return;
@@ -617,12 +683,92 @@ namespace AutopilotMonitor.Agent.V2.Core.Termination
private void DrainSpool()
{
// V1 parity — block briefly so pending events can land before the next destructive
// step (shutdown.exe, self-destruct). A precise HasPending/PendingCount signal is
// internal to the orchestrator transport, so the V2 equivalent is a bounded wait.
// Block briefly so pending events can land before the next destructive step
// (shutdown.exe, self-destruct). Two phases share the same bounded budget:
//
// Phase A — wait for the SignalIngress queue to fully process. The handler
// itself just posted lifecycle events (agent_shutting_down,
// whiteglove_part1_complete, analyzer events) that the worker only sees
// after we yielded — without this wait the spool-poll would trivially see
// "pending=0" because those events haven't even been reduced + spooled yet.
//
// Phase B — wait for the spool to acknowledge upload to the backend (the
// pre-Codex-Finding-2 behaviour, kept).
//
// When neither accessor is wired (tests / paranoid fallback), we keep V1
// parity and just sleep the bounded period.
if (_spoolDrainPeriod <= TimeSpan.Zero) return;
try { Task.Delay(_spoolDrainPeriod).Wait(); }
catch { /* best-effort */ }
if (_ingressPendingSignalCountAccessor == null && _pendingItemCountAccessor == null)
{
try { Task.Delay(_spoolDrainPeriod).Wait(); }
catch { /* best-effort */ }
return;
}
var deadline = DateTime.UtcNow + _spoolDrainPeriod;
if (_ingressPendingSignalCountAccessor != null)
{
WaitFor(
label: "ingress",
pollAccessor: () => _ingressPendingSignalCountAccessor(),
deadline: deadline);
}
if (_pendingItemCountAccessor != null)
{
WaitFor(
label: "spool",
pollAccessor: () => (long)_pendingItemCountAccessor(),
deadline: deadline);
}
}
/// <summary>
/// Bounded polling helper used by <see cref="DrainSpool"/>. Returns when either
/// <paramref name="pollAccessor"/> reports zero or <paramref name="deadline"/>
/// has passed. Accessor exceptions degrade to a single bounded sleep — never throw
/// out of the termination path.
/// </summary>
private void WaitFor(string label, Func<long> pollAccessor, DateTime deadline)
{
while (DateTime.UtcNow < deadline)
{
long pending;
try { pending = pollAccessor(); }
catch (Exception ex)
{
_logger.Warning($"EnrollmentTerminationHandler: {label} accessor threw: {ex.Message}");
var fallback = deadline - DateTime.UtcNow;
if (fallback > TimeSpan.Zero)
{
try { Task.Delay(fallback).Wait(); }
catch { /* best-effort */ }
}
return;
}
if (pending <= 0)
{
_logger.Debug($"EnrollmentTerminationHandler: {label} drained (pending=0) — moving on.");
return;
}
var remaining = deadline - DateTime.UtcNow;
if (remaining <= TimeSpan.Zero) break;
var sleep = remaining < SpoolDrainPollInterval ? remaining : SpoolDrainPollInterval;
try { Task.Delay(sleep).Wait(); }
catch { return; }
}
try
{
var pendingAtTimeout = pollAccessor();
_logger.Warning(
$"EnrollmentTerminationHandler: {label} drain budget exhausted (pending={pendingAtTimeout}).");
}
catch { /* logging is best-effort */ }
}
private void EmitEventSafe(EnrollmentEvent evt)
@@ -59,6 +59,25 @@ namespace AutopilotMonitor.Agent.V2
// ----------------------------------------------------------------- clean-exit marker
/// <summary>
/// Writes the <c>clean-exit.marker</c> file directly. Called both from the
/// <see cref="AppDomain.ProcessExit"/> handler registered by
/// <see cref="RegisterCleanExitMarker"/> AND — explicitly, before
/// <c>EnrollmentTerminationHandler._signalShutdown</c> returns — from the V2 graceful
/// shutdown sequence so admin-triggered reseal-reboots can't pre-empt the marker.
/// Best-effort: I/O exceptions are swallowed (the marker is observability, not load-bearing).
/// </summary>
public static void WriteCleanExitMarker(string dataDirectory)
{
if (string.IsNullOrEmpty(dataDirectory)) return;
try
{
var markerPath = Path.Combine(dataDirectory, CleanExitMarkerFileName);
File.WriteAllText(markerPath, DateTime.UtcNow.ToString("O"));
}
catch { /* best-effort */ }
}
/// <summary>
/// Registers a <see cref="AppDomain.ProcessExit"/> handler that writes <c>clean-exit.marker</c>
/// as a last action so the next start can distinguish clean shutdown from hard kill.
@@ -66,11 +85,7 @@ namespace AutopilotMonitor.Agent.V2
/// </summary>
internal static EventHandler RegisterCleanExitMarker(string dataDirectory)
{
var markerPath = Path.Combine(dataDirectory, CleanExitMarkerFileName);
EventHandler handler = (_, _) =>
{
try { File.WriteAllText(markerPath, DateTime.UtcNow.ToString("O")); } catch { }
};
EventHandler handler = (_, _) => WriteCleanExitMarker(dataDirectory);
AppDomain.CurrentDomain.ProcessExit += handler;
return handler;
}
@@ -227,6 +227,7 @@ namespace AutopilotMonitor.Agent.V2
runtimeConfig: runtimeConfig,
telemetry: telemetry,
registration: registration,
dataDirectory: dataDirectory,
stateSubdir: stateSubdir,
transportDir: transportDir,
agentVersion: GetAgentVersion(),
@@ -47,6 +47,7 @@ namespace AutopilotMonitor.Agent.V2.Runtime
RuntimeConfigBundle runtimeConfig,
TelemetryClientResult telemetry,
SessionRegistrationOutcomeResult registration,
string dataDirectory,
string stateSubdir,
string transportDir,
string agentVersion,
@@ -252,7 +253,26 @@ namespace AutopilotMonitor.Agent.V2.Runtime
stopPeripheralCollectors: () => orchestrator.StopCollectorHosts(),
// V1-parity ignoredCount for app_tracking_summary — lives on the
// live AppPackageStateList only (phase snapshots don't carry it).
ignoredCountAccessor: () => componentFactory.ImeIgnoredCount);
ignoredCountAccessor: () => componentFactory.ImeIgnoredCount,
// Option 1 (WG Part 1 graceful-exit hardening, 2026-04-30) —
// active spool-empty polling so DrainSpool exits as soon as the
// backend ack'd the last lifecycle event instead of waiting the
// full 10s timeout. Critical when an admin reseal-reboot races
// termination.
pendingItemCountAccessor: () => orchestrator.PendingItemCount,
// Option 2 (same hardening) — write clean-exit.marker before
// _signalShutdown returns, instead of relying solely on
// AppDomain.ProcessExit which Windows can pre-empt during a
// shutdown.exe / reseal-triggered reboot.
writeCleanExitMarker: () => Program.WriteCleanExitMarker(dataDirectory),
// Codex Finding 2 (2026-04-30) — DrainSpool needs to wait
// for the ingress to finish processing the lifecycle events
// the handler just posted (agent_shutting_down,
// whiteglove_part1_complete, analyzer events) BEFORE polling
// spool-empty. Pair with off-worker dispatch in
// EnrollmentOrchestrator.OnDecisionTerminalStage so the
// wait can actually make progress.
ingressPendingSignalCountAccessor: () => orchestrator.IngressPendingSignalCount);
// ServerActionDispatcher (plan §5.3) — constructed inside this
// hook so lifecyclePost + terminationHandler are guaranteed
@@ -1 +1 @@
477
498
@@ -1 +1 @@
1222
1226
@@ -195,12 +195,20 @@ namespace AutopilotMonitor.DecisionCore.Tests
[Fact]
public void WhiteGloveComplete_carries_audit_trail_with_classifier_verdict_and_inputs()
{
// NOTE: this test exercises the slow-path verdict-applier (HandleClassifierVerdictIssuedV1).
// The newer fast-path (introduced 2026-04-30, see WhiteGloveCompleteFastPath_*
// test below) handles the typical WG Part 1 sequence and emits the
// whiteglove_complete event with its own audit-trail trigger
// ("WhiteGloveShellCoreSuccess:FastPath:Confirmed"). Both paths must produce a
// structurally identical TypedPayload; the fast-path test guards that contract.
var engine = new DecisionEngine();
var state = DecisionState.CreateInitial("sess-wg", "tenant-wg");
// Drive through SessionStarted so the StepIndex / LastAppliedSignalOrdinal are real.
state = engine.Reduce(state, MakeSignal(0, DecisionSignalKind.SessionStarted, T0)).NewState;
// Observe ShellCore success — the inline-WG fast path: a single high-confidence signal
// is enough for the classifier to seal at 80/100.
// Observe ShellCore success — fast-path inline-classifies and seals immediately.
// We then synthesize a ClassifierVerdictIssued signal so the slow-path verdict
// applier ALSO runs (idempotent on already-sealed state) and we can assert on
// its TypedPayload independently of the fast-path emission.
var shellSignal = MakeSignal(1, DecisionSignalKind.WhiteGloveShellCoreSuccess, T0.AddMinutes(5));
state = engine.Reduce(state, shellSignal).NewState;
@@ -243,6 +251,57 @@ namespace AutopilotMonitor.DecisionCore.Tests
Assert.Equal(EnrollmentMode.WhiteGlove.ToString(), scenario["mode"]);
}
[Fact]
public void WhiteGloveCompleteFastPath_attaches_full_audit_trail_with_inline_verdict()
{
// Codex review follow-up (2026-04-30): the fast-path in
// HandleWhiteGloveShellCoreSuccessV1 must produce a TypedPayload byte-stable to
// the slow-path emission. Earlier impl only set parameters[eventType] and left
// TypedPayload empty — this test pins the contract so a regression is caught at
// the engine boundary.
var engine = new DecisionEngine();
var state = DecisionState.CreateInitial("sess-wg-fp", "tenant-wg-fp");
state = engine.Reduce(state, MakeSignal(0, DecisionSignalKind.SessionStarted, T0)).NewState;
var shellSignal = MakeSignal(1, DecisionSignalKind.WhiteGloveShellCoreSuccess, T0.AddMinutes(5));
var step = engine.Reduce(state, shellSignal);
// Fast-path → terminal stage in this single reduce call.
Assert.Equal(SessionStage.WhiteGloveSealed, step.NewState.Stage);
var effect = SingleTimelineEffect(step, "whiteglove_complete");
var payload = Assert.IsType<Dictionary<string, object>>(effect.TypedPayload);
// Mandatory anchors — same set the slow-path emits, with the fast-path trigger.
Assert.Equal("DecisionEngine", payload["decisionSource"]);
Assert.Equal(
$"{nameof(DecisionSignalKind.WhiteGloveShellCoreSuccess)}:FastPath:Confirmed",
payload["trigger"]);
Assert.Equal(nameof(SessionStage.WhiteGloveSealed), payload["sessionStage"]);
// Census fields — at minimum the ShellCore signal that drove the decision.
var signalsSeen = Assert.IsType<List<string>>(payload["signalsSeen"]);
Assert.Contains("shellcore_whiteglove_success", signalsSeen);
// Classifier verdict reflects the inline-computed result — Score=80 (just
// ShellCore in a clean state), Confirmed, real (non-empty) input hash so the
// EffectRunner's anti-loop would skip a racing RunClassifier effect.
var classifier = Assert.IsType<Dictionary<string, object>>(payload["classifier"]);
Assert.Equal(WhiteGloveSealingClassifier.ClassifierId, classifier["id"]);
Assert.Equal("Confirmed", classifier["level"]);
Assert.Equal(80, classifier["score"]);
Assert.False(string.IsNullOrEmpty((string)classifier["inputHash"]));
// Classifier-input snapshot must be present (flattened) — this is the field a
// reviewer or backend forensics expects when triaging an unexpected WG sealing.
var inputs = Assert.IsType<Dictionary<string, object>>(payload["classifierInputs"]);
Assert.Equal(true, inputs["shellCoreWhiteGloveSuccessSeen"]);
Assert.Equal(false, inputs["aadJoinedWithUser"]);
var scenario = Assert.IsType<Dictionary<string, object>>(payload["scenario"]);
Assert.Equal(EnrollmentMode.WhiteGlove.ToString(), scenario["mode"]);
}
// ============================================================ whiteglove_resumed
[Fact]
@@ -54,12 +54,16 @@ namespace AutopilotMonitor.DecisionCore.Tests.Scenarios
Assert.Equal(HypothesisLevel.Confirmed, result.FinalState.ClassifierOutcomes.WhiteGloveSealing.Level);
Assert.True(result.FinalState.ClassifierOutcomes.WhiteGloveSealing.Score >= WhiteGloveSealingClassifier.HighThreshold);
// The tick fired between the two WG signals — snapshot unchanged at that point,
// so it was anti-loop-skipped. The SealingPattern run and the ShellCore run both
// produced fresh verdicts.
// Option 3 fast-path (WG Part 1 graceful-exit hardening, 2026-04-30): the
// ShellCoreSuccess handler now classifies inline and transitions to
// WhiteGloveSealed in a single reducer step, so it does NOT emit a RunClassifier
// effect. Only the initial SealingPattern arrival emits RunClassifier (1 run);
// the ClassifierTick that fires between the two WG signals re-evaluates the
// unchanged snapshot and is anti-loop-skipped (1 skip). The second classifier
// verdict is computed inline by the engine — invisible to the harness counters.
var runs = harness.ClassifierRunStats.TryGetValue("whiteglove-sealing:run", out var r) ? r : 0;
var skipped = harness.ClassifierRunStats.TryGetValue("whiteglove-sealing:skipped_by_antiloop", out var s) ? s : 0;
Assert.Equal(2, runs);
Assert.Equal(1, runs);
Assert.Equal(1, skipped);
}
@@ -102,5 +106,50 @@ namespace AutopilotMonitor.DecisionCore.Tests.Scenarios
Assert.Equal(1, runs);
Assert.Equal(2, skipped);
}
// ============================================================ Option 3 — fast-path
// (WG Part 1 graceful-exit hardening, 2026-04-30). The strong WG signal
// (ShellCoreSuccess) now classifies inline and transitions to WhiteGloveSealed in a
// single reducer step; no RunClassifier effect is emitted on the fast path.
[Fact]
public void FastPath_shellCoreSuccess_emits_no_runClassifier_effect()
{
var harness = NewHarness();
var signals = LoadFixture("whiteglove-inline-v1.jsonl");
var result = harness.Replay("session-anon-0014-fp1", "tenant-anon-0014", signals);
Assert.Equal(SessionStage.WhiteGloveSealed, result.FinalState.Stage);
Assert.Equal(SessionOutcome.WhiteGlovePart1Sealed, result.FinalState.Outcome);
// The fast-path classifies inline — no RunClassifier effect should have been
// posted, so the harness's run/skip counters stay at zero. (Compare against
// the slow-path correlated test which records 1 run + 1 skip.)
var runs = harness.ClassifierRunStats.TryGetValue("whiteglove-sealing:run", out var r) ? r : 0;
var skipped = harness.ClassifierRunStats.TryGetValue("whiteglove-sealing:skipped_by_antiloop", out var s) ? s : 0;
Assert.Equal(0, runs);
Assert.Equal(0, skipped);
// Sanity — no ClassifierTick deadline left armed when the fast-path transitions
// straight to terminal.
Assert.Empty(result.FinalState.Deadlines);
}
[Fact]
public void FastPath_records_inline_classifier_outcome_on_state()
{
var harness = NewHarness();
var signals = LoadFixture("whiteglove-inline-v1.jsonl");
var result = harness.Replay("session-anon-0014-fp2", "tenant-anon-0014", signals);
// Inline-computed verdict must be persisted on ClassifierOutcomes — same shape
// as the slow-path verdict so Inspector + downstream consumers see no diff.
Assert.Equal(HypothesisLevel.Confirmed, result.FinalState.ClassifierOutcomes.WhiteGloveSealing.Level);
Assert.True(result.FinalState.ClassifierOutcomes.WhiteGloveSealing.Score
>= WhiteGloveSealingClassifier.HighThreshold);
// Verdict-input-hash must be set so any racing RunClassifier effect would be
// anti-loop-skipped against the inline result.
Assert.False(string.IsNullOrEmpty(result.FinalState.ClassifierOutcomes.WhiteGloveSealing.LastClassifierVerdictId));
}
}
}
@@ -0,0 +1,123 @@
using System;
using System.Collections.Generic;
using System.Linq;
using AutopilotMonitor.DecisionCore.Classifiers;
using AutopilotMonitor.DecisionCore.Engine;
using AutopilotMonitor.DecisionCore.Signals;
using AutopilotMonitor.DecisionCore.State;
using Xunit;
namespace AutopilotMonitor.DecisionCore.Tests
{
/// <summary>
/// Option 3 (WG Part 1 graceful-exit hardening, 2026-04-30) — covers the slow-path
/// fallback in <see cref="DecisionEngine"/>.HandleWhiteGloveShellCoreSuccessV1.
/// <para>
/// The fast-path is exercised by the scenario-replay tests in
/// <c>WhiteGlovePart1ScenarioTests</c>. Here we verify that when an excluding
/// observation already lives on state (AccountSetup activity), the inline classifier
/// scores below <see cref="WhiteGloveSealingClassifier.HighThreshold"/> and the
/// engine emits the legacy <see cref="DecisionEffectKind.RunClassifier"/> +
/// <see cref="DecisionEffectKind.ScheduleDeadline"/>(<see cref="DeadlineNames.ClassifierTick"/>)
/// effects instead of transitioning to <see cref="SessionStage.WhiteGloveSealed"/>.
/// </para>
/// </summary>
public sealed class WhiteGloveFastPathFallbackTests
{
private static readonly DateTime T0 = new DateTime(2026, 4, 30, 8, 45, 0, DateTimeKind.Utc);
[Fact]
public void ShellCoreSuccess_with_account_setup_activity_falls_back_to_slow_path()
{
var engine = new DecisionEngine();
// Pre-seed AccountSetupEnteredUtc — once the AccountSetup ESP phase fires the
// WG sealing classifier's HasAccountSetupActivity excluder kicks in (-40),
// pushing ShellCore-alone (+80) to score 40 (Weak), below HighThreshold (70).
var state = DecisionState.CreateInitial("s-fb", "t-fb").ToBuilder()
.WithStage(SessionStage.EspAccountSetup)
.WithStepIndex(3)
.WithLastAppliedSignalOrdinal(2);
state.AccountSetupEnteredUtc = new SignalFact<DateTime>(T0.AddMinutes(-2), 2);
var shellCore = MakeShellCoreSignal(ordinal: 3);
var step = engine.Reduce(state.Build(), shellCore);
// Slow path — stage unchanged, no WhiteGloveSealed transition.
Assert.NotEqual(SessionStage.WhiteGloveSealed, step.NewState.Stage);
Assert.Equal(SessionStage.EspAccountSetup, step.NewState.Stage);
// The observation IS still recorded on the new state — slow path is about
// *how* we get to a verdict, not whether we record the signal.
Assert.NotNull(step.NewState.ScenarioObservations.ShellCoreWhiteGloveSuccessSeen);
Assert.True(step.NewState.ScenarioObservations.ShellCoreWhiteGloveSuccessSeen!.Value);
// Slow path emits the RunClassifier effect (the EffectRunner will execute the
// classifier, post ClassifierVerdictIssued, and the verdict-applier handles
// the asymmetric-conservative rejection at HypothesisLevel.Weak).
Assert.Contains(step.Effects, e => e.Kind == DecisionEffectKind.RunClassifier);
// And arms the 30s ClassifierTick deadline so a later state change (e.g. an
// AAD-sign-in invalidating the WG hypothesis) gets re-evaluated.
Assert.Contains(step.Effects, e =>
e.Kind == DecisionEffectKind.ScheduleDeadline &&
e.Deadline?.Name == DeadlineNames.ClassifierTick);
// Crucially, the fast-path's WhiteGloveSealing outcome is NOT recorded — that
// path only writes a verdict on the inline-Confirmed branch.
Assert.Equal(HypothesisLevel.Unknown,
step.NewState.ClassifierOutcomes.WhiteGloveSealing.Level);
}
[Fact]
public void ShellCoreSuccess_in_clean_state_takes_fast_path_in_single_step()
{
var engine = new DecisionEngine();
// Clean state — no excluders. ShellCoreSuccess alone should hit Confirmed
// inline and seal in this single Reduce call.
var state = DecisionState.CreateInitial("s-fp", "t-fp").ToBuilder()
.WithStage(SessionStage.EspDeviceSetup)
.WithStepIndex(1)
.WithLastAppliedSignalOrdinal(0)
.Build();
var shellCore = MakeShellCoreSignal(ordinal: 1);
var step = engine.Reduce(state, shellCore);
// Fast path — terminal stage in one step.
Assert.Equal(SessionStage.WhiteGloveSealed, step.NewState.Stage);
Assert.Equal(SessionOutcome.WhiteGlovePart1Sealed, step.NewState.Outcome);
Assert.Equal(HypothesisLevel.Confirmed,
step.NewState.ClassifierOutcomes.WhiteGloveSealing.Level);
// No RunClassifier / ClassifierTick effects — the engine classified inline.
Assert.DoesNotContain(step.Effects, e => e.Kind == DecisionEffectKind.RunClassifier);
Assert.DoesNotContain(step.Effects, e =>
e.Kind == DecisionEffectKind.ScheduleDeadline &&
e.Deadline?.Name == DeadlineNames.ClassifierTick);
// Single timeline-emit effect: the whiteglove_complete event.
var emit = Assert.Single(step.Effects, e => e.Kind == DecisionEffectKind.EmitEventTimelineEntry);
Assert.NotNull(emit.Parameters);
Assert.Equal("whiteglove_complete", emit.Parameters!["eventType"]);
// No deadlines left armed — terminal stage clears them.
Assert.Empty(step.NewState.Deadlines);
}
private static DecisionSignal MakeShellCoreSignal(long ordinal) =>
new DecisionSignal(
sessionSignalOrdinal: ordinal,
sessionTraceOrdinal: ordinal,
kind: DecisionSignalKind.WhiteGloveShellCoreSuccess,
kindSchemaVersion: 1,
occurredAtUtc: T0,
sourceOrigin: "ShellCoreTracker",
evidence: new Evidence(
kind: EvidenceKind.Raw,
identifier: "ShellCore-62407",
summary: "WhiteGlove_Success"),
payload: new Dictionary<string, string>());
}
}
@@ -13,12 +13,36 @@ namespace AutopilotMonitor.DecisionCore.Engine
// Plan §2.6 — ClassifierTick recurrence. 30 s is the legacy polling cadence.
internal static readonly TimeSpan s_classifierTickInterval = TimeSpan.FromSeconds(30);
// Option 3 (WG Part 1 graceful-exit hardening, 2026-04-30): inline classifier
// instance, used only by the strong-signal fast-path in
// <see cref="HandleWhiteGloveShellCoreSuccessV1"/>. Stateless / pure — safe to share.
private static readonly WhiteGloveSealingClassifier s_whiteGloveSealingClassifier =
new WhiteGloveSealingClassifier();
/// <summary>
/// Handle <see cref="DecisionSignalKind.WhiteGloveShellCoreSuccess"/>. Records the
/// <see cref="DecisionState.ShellCoreWhiteGloveSuccessSeen"/> fact and emits a
/// <see cref="DecisionEffectKind.RunClassifier"/> effect for the WhiteGlove-sealing
/// classifier. Also arms the <see cref="DeadlineNames.ClassifierTick"/> deadline
/// if not already armed.
/// <see cref="DecisionState.ShellCoreWhiteGloveSuccessSeen"/> fact.
/// <para>
/// <b>Fast-path</b> (Option 3 of the WG Part 1 graceful-exit hardening, 2026-04-30):
/// the WG-sealing classifier is a pure function of the snapshot built from this state,
/// so we can run it inline. When ShellCoreSuccess arrives in a state that scores
/// <see cref="HypothesisLevel.Confirmed"/> on its own (the typical WG Part 1 case —
/// no AAD-with-user / desktop / hello observed yet), we transition straight to
/// <see cref="SessionStage.WhiteGloveSealed"/> + <see cref="SessionOutcome.WhiteGlovePart1Sealed"/>
/// in this single reducer step and emit <c>whiteglove_complete</c>. This collapses
/// the previous two-step round-trip (RunClassifier effect → ClassifierVerdictIssued
/// signal → reducer applies verdict) into one, eliminating the journal+snapshot
/// write between the two, which buys us tens to hundreds of milliseconds before the
/// admin-triggered reseal-reboot pre-empts the agent.
/// </para>
/// <para>
/// <b>Slow-path fallback</b>: when an excluding observation is already present (e.g.
/// <c>AadUserJoinWithUserObserved</c>) the inline classifier scores below
/// <see cref="WhiteGloveSealingClassifier.HighThreshold"/>, so we keep the legacy
/// behaviour — emit a <see cref="DecisionEffectKind.RunClassifier"/> effect + arm
/// the <see cref="DeadlineNames.ClassifierTick"/> deadline. The asymmetric-
/// conservative semantics still apply: only Confirmed transitions sealed.
/// </para>
/// </summary>
private DecisionStep HandleWhiteGloveShellCoreSuccessV1(DecisionState state, DecisionSignal signal)
{
@@ -28,7 +52,83 @@ namespace AutopilotMonitor.DecisionCore.Engine
.WithLastAppliedSignalOrdinal(signal.SessionSignalOrdinal);
builder.ScenarioObservations = builder.ScenarioObservations.WithShellCoreWhiteGloveSuccessSeen(signal.SessionSignalOrdinal);
var (newState, effects) = AttachWhiteGloveClassifierEffects(builder.Build(), signal);
// Option 3 fast-path: inline classifier evaluation on the strong WG signal.
var afterObservation = builder.Build();
var inlineSnapshot = BuildWhiteGloveSealingSnapshot(afterObservation);
var inlineVerdict = s_whiteGloveSealingClassifier.Classify(inlineSnapshot);
if (inlineVerdict.Level == HypothesisLevel.Confirmed)
{
var sealedBuilder = afterObservation.ToBuilder();
// Mirror the WG decision in the profile: Mode=WhiteGlove @ High confidence —
// identical to the slow-path branch in HandleClassifierVerdictIssuedV1 so
// downstream consumers see the same ScenarioProfile regardless of which
// path produced the verdict.
sealedBuilder.ScenarioProfile = EnrollmentScenarioProfileUpdater.ApplyWhiteGloveSealingConfirmed(
sealedBuilder.ScenarioProfile, signal);
// Record the inline verdict on the classifier outcome — same shape as the
// slow path. The verdict's InputHash also seeds the EffectRunner anti-loop
// table (via ClassifierVerdictLookup) so any RunClassifier effect that might
// still be in-flight (it shouldn't be, since we don't emit one here) would
// skip with snapshotHash unchanged.
var inlineSealing = afterObservation.ClassifierOutcomes.WhiteGloveSealing.With(
level: HypothesisLevel.Confirmed,
reason: inlineVerdict.Reason,
score: inlineVerdict.Score,
lastUpdatedUtc: signal.OccurredAtUtc,
lastClassifierVerdictId: inlineVerdict.InputHash);
sealedBuilder.ClassifierOutcomes = afterObservation.ClassifierOutcomes.WithWhiteGloveSealing(inlineSealing);
sealedBuilder
.WithStage(SessionStage.WhiteGloveSealed)
.WithOutcome(SessionOutcome.WhiteGlovePart1Sealed)
.ClearDeadlines();
var sealedState = sealedBuilder.Build();
var fastPathTrigger = $"{nameof(DecisionSignalKind.WhiteGloveShellCoreSuccess)}:FastPath:Confirmed";
var sealedTransition = BuildTakenTransition(
before: state,
signal: signal,
toStage: SessionStage.WhiteGloveSealed,
nextStepIndex: nextStep,
trigger: fastPathTrigger);
// Codex review follow-up (Finding 1, 2026-04-30) — the slow path attaches the
// full DecisionAuditTrailBuilder.Build(...) payload (decisionSource, trigger,
// sessionStage, signalsSeen, signalEvidence, scenario, classifier verdict,
// classifierInputs). Without this the new normal WG Part 1 path emits a
// degraded audit trail compared to the legacy two-step path. We replicate
// the exact same build here using the inline verdict so the timeline event
// (and any downstream forensics that rely on TypedPayload) are byte-stable
// across the fast/slow paths.
var inlineVerdictInfo = new ClassifierVerdictInfo(
id: inlineVerdict.ClassifierId,
level: inlineVerdict.Level.ToString(),
score: inlineVerdict.Score,
reason: inlineVerdict.Reason,
inputHash: inlineVerdict.InputHash);
var sealedEffects = new[]
{
new DecisionEffect(
DecisionEffectKind.EmitEventTimelineEntry,
parameters: new Dictionary<string, string> { ["eventType"] = "whiteglove_complete" },
typedPayload: DecisionAuditTrailBuilder.Build(
postState: sealedState,
decidedStage: SessionStage.WhiteGloveSealed,
trigger: fastPathTrigger,
classifier: inlineVerdictInfo,
classifierInputs: inlineSnapshot)),
};
return new DecisionStep(sealedState, sealedTransition, sealedEffects);
}
// Slow path — legacy behaviour, preserved as fallback when an excluding signal
// already lives in state and pulls the inline score below HighThreshold.
var (newState, effects) = AttachWhiteGloveClassifierEffects(afterObservation, signal);
var transition = BuildTakenTransition(
before: state,