diff --git a/src/Agent/AutopilotMonitor.Agent.V2.Core.Tests/Integration/ScenarioTests.cs b/src/Agent/AutopilotMonitor.Agent.V2.Core.Tests/Integration/ScenarioTests.cs index 1cb94b40..5711a67c 100644 --- a/src/Agent/AutopilotMonitor.Agent.V2.Core.Tests/Integration/ScenarioTests.cs +++ b/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 { /// @@ -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(); } diff --git a/src/Agent/AutopilotMonitor.Agent.V2.Core.Tests/Monitoring/HelloTrackerImmediateUploadTests.cs b/src/Agent/AutopilotMonitor.Agent.V2.Core.Tests/Monitoring/HelloTrackerImmediateUploadTests.cs index 316e0085..3453b3a9 100644 --- a/src/Agent/AutopilotMonitor.Agent.V2.Core.Tests/Monitoring/HelloTrackerImmediateUploadTests.cs +++ b/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 /// (hello_policy_detected) and on terminal events /// (hello_provisioning_completed/failed/blocked, hello_skipped); and stays - /// false on snapshot-type events that can flip multiple times - /// (hello_provisioning_willlaunch, hello_provisioning_willnotlaunch, - /// hello_pin_status) so transient re-registrations don't trigger a flush each time. + /// false on the snapshot-type hello_pin_status ticker so transient re-registrations + /// don't trigger a flush each time. + /// + /// The willlaunch / willnotlaunch 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 project_hello_willlaunch_unreliable. Suppression + /// is asserted below. + /// /// 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] diff --git a/src/Agent/AutopilotMonitor.Agent.V2.Core.Tests/Orchestration/EnrollmentOrchestratorTests.cs b/src/Agent/AutopilotMonitor.Agent.V2.Core.Tests/Orchestration/EnrollmentOrchestratorTests.cs index e633a520..57f6fe8c 100644 --- a/src/Agent/AutopilotMonitor.Agent.V2.Core.Tests/Orchestration/EnrollmentOrchestratorTests.cs +++ b/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 diff --git a/src/Agent/AutopilotMonitor.Agent.V2.Core.Tests/Orchestration/RecoveryPathTests.cs b/src/Agent/AutopilotMonitor.Agent.V2.Core.Tests/Orchestration/RecoveryPathTests.cs index 97d4d388..c89d0fe6 100644 --- a/src/Agent/AutopilotMonitor.Agent.V2.Core.Tests/Orchestration/RecoveryPathTests.cs +++ b/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(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(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."); diff --git a/src/Agent/AutopilotMonitor.Agent.V2.Core.Tests/Transport/TelemetrySpoolTests.cs b/src/Agent/AutopilotMonitor.Agent.V2.Core.Tests/Transport/TelemetrySpoolTests.cs index f90ab1db..c3394b6e 100644 --- a/src/Agent/AutopilotMonitor.Agent.V2.Core.Tests/Transport/TelemetrySpoolTests.cs +++ b/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)); + } } } diff --git a/src/Agent/AutopilotMonitor.Agent.V2.Core/Monitoring/Enrollment/SystemSignals/HelloTracker.cs b/src/Agent/AutopilotMonitor.Agent.V2.Core/Monitoring/Enrollment/SystemSignals/HelloTracker.cs index 70bd3bf5..f1730431 100644 --- a/src/Agent/AutopilotMonitor.Agent.V2.Core/Monitoring/Enrollment/SystemSignals/HelloTracker.cs +++ b/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"; diff --git a/src/Agent/AutopilotMonitor.Agent.V2.Core/Orchestration/EnrollmentOrchestrator.cs b/src/Agent/AutopilotMonitor.Agent.V2.Core/Orchestration/EnrollmentOrchestrator.cs index fc279a4c..36309253 100644 --- a/src/Agent/AutopilotMonitor.Agent.V2.Core/Orchestration/EnrollmentOrchestrator.cs +++ b/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? _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."); + /// + /// Test-only accessor for the internal . 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. + /// + internal ITelemetrySpool? SpoolForTests => _spool; + /// /// Exposed so Program.cs can subscribe to /// 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(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(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(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(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(TaskCreationOptions.RunContinuationsAsynchronously)); + try { var result = await _transport!.DrainAllAsync(ct).ConfigureAwait(false); diff --git a/src/Agent/AutopilotMonitor.Agent.V2.Core/Transport/Telemetry/ITelemetrySpool.cs b/src/Agent/AutopilotMonitor.Agent.V2.Core/Transport/Telemetry/ITelemetrySpool.cs index 0c168336..03dde985 100644 --- a/src/Agent/AutopilotMonitor.Agent.V2.Core/Transport/Telemetry/ITelemetrySpool.cs +++ b/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 /// Cursor: bis zu dieser ID inklusive wurde bereits erfolgreich hochgeladen; -1 initial. long LastUploadedItemId { get; } + + /// + /// Raised after persists an item whose + /// was true. The orchestrator uses this to wake the drain loop early — without a + /// wakeup, ImmediateUpload=true items wait for the full periodic drain interval + /// (default 30 s) before reaching the backend, making the first burst of lifecycle events + /// (agent_started, enrollment_type_detected, first esp_phase_changed, …) + /// appear as a 30 s gap in the UI timeline. + /// + event EventHandler? ImmediateFlushRequested; } } diff --git a/src/Agent/AutopilotMonitor.Agent.V2.Core/Transport/Telemetry/TelemetrySpool.cs b/src/Agent/AutopilotMonitor.Agent.V2.Core/Transport/Telemetry/TelemetrySpool.cs index c3f3dbf6..f8a7b685 100644 --- a/src/Agent/AutopilotMonitor.Agent.V2.Core/Transport/Telemetry/TelemetrySpool.cs +++ b/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 Peek(int max)