Merge pull request #54 from okieselbach/v2/fix-deadline-keys-and-immediate-flush

fix(v2): first-session regressions — deadline payload key + immediate-flush wakeup + willlaunch noise

src/Agent/AutopilotMonitor.Agent.V2.Core.Tests/Integration/ScenarioTests.cs

@@ -1,8 +1,13 @@
+using System;
 using System.Linq;
+using System.Threading;
+using AutopilotMonitor.Agent.V2.Core.Transport.Telemetry;
 using AutopilotMonitor.DecisionCore.Classifiers;
 using AutopilotMonitor.DecisionCore.State;
 using Xunit;
 
+#pragma warning disable xUnit1031 // SpinWait.SpinUntil for uploader-batch-arrival assertion
+
 namespace AutopilotMonitor.Agent.V2.Core.Tests.Integration
 {
     /// <summary>
@@ -254,17 +259,32 @@ namespace AutopilotMonitor.Agent.V2.Core.Tests.Integration
         public void Full_pipeline_enqueues_telemetry_items_for_terminal_events()
         {
             // Verifies that beyond state-shape, the EffectRunner emitted at least one
-            // EmitEventTimelineEntry effect (e.g. enrollment_complete) that reached the Spool.
+            // EmitEventTimelineEntry effect (e.g. enrollment_complete) that flowed through
+            // the Spool into the uploader. Before the P1 "immediate-flush wakeup" fix this
+            // test inspected pending items in the spool directly — but now the drain loop
+            // wakes up on each RequiresImmediateFlush=true enqueue and uploads the item
+            // immediately, so terminal enrollment_complete no longer sits in the pending
+            // queue. The uploader's Received batches are the authoritative "left the agent"
+            // signal and are stable across both pre-fix and post-fix behavior.
             using var f = new EnrollmentOrchestratorFixture();
             f.Start();
             f.PostFixture("userdriven-happy-v1.jsonl");
 
             Assert.True(f.WaitForStage(DefaultTerminalTimeoutMs, SessionStage.Completed));
 
-            var eventItems = f.AllEventItemsInSpool();
-            Assert.NotEmpty(eventItems);
-            // At least one item should be a terminal enrollment_complete event.
-            Assert.Contains(eventItems, i => i.PayloadJson.Contains("enrollment_complete"));
+            // Give the drain loop time to finish the post-completion flush — this test runs
+            // in parallel with other orchestrator tests and the ThreadPool can be saturated,
+            // so use a generous timeout. Stop() below also performs a terminal drain, but we
+            // want to see the wakeup-driven upload here explicitly because that is the P1
+            // behavior under test (EnrollmentOrchestrator.OnImmediateFlushRequested).
+            Assert.True(
+                SpinWait.SpinUntil(
+                    () => f.Uploader.Received
+                        .SelectMany(batch => batch)
+                        .Any(i => i.Kind == TelemetryItemKind.Event
+                                  && i.PayloadJson.Contains("enrollment_complete")),
+                    10000),
+                "Expected enrollment_complete event to reach the uploader after Stage=Completed.");
 
             f.Stop();
         }

src/Agent/AutopilotMonitor.Agent.V2.Core.Tests/Monitoring/HelloTrackerImmediateUploadTests.cs

@@ -16,9 +16,14 @@ namespace AutopilotMonitor.Agent.V2.Core.Tests.Monitoring
     /// policy-level Hello event that must reach the backend within seconds
     /// (<c>hello_policy_detected</c>) and on terminal events
     /// (<c>hello_provisioning_completed/failed/blocked</c>, <c>hello_skipped</c>); and stays
-    /// false on snapshot-type events that can flip multiple times
-    /// (<c>hello_provisioning_willlaunch</c>, <c>hello_provisioning_willnotlaunch</c>,
-    /// <c>hello_pin_status</c>) so transient re-registrations don't trigger a flush each time.
+    /// false on the snapshot-type <c>hello_pin_status</c> ticker so transient re-registrations
+    /// don't trigger a flush each time.
+    /// <para>
+    /// The <c>willlaunch</c> / <c>willnotlaunch</c> snapshots (EventID 358/360) are no longer
+    /// emitted at all — they flip multiple times per session, create pure timeline noise, and
+    /// are documented as non-evidence in <c>project_hello_willlaunch_unreliable</c>. Suppression
+    /// is asserted below.
+    /// </para>
     /// </summary>
     public sealed class HelloTrackerImmediateUploadTests
     {
@@ -69,18 +74,22 @@ namespace AutopilotMonitor.Agent.V2.Core.Tests.Monitoring
         }
 
         [Fact]
-        public void ProcessHelloEvent_358_willlaunch_stays_batched()
+        public void ProcessHelloEvent_358_willlaunch_is_suppressed()
         {
             using var f = new Fixture();
 
-            // Event 358 = ProvisioningWillLaunch. Snapshot-only prerequisites-passed flag
-            // (project_hello_willlaunch_unreliable): flips between willlaunch/willnotlaunch
-            // multiple times per session and is not decision-relevant. Keep batched so each
-            // flip doesn't trigger a flush.
+            // Event 358 = ProvisioningWillLaunch. Snapshot-only prerequisites-passed flag per
+            // project_hello_willlaunch_unreliable — flips multiple times per session (session
+            // 9ed7021e saw 6× 358, three within 232 ms) and is never decision-relevant. The
+            // tracker now suppresses the backend event entirely; only a DEBUG log line remains
+            // so diagnostics can still reconstruct the sequence.
             f.Tracker.ProcessHelloEvent(358, Fixed, providerName: "prov", isBackfill: false);
 
-            var info = f.InfoEvent("hello_provisioning_willlaunch");
-            Assert.Equal("false", info.Payload![SignalPayloadKeys.ImmediateUpload]);
+            Assert.DoesNotContain(f.Ingress.Posted, p =>
+                p.Kind == DecisionSignalKind.InformationalEvent
+                && p.Payload != null
+                && p.Payload.TryGetValue(SignalPayloadKeys.EventType, out var et)
+                && et == "hello_provisioning_willlaunch");
         }
 
         [Fact]
@@ -108,17 +117,19 @@ namespace AutopilotMonitor.Agent.V2.Core.Tests.Monitoring
         }
 
         [Fact]
-        public void ProcessHelloEvent_360_willnotlaunch_stays_batched()
+        public void ProcessHelloEvent_360_willnotlaunch_is_suppressed()
         {
             using var f = new Fixture();
 
-            // Event 360 = ProvisioningWillNotLaunch — snapshot-only, can flip back; keep batched
-            // to avoid flushing a transient state. (See project_hello_willlaunch_unreliable memory:
-            // willlaunch/willnotlaunch flip multiple times — expensive to flush each transition.)
+            // Event 360 = ProvisioningWillNotLaunch — same snapshot-only character as 358, and
+            // suppressed for the same reason. No backend event emitted.
             f.Tracker.ProcessHelloEvent(360, Fixed, providerName: "prov", isBackfill: false);
 
-            var info = f.InfoEvent("hello_provisioning_willnotlaunch");
-            Assert.Equal("false", info.Payload![SignalPayloadKeys.ImmediateUpload]);
+            Assert.DoesNotContain(f.Ingress.Posted, p =>
+                p.Kind == DecisionSignalKind.InformationalEvent
+                && p.Payload != null
+                && p.Payload.TryGetValue(SignalPayloadKeys.EventType, out var et)
+                && et == "hello_provisioning_willnotlaunch");
         }
 
         [Fact]

src/Agent/AutopilotMonitor.Agent.V2.Core.Tests/Orchestration/EnrollmentOrchestratorTests.cs

@@ -196,18 +196,25 @@ namespace AutopilotMonitor.Agent.V2.Core.Tests.Orchestration
                 sourceOrigin: "Test",
                 evidence: new Evidence(EvidenceKind.Synthetic, "test", "test"));
 
-            // Wait until the worker applied it.
+            // Wait until the worker applied it (StepIndex > 0 at minimum).
             Assert.True(SpinWait.SpinUntil(() => sut.CurrentState.StepIndex > 0, 2000));
             var stepIndexBeforeStop = sut.CurrentState.StepIndex;
 
             sut.Stop();
 
-            // Snapshot file must exist + contain the final state.
+            // Snapshot file must exist + contain the state AT STOP. SessionStarted may cascade
+            // through additional reducer steps (ClassifierTick arm/fire, bootstrap effects,
+            // now that the P0 fix actually lets deadlines run instead of dead-ending), so the
+            // final StepIndex is >= the value observed before Stop. Stop drains the ingress
+            // synchronously before saving the snapshot — so the saved StepIndex matches the
+            // state at the moment Save is called, which is what this test is really asserting.
             var snapshotPath = Path.Combine(rig.StateDir, "snapshot.json");
             Assert.True(File.Exists(snapshotPath));
             var reloaded = new SnapshotPersistence(snapshotPath).Load();
             Assert.NotNull(reloaded);
-            Assert.Equal(stepIndexBeforeStop, reloaded!.StepIndex);
+            Assert.Equal(sut.CurrentState.StepIndex, reloaded!.StepIndex);
+            Assert.True(reloaded!.StepIndex >= stepIndexBeforeStop,
+                $"Snapshot StepIndex ({reloaded.StepIndex}) regressed below observed ({stepIndexBeforeStop}).");
         }
 
         // ========================================================================= Dispose

src/Agent/AutopilotMonitor.Agent.V2.Core.Tests/Orchestration/RecoveryPathTests.cs

@@ -153,8 +153,26 @@ namespace AutopilotMonitor.Agent.V2.Core.Tests.Orchestration
             var signalLogPath = Path.Combine(rig.StateDir, "signal-log.jsonl");
             if (File.Exists(signalLogPath))
             {
-                // Fresh file should be empty (writer reinstantiated).
-                Assert.True(new FileInfo(signalLogPath).Length == 0);
+                // Fresh writer: the stale content MUST be gone. A non-empty fresh log is fine —
+                // Start() posts bootstrap signals (EspConfigDetected, deadline fires now that
+                // the P0 payload-key fix lets them transition cleanly, classifier ticks) and
+                // those legitimately land on the new log. What we assert here is strictly that
+                // the pre-quarantine "fake stale signal" line did NOT survive into the fresh log.
+                //
+                // Use FileShare.ReadWrite to avoid racing with an in-flight Append() from the
+                // ingress worker: SignalLogWriter.Append opens with FileShare.Read, which
+                // disallows a second writer but allows a reader that also grants FileShare.Write.
+                // File.ReadAllText uses FileShare.Read, which occasionally throws IOException
+                // exactly in the microseconds the writer holds the handle. A manual read with
+                // FileShare.ReadWrite is concurrency-safe for our snapshot-of-content check.
+                string fresh;
+                using (var fs = new FileStream(signalLogPath, FileMode.Open, FileAccess.Read, FileShare.ReadWrite))
+                using (var sr = new StreamReader(fs, Encoding.UTF8))
+                {
+                    fresh = sr.ReadToEnd();
+                }
+                Assert.DoesNotContain("fake", fresh);
+                Assert.DoesNotContain("stale signal", fresh);
             }
 
             var quarantineRoot = Path.Combine(rig.StateDir, ".quarantine");
@@ -539,6 +557,70 @@ namespace AutopilotMonitor.Agent.V2.Core.Tests.Orchestration
             sut.Stop();
         }
 
+        // ============================================================ First-session regression — deadline payload key
+
+        [Fact]
+        public void Past_due_FinalizingGrace_deadline_advances_stage_to_Completed()
+        {
+            // Regression guard for session 9ed7021e: the orchestrator's DeadlineFired bridge
+            // used to post payload `["deadlineName"]=...` while the reducer reads under
+            // SignalPayloadKeys.Deadline. Result: every deadline (FinalizingGrace, HelloSafety,
+            // ClassifierTick, …) dead-ended as "deadline_fired_without_name" — FinalizingGrace
+            // specifically left every Classic session hanging in Stage=Finalizing forever and
+            // blocked enrollment_complete emission, log flush, and diagnostics upload.
+            //
+            // This test seeds a state that mirrors the real failure: Stage=Finalizing with an
+            // already-past-due FinalizingGrace deadline (DueAt = UtcNow - 5m). After Start()
+            // rehydrates the deadline the scheduler fires it synchronously via ThreadPool; the
+            // bridge must forward ActiveDeadline.FiresPayload verbatim so the reducer can
+            // route to HandleFinalizingGraceDeadlineFired and transition Finalizing→Completed.
+            using var rig = new Rig();
+            Directory.CreateDirectory(rig.StateDir);
+
+            var pastDueFinalizing = new ActiveDeadline(
+                name: DeadlineNames.FinalizingGrace,
+                dueAtUtc: At.AddMinutes(-5),
+                firesSignalKind: DecisionSignalKind.DeadlineFired,
+                firesPayload: new Dictionary<string, string>(StringComparer.Ordinal)
+                {
+                    [SignalPayloadKeys.Deadline] = DeadlineNames.FinalizingGrace,
+                });
+            var finalizingSeed = DecisionState.CreateInitial("S1", "T1")
+                .ToBuilder()
+                .WithStage(SessionStage.Finalizing)
+                .WithStepIndex(10)
+                .WithLastAppliedSignalOrdinal(9)
+                .AddDeadline(pastDueFinalizing)
+                .Build();
+            new SnapshotPersistence(
+                Path.Combine(rig.StateDir, "snapshot.json"),
+                () => rig.Clock.UtcNow).Save(finalizingSeed);
+
+            var sut = rig.Build();
+            sut.Start();
+
+            Assert.True(
+                SpinWait.SpinUntil(() => sut.CurrentState.Stage == SessionStage.Completed, 5000),
+                $"Expected Stage=Completed after past-due FinalizingGrace fired; stuck at {sut.CurrentState.Stage}.");
+            Assert.Equal(SessionOutcome.EnrollmentComplete, sut.CurrentState.Outcome);
+
+            // Paranoid belt-and-braces: the posted DeadlineFired signal must carry the deadline
+            // name under SignalPayloadKeys.Deadline (the key the reducer reads). Absence of this
+            // key is precisely how the bug manifested — dead-end "deadline_fired_without_name".
+            var signalLog = GetSignalLog(sut);
+            var fired = signalLog.ReadAll()
+                .FirstOrDefault(s => s.Kind == DecisionSignalKind.DeadlineFired);
+            Assert.NotNull(fired);
+            Assert.NotNull(fired!.Payload);
+            Assert.True(
+                fired.Payload!.TryGetValue(SignalPayloadKeys.Deadline, out var name)
+                    && name == DeadlineNames.FinalizingGrace,
+                $"Expected payload[{SignalPayloadKeys.Deadline}]={DeadlineNames.FinalizingGrace}; " +
+                $"got keys=[{string.Join(",", fired.Payload.Keys)}].");
+
+            sut.Stop();
+        }
+
         // ============================================================ Codex #1 Phase 3 — deadline re-arm
 
         [Fact]
@@ -556,7 +638,7 @@ namespace AutopilotMonitor.Agent.V2.Core.Tests.Orchestration
                 firesSignalKind: DecisionSignalKind.DeadlineFired,
                 firesPayload: new Dictionary<string, string>(StringComparer.Ordinal)
                 {
-                    ["deadlineName"] = "hello_safety",
+                    [SignalPayloadKeys.Deadline] = "hello_safety",
                 });
             var seed = DecisionState.CreateInitial("S1", "T1")
                 .ToBuilder()
@@ -578,7 +660,7 @@ namespace AutopilotMonitor.Agent.V2.Core.Tests.Orchestration
                     () => signalLog.ReadAll().Any(s =>
                         s.Kind == DecisionSignalKind.DeadlineFired &&
                         s.Payload != null &&
-                        s.Payload.TryGetValue("deadlineName", out var n) &&
+                        s.Payload.TryGetValue(SignalPayloadKeys.Deadline, out var n) &&
                         n == "hello_safety"),
                     3000),
                 "Expected DeadlineFired signal for re-armed past-due deadline.");

src/Agent/AutopilotMonitor.Agent.V2.Core.Tests/Transport/TelemetrySpoolTests.cs

@@ -193,5 +193,50 @@ namespace AutopilotMonitor.Agent.V2.Core.Tests.Transport
             var next = s2.Enqueue(EventDraft("next"));
             Assert.Equal(1, next.TelemetryItemId);
         }
+
+        [Fact]
+        public void Enqueue_with_RequiresImmediateFlush_raises_ImmediateFlushRequested()
+        {
+            // Regression guard for the "30 s gap at session start" first-session bug. Before
+            // this wakeup, RequiresImmediateFlush=true items were persisted but the drain loop
+            // still slept the full _drainInterval before picking them up. If this event ever
+            // stops firing the delay comes back, and critical lifecycle items (agent_started,
+            // enrollment_failed on auth-failure, …) will again wait up to 30 s before upload.
+            using var tmp = new TempDirectory();
+            var spool = new TelemetrySpool(tmp.Path, Clock());
+
+            var signalled = 0;
+            spool.ImmediateFlushRequested += (s, e) => System.Threading.Interlocked.Increment(ref signalled);
+
+            var immediate = new TelemetryItemDraft(
+                kind: TelemetryItemKind.Event,
+                partitionKey: "tenant_session",
+                rowKey: "immediate",
+                payloadJson: "{\"EventType\":\"agent_started\"}",
+                isSessionScoped: true,
+                requiresImmediateFlush: true);
+
+            spool.Enqueue(immediate);
+
+            Assert.Equal(1, System.Threading.Volatile.Read(ref signalled));
+        }
+
+        [Fact]
+        public void Enqueue_without_RequiresImmediateFlush_does_not_raise_ImmediateFlushRequested()
+        {
+            // Batched items (performance_snapshot, routine signal/transition JSONL) must not
+            // wake the drain loop — otherwise the wakeup degenerates into a tight loop and the
+            // periodic-drain cadence loses its purpose.
+            using var tmp = new TempDirectory();
+            var spool = new TelemetrySpool(tmp.Path, Clock());
+
+            var signalled = 0;
+            spool.ImmediateFlushRequested += (s, e) => System.Threading.Interlocked.Increment(ref signalled);
+
+            spool.Enqueue(EventDraft("batched"));      // requiresImmediateFlush defaults to false
+            spool.Enqueue(SignalDraft("0000000001"));
+
+            Assert.Equal(0, System.Threading.Volatile.Read(ref signalled));
+        }
     }
 }

src/Agent/AutopilotMonitor.Agent.V2.Core/Monitoring/Enrollment/SystemSignals/HelloTracker.cs

@@ -456,11 +456,16 @@ namespace AutopilotMonitor.Agent.V2.Core.Monitoring.Enrollment.SystemSignals
             switch (eventId)
             {
                 case EventId_ProvisioningWillLaunch: // 358
-                    eventType = "hello_provisioning_willlaunch";
-                    severity = EventSeverity.Info;
-                    message = "Windows Hello for Business provisioning will launch - prerequisites passed (snapshot only, not a final state)";
-                    _logger.Info("Windows Hello provisioning will launch - prerequisites passed (snapshot only, not a final state)");
-                    break;
+                    // EventID 358 is a prerequisites-passed SNAPSHOT that flips multiple times
+                    // during a single enrollment (see session 9ed7021e: 358 fired 6×, including
+                    // three times within 232 ms around desktop arrival). Per project memory
+                    // `project_hello_willlaunch_unreliable` this must NEVER be used as Hello
+                    // resolution evidence, and the DecisionEngine already ignores it — the
+                    // event only fueled UI-timeline noise. Keep the debug log line so the
+                    // sequence is still reconstructible from agent-logs during diagnostics,
+                    // but do not emit a backend event.
+                    _logger.Debug("Hello EventID 358 (willlaunch) observed — snapshot only, suppressed");
+                    return;
 
                 case EventId_NgcKeyRegistered: // 300
                     eventType = "hello_provisioning_completed";
@@ -479,12 +484,10 @@ namespace AutopilotMonitor.Agent.V2.Core.Monitoring.Enrollment.SystemSignals
                     break;
 
                 case EventId_ProvisioningWillNotLaunch: // 360
-                    eventType = "hello_provisioning_willnotlaunch";
-                    severity = EventSeverity.Warning;
-                    message = "Windows Hello for Business provisioning prerequisites not met (snapshot only, not a final state)";
-                    // DO NOT mark Hello as completed - event 360 is just a snapshot and can change
-                    _logger.Info("Windows Hello provisioning prerequisites not met (snapshot, not terminal)");
-                    break;
+                    // Sibling of 358 — also a snapshot flag. DecisionEngine does not treat it
+                    // as terminal either, so suppress for the same noise-reduction reason.
+                    _logger.Debug("Hello EventID 360 (willnotlaunch) observed — snapshot only, suppressed");
+                    return;
 
                 case EventId_ProvisioningBlocked: // 362
                     eventType = "hello_provisioning_blocked";

src/Agent/AutopilotMonitor.Agent.V2.Core/Orchestration/EnrollmentOrchestrator.cs

@@ -115,6 +115,12 @@ namespace AutopilotMonitor.Agent.V2.Core.Orchestration
         // Drain-Loop
         private CancellationTokenSource? _drainCts;
         private Task? _drainTask;
+        // Set by TelemetrySpool.ImmediateFlushRequested. Replaced with a fresh instance at the
+        // start of each drain iteration so wakeups don't leak across iterations. Swapped via
+        // Volatile.Read/Write because producer (spool thread) and consumer (drain loop) run on
+        // different threads.
+        private TaskCompletionSource<bool>? _immediateFlushSignal;
+        private EventHandler? _immediateFlushBridge;
 
         // Lifecycle
         private int _started;
@@ -227,6 +233,14 @@ namespace AutopilotMonitor.Agent.V2.Core.Orchestration
         public ISignalIngressSink IngressSink =>
             (ISignalIngressSink?)_ingress ?? throw new InvalidOperationException("Orchestrator not started.");
 
+        /// <summary>
+        /// Test-only accessor for the internal <see cref="ITelemetrySpool"/>. Used by the
+        /// drain-wakeup regression tests to enqueue an immediate-flush item directly and
+        /// assert the drain loop wakes up early (without having to drive a real collector).
+        /// Production callers never need this — the orchestrator manages the spool lifecycle.
+        /// </summary>
+        internal ITelemetrySpool? SpoolForTests => _spool;
+
         /// <summary>
         /// Exposed so Program.cs can subscribe to <see cref="TelemetryUploadOrchestrator.ServerResponseReceived"/>
         /// for M4.6.ε DeviceBlocked / DeviceKillSignal / AdminAction / Actions plumbing.
@@ -427,6 +441,20 @@ namespace AutopilotMonitor.Agent.V2.Core.Orchestration
             _spool = new TelemetrySpool(_transportDirectory, _clock, _logger);
             _transport = new TelemetryUploadOrchestrator(_spool, _uploader, _clock);
 
+            // 3a) Immediate-flush wakeup — lifecycle-critical items (agent_started, hello wizard,
+            //     auth-failure shutdown, version-check, enrollment_failed on max-lifetime, …)
+            //     enqueue with RequiresImmediateFlush=true. Without this bridge they would sit in
+            //     the spool for up to _drainInterval before the periodic loop uploads them,
+            //     producing the ~30 s "session-register then silence" gap at session start.
+            //
+            //     The signal TCS is installed BEFORE the subscription — and both BEFORE the drain
+            //     task starts at step 15 — so any Enqueue fired during the rest of Start() (e.g.
+            //     the EspConfigDetected bootstrap at step 13c or the synchronous Classifier-Start
+            //     at step 11.5) sets the current signal rather than hitting a null field.
+            _immediateFlushSignal = new TaskCompletionSource<bool>(TaskCreationOptions.RunContinuationsAsynchronously);
+            _immediateFlushBridge = OnImmediateFlushRequested;
+            _spool.ImmediateFlushRequested += _immediateFlushBridge;
+
             // 4) Event-Emitter-Kette. Single-rail (Plan §5.10): der TelemetryEventEmitter wird
             //    lokal gebaut und nur in die zwei erlaubten Caller (EventTimelineEmitter,
             //    BackPressureEventObserver) injiziert. Kein Feld mehr auf dem Orchestrator —
@@ -838,6 +866,17 @@ namespace AutopilotMonitor.Agent.V2.Core.Orchestration
             }
             catch (Exception ex) { _logger.Warning($"EnrollmentOrchestrator: unsubscribe scheduler failed: {ex.Message}"); }
 
+            // 1a) Immediate-flush-Bridge abmelden — spätere Enqueues sollen den Drain-Loop nicht
+            //     wecken wenn wir ihn gerade terminieren.
+            try
+            {
+                if (_spool != null && _immediateFlushBridge != null)
+                {
+                    _spool.ImmediateFlushRequested -= _immediateFlushBridge;
+                }
+            }
+            catch (Exception ex) { _logger.Warning($"EnrollmentOrchestrator: unsubscribe immediate-flush failed: {ex.Message}"); }
+
             // 2) Scheduler disposen — stoppt alle aktiven Timer.
             try { _scheduler?.Dispose(); }
             catch (Exception ex) { _logger.Warning($"EnrollmentOrchestrator: scheduler dispose failed: {ex.Message}"); }
@@ -937,10 +976,24 @@ namespace AutopilotMonitor.Agent.V2.Core.Orchestration
                     identifier: e.Deadline.Name,
                     summary: $"deadline '{e.Deadline.Name}' fired at {e.Deadline.DueAtUtc:O}");
 
-                var payload = new Dictionary<string, string>(StringComparer.Ordinal)
+                // The reducer reads the deadline name under SignalPayloadKeys.Deadline
+                // (see HandleDeadlineFiredV1 in DecisionEngine.Shared). Forward the
+                // originating ActiveDeadline.FiresPayload — it already carries that key
+                // (see e.g. DecisionEngine.Classic.TransitionToFinalizing). Using a
+                // different key here dead-ends every deadline with "deadline_fired_without_name".
+                var payload = new Dictionary<string, string>(StringComparer.Ordinal);
+                if (e.Deadline.FiresPayload != null)
                 {
-                    ["deadlineName"] = e.Deadline.Name,
-                };
+                    foreach (var kv in e.Deadline.FiresPayload)
+                    {
+                        payload[kv.Key] = kv.Value;
+                    }
+                }
+
+                if (!payload.ContainsKey(SignalPayloadKeys.Deadline))
+                {
+                    payload[SignalPayloadKeys.Deadline] = e.Deadline.Name;
+                }
 
                 // OccurredAtUtc = DueAtUtc (not firedAt) — replay-determinism per DeadlineFiredEventArgs doc.
                 _ingress.Post(
@@ -957,23 +1010,57 @@ namespace AutopilotMonitor.Agent.V2.Core.Orchestration
             }
         }
 
+        private void OnImmediateFlushRequested(object? sender, EventArgs e)
+        {
+            // Release the drain loop's current iteration early. Multiple rapid wakeups coalesce
+            // naturally into a single drain (WhenAny returns once; the next iteration installs a
+            // fresh TCS). TrySetResult is idempotent, so the second wakeup is a no-op.
+            Volatile.Read(ref _immediateFlushSignal)?.TrySetResult(true);
+        }
+
         private async Task DrainLoopAsync(CancellationToken ct)
         {
             try
             {
                 while (!ct.IsCancellationRequested)
                 {
+                    // Read the current wakeup signal. Step 3a installs the very first TCS
+                    // BEFORE subscribing the bridge and before the drain task starts; every
+                    // subsequent iteration receives a freshly-installed TCS from the drain
+                    // that just completed below. OnImmediateFlushRequested races with
+                    // Task.Delay so RequiresImmediateFlush=true items reach the backend within
+                    // milliseconds instead of waiting for the full _drainInterval window.
+                    var signal = Volatile.Read(ref _immediateFlushSignal);
+                    if (signal == null)
+                    {
+                        // Defensive: if Start() didn't seed the TCS (should never happen), fall
+                        // back to a periodic-only drain so the loop still makes progress.
+                        signal = new TaskCompletionSource<bool>(TaskCreationOptions.RunContinuationsAsynchronously);
+                        Volatile.Write(ref _immediateFlushSignal, signal);
+                    }
+
                     try
                     {
                         // Wall-clock delay, NOT _clock.Delay: drain is a real network cadence
                         // independent of the decision-engine's logical time. VirtualClock.Delay
                         // returns immediately and would cause a tight loop in unit tests.
-                        await Task.Delay(_drainInterval, ct).ConfigureAwait(false);
+                        var delayTask = Task.Delay(_drainInterval, ct);
+                        await Task.WhenAny(delayTask, signal.Task).ConfigureAwait(false);
                     }
                     catch (OperationCanceledException) { return; }
 
                     if (ct.IsCancellationRequested) return;
 
+                    // Swap in a fresh TCS BEFORE draining so wakeups raised during this drain
+                    // land on the next iteration's signal rather than on the stale completed
+                    // TCS (which TrySetResult is a no-op for, losing the wakeup). The tiny
+                    // window between WhenAny returning and this Volatile.Write is acceptable:
+                    // a wakeup there sees the just-completed TCS, is dropped, but we're
+                    // already about to drain in the same iteration — so the item ships anyway.
+                    Volatile.Write(
+                        ref _immediateFlushSignal,
+                        new TaskCompletionSource<bool>(TaskCreationOptions.RunContinuationsAsynchronously));
+
                     try
                     {
                         var result = await _transport!.DrainAllAsync(ct).ConfigureAwait(false);

src/Agent/AutopilotMonitor.Agent.V2.Core/Transport/Telemetry/ITelemetrySpool.cs

@@ -1,4 +1,5 @@
 #nullable enable
+using System;
 using System.Collections.Generic;
 
 namespace AutopilotMonitor.Agent.V2.Core.Transport.Telemetry
@@ -42,5 +43,15 @@ namespace AutopilotMonitor.Agent.V2.Core.Transport.Telemetry
 
         /// <summary>Cursor: bis zu dieser ID inklusive wurde bereits erfolgreich hochgeladen; -1 initial.</summary>
         long LastUploadedItemId { get; }
+
+        /// <summary>
+        /// Raised after <see cref="Enqueue"/> persists an item whose <see cref="TelemetryItemDraft.RequiresImmediateFlush"/>
+        /// was <c>true</c>. The orchestrator uses this to wake the drain loop early — without a
+        /// wakeup, <c>ImmediateUpload=true</c> items wait for the full periodic drain interval
+        /// (default 30 s) before reaching the backend, making the first burst of lifecycle events
+        /// (<c>agent_started</c>, <c>enrollment_type_detected</c>, first <c>esp_phase_changed</c>, …)
+        /// appear as a 30 s gap in the UI timeline.
+        /// </summary>
+        event EventHandler? ImmediateFlushRequested;
     }
 }

src/Agent/AutopilotMonitor.Agent.V2.Core/Transport/Telemetry/TelemetrySpool.cs

@@ -66,16 +66,19 @@ namespace AutopilotMonitor.Agent.V2.Core.Transport.Telemetry
             get { lock (_lock) return _lastUploadedItemId; }
         }
 
+        public event EventHandler? ImmediateFlushRequested;
+
         public TelemetryItem Enqueue(TelemetryItemDraft draft)
         {
             if (draft == null) throw new ArgumentNullException(nameof(draft));
 
+            TelemetryItem item;
             lock (_lock)
             {
                 var itemId = _lastAssignedItemId + 1;
                 long? traceOrdinal = draft.IsSessionScoped ? itemId : (long?)null;
 
-                var item = new TelemetryItem(
+                item = new TelemetryItem(
                     kind: draft.Kind,
                     partitionKey: draft.PartitionKey,
                     rowKey: draft.RowKey,
@@ -105,8 +108,21 @@ namespace AutopilotMonitor.Agent.V2.Core.Transport.Telemetry
                 // event flows; users can enable at troubleshoot-time).
                 _logger?.Debug(
                     $"TelemetrySpool: enqueued itemId={itemId} kind={item.Kind} immediate={item.RequiresImmediateFlush} pending={itemId - _lastUploadedItemId}.");
-                return item;
             }
+
+            // Fire outside the lock — subscribers must not be able to re-enter Enqueue or
+            // hold the write path while waking the drain loop. Swallow handler exceptions:
+            // a buggy subscriber must not break telemetry persistence.
+            if (draft.RequiresImmediateFlush)
+            {
+                try { ImmediateFlushRequested?.Invoke(this, EventArgs.Empty); }
+                catch (Exception ex)
+                {
+                    _logger?.Warning($"TelemetrySpool: ImmediateFlushRequested handler threw: {ex.Message}");
+                }
+            }
+
+            return item;
         }
 
         public IReadOnlyList<TelemetryItem> Peek(int max)