From eb94c612e7c6c4324430a607e7df0d554cbb10bc Mon Sep 17 00:00:00 2001 From: Oliver Kieselbach Date: Tue, 4 Aug 2026 14:44:35 +0200 Subject: [PATCH] fix(agent): gather phase/event triggers follow the engine-reduced timeline phase_change/phase_exit/on_event gather triggers (and the phase-scope gates) were driven by the pre-reduce SignalIngress.SignalPosted stream, so any raw signal carrying an EspPhase payload fired them before the decision engine had applied its gates. In session 32312a32 (RealmJoin tenant) a Phase Start rule on FinalizingSetup fired at the raw ESP-exit EspPhaseChanged(FinalizingSetup) signal, 7 minutes before the engine's RealmJoin-gated phase_transition(FinalizingSetup) - and read HKLM\SOFTWARE\RealmJoin\Custom\BIOS before the RealmJoin package that creates it had run (exists=false). GatherRuleExecutorHost now subscribes to the new TimelineEventStream: EventTimelineEmitter publishes every emitted event's (eventType, phase) post-reduce, so phase triggers fire exactly when the UI timeline shows the phase, and on_event sees engine-emitted event types (enrollment_complete, phase_transition, realmjoin_resolved, ...) that never existed on the raw signal stream - restoring V1 semantics and the documented "On Event with enrollment_complete" pattern. Raw-signal consumers with OS-state semantics (DeviceInfoHost, UserEspKeepAwakeHost, ProvisioningPackageHost) stay on the raw feed deliberately. Co-Authored-By: Claude Fable 5 --- docs/log.md | 4 + docs/rules/gather-rule-phase-scoping.md | 35 +++- .../EnrollmentOrchestratorEventBridgeTests.cs | 5 +- ...lmentOrchestratorWhiteGlovePatternTests.cs | 3 +- ...GatherRuleExecutorHostTimelineFeedTests.cs | 189 ++++++++++++++++++ .../Events/EventTimelineEmitterTests.cs | 52 +++++ .../Orchestration/DefaultComponentFactory.cs | 12 +- .../Orchestration/EnrollmentOrchestrator.cs | 11 +- .../Hosts/GatherRuleExecutorHost.cs | 84 ++++---- .../Orchestration/IComponentFactory.cs | 10 +- .../Orchestration/TimelineEventStream.cs | 55 +++++ .../Telemetry/Events/EventTimelineEmitter.cs | 9 +- 12 files changed, 417 insertions(+), 52 deletions(-) create mode 100644 src/Agent/AutopilotMonitor.Agent.V2.Core.Tests/Orchestration/GatherRuleExecutorHostTimelineFeedTests.cs create mode 100644 src/Agent/AutopilotMonitor.Agent.V2.Core/Orchestration/TimelineEventStream.cs diff --git a/docs/log.md b/docs/log.md index e348bc1b..10040f69 100644 --- a/docs/log.md +++ b/docs/log.md @@ -1,5 +1,9 @@ # Log +## 2026-08-04 + +* **Update**: `rules/gather-rule-phase-scoping.md` — new section "Which phase feeds the triggers": `GatherRuleExecutorHost` now drives `phase_change`/`phase_exit`/`on_event` gather triggers (and the phase-scope gates) from the post-reduce `TimelineEventStream` (every emitted event's `(eventType, phase)` published by `EventTimelineEmitter`) instead of the pre-reduce `SignalIngress.SignalPosted` payloads. Trigger: session `32312a32` (RealmJoin tenant rsneuffen.de) — a Phase-Start rule on FinalizingSetup fired at the raw ESP-exit `EspPhaseChanged(FinalizingSetup)` signal, 7 minutes before the engine's RealmJoin-gated `phase_transition(FinalizingSetup)`, reading `HKLM\SOFTWARE\RealmJoin\Custom\BIOS` before the RealmJoin package created it. Phase triggers now fire exactly when the timeline shows the phase; `on_event` additionally sees engine-emitted event types (`enrollment_complete`, `realmjoin_resolved`, …) for the first time in V2, restoring V1 semantics and the documented end-of-enrollment pattern. Other raw-signal hosts (DeviceInfo, keep-awake, PPKG) deliberately unchanged. + ## 2026-08-02 * **Creation**: Added `backend/tenant-activation-welcome-mail.md` — triggered by the first production auto-approve (activation at signup+64 s, notification address saved at +69 s → welcome mail silently skipped, since the approval path is the only sender and reads the address once). Fix documented in the doc: the notification-email save path now also triggers the send (`PUT /api/preview/notification-email` → fresh approval check `IsApprovedFreshAsync`, bypassing the 30 s negative cache), and a conditional-insert `welcome-email-sent` marker in `PreviewWhitelist` makes the send once-only across both racing paths (marker consumed strictly after the address check; revoke clears it, offboarding wipes the partition, the GA resend endpoint sends unconditionally and consumes it best-effort). diff --git a/docs/rules/gather-rule-phase-scoping.md b/docs/rules/gather-rule-phase-scoping.md index d785e490..953b2541 100644 --- a/docs/rules/gather-rule-phase-scoping.md +++ b/docs/rules/gather-rule-phase-scoping.md @@ -8,7 +8,7 @@ tags: - backend - gather-rules - config -timestamp: 2026-07-20T00:00:00+02:00 +timestamp: 2026-08-04T00:00:00+02:00 --- # Problem @@ -64,6 +64,35 @@ enter-rule and an exit-rule on the same phase never cannibalise each other's slo * Old agents do not know the trigger value and simply never fire the rule (silent no-op) — which is why `phase_exit` needs no ConfigVersion bump: no new field is transported. +# Which "phase" feeds the triggers — the emitted timeline, not raw signals + +The phase the executor sees is fed by `GatherRuleExecutorHost` from the **post-reduce +`TimelineEventStream`**: `EventTimelineEmitter` publishes every emitted event's +`(eventType, phase)` after the decision engine reduced the originating signal. Phase triggers +therefore fire exactly when the UI timeline shows the phase — by construction. + +This replaced the original V2 wiring (`SignalIngress.SignalPosted`), which fires at enqueue +time with the **raw** collector payload. The raw phase can run minutes ahead of the +engine-declared one: in session `32312a32` (RealmJoin tenant) the Shell-Core ESP-exit +synthesized `EspPhaseChanged(FinalizingSetup)` at 12:18, while the engine — holding the +transition behind the RealmJoin completion gate until the first deployment resolved — declared +`phase_transition(FinalizingSetup)` at 12:25. A `phase_change` rule on `FinalizingSetup` +reading `HKLM\SOFTWARE\RealmJoin\Custom\BIOS` fired 7 minutes early and read the key before +the RealmJoin package wrote it (`exists=false`). + +Consequences of the emitted-event feed: + +* `phase_change` / `phase_exit` / phase **scope** gates all follow the engine-reduced phase + (single feed, consistent with the portal's "collects once when the enrollment reaches …"). +* `on_event` now sees **every emitted event type**, including engine-emitted ones + (`enrollment_complete`, `phase_transition`, `realmjoin_resolved`, …) that never existed on + the raw signal stream — the documented "On Event with `enrollment_complete`" end-of-enrollment + pattern only works because of this. This restores V1 semantics (V1 fired triggers from every + emitted `EnrollmentEvent`). +* Other hosts (`DeviceInfoHost`, `UserEspKeepAwakeHost`, `ProvisioningPackageHost`) still key + off the raw signal stream deliberately — their collections are about OS state at the raw + boundary (ESP exit, keep-awake), not about timeline agreement. + # Scope semantics (agent, `GatherRuleExecutor`) * Scope applies to **all trigger types**: @@ -130,7 +159,9 @@ enter-rule and an exit-rule on the same phase never cannibalise each other's slo # Citations * `src/Agent/AutopilotMonitor.Agent.V2.Core/Monitoring/Telemetry/Gather/GatherRuleExecutor.cs` — scope gates, latch, canonical hash +* `src/Agent/AutopilotMonitor.Agent.V2.Core/Orchestration/Hosts/GatherRuleExecutorHost.cs` — timeline-event trigger feed +* `src/Agent/AutopilotMonitor.Agent.V2.Core/Orchestration/TimelineEventStream.cs` — post-reduce observer surface * `src/Backend/AutopilotMonitor.Functions/Functions/Rules/GatherRulesFunction.cs` — `ValidateScopeAndEmitMode` * `src/Backend/AutopilotMonitor.Functions/Services/GatherRuleService.cs` — partial-PUT merge, `ContentEquivalent` * `src/Shared/AutopilotMonitor.Shared/Models/Rules/GatherRule.cs` — field contracts -* Tests: `GatherRuleExecutorPhaseScopeTests`, `GatherRuleExecutorEmitModeTests` (agent); `GatherRuleUpdatePartialMergeTests`, `GatherRuleScopeFieldsTests` (backend) +* Tests: `GatherRuleExecutorPhaseScopeTests`, `GatherRuleExecutorEmitModeTests`, `GatherRuleExecutorHostTimelineFeedTests` (agent); `GatherRuleUpdatePartialMergeTests`, `GatherRuleScopeFieldsTests` (backend) diff --git a/src/Agent/AutopilotMonitor.Agent.V2.Core.Tests/Orchestration/EnrollmentOrchestratorEventBridgeTests.cs b/src/Agent/AutopilotMonitor.Agent.V2.Core.Tests/Orchestration/EnrollmentOrchestratorEventBridgeTests.cs index a13f41cc..25f7e8df 100644 --- a/src/Agent/AutopilotMonitor.Agent.V2.Core.Tests/Orchestration/EnrollmentOrchestratorEventBridgeTests.cs +++ b/src/Agent/AutopilotMonitor.Agent.V2.Core.Tests/Orchestration/EnrollmentOrchestratorEventBridgeTests.cs @@ -54,6 +54,7 @@ namespace AutopilotMonitor.Agent.V2.Core.Tests.Orchestration public ISignalIngressSink? CapturedIngress { get; private set; } public IClock? CapturedClock { get; private set; } public AutopilotMonitor.Agent.V2.Core.Transport.Telemetry.ITelemetrySpool? CapturedTelemetrySpool { get; private set; } + public TimelineEventStream? CapturedTimelineEvents { get; private set; } public CollectorSurfaces CreateCollectorHosts( string sessionId, @@ -62,12 +63,14 @@ namespace AutopilotMonitor.Agent.V2.Core.Tests.Orchestration IReadOnlyCollection whiteGloveSealingPatternIds, ISignalIngressSink ingress, IClock clock, - AutopilotMonitor.Agent.V2.Core.Transport.Telemetry.ITelemetrySpool? telemetrySpool) + AutopilotMonitor.Agent.V2.Core.Transport.Telemetry.ITelemetrySpool? telemetrySpool, + TimelineEventStream? timelineEvents = null) { CapturedWhiteGloveSealingPatternIds = whiteGloveSealingPatternIds; CapturedIngress = ingress; CapturedClock = clock; CapturedTelemetrySpool = telemetrySpool; + CapturedTimelineEvents = timelineEvents; foreach (var name in HostNames) { Hosts.Add(new FakeCollectorHost(name)); diff --git a/src/Agent/AutopilotMonitor.Agent.V2.Core.Tests/Orchestration/EnrollmentOrchestratorWhiteGlovePatternTests.cs b/src/Agent/AutopilotMonitor.Agent.V2.Core.Tests/Orchestration/EnrollmentOrchestratorWhiteGlovePatternTests.cs index 8e54f5eb..030d0039 100644 --- a/src/Agent/AutopilotMonitor.Agent.V2.Core.Tests/Orchestration/EnrollmentOrchestratorWhiteGlovePatternTests.cs +++ b/src/Agent/AutopilotMonitor.Agent.V2.Core.Tests/Orchestration/EnrollmentOrchestratorWhiteGlovePatternTests.cs @@ -27,7 +27,8 @@ namespace AutopilotMonitor.Agent.V2.Core.Tests.Orchestration IReadOnlyCollection whiteGloveSealingPatternIds, ISignalIngressSink ingress, IClock clock, - AutopilotMonitor.Agent.V2.Core.Transport.Telemetry.ITelemetrySpool? telemetrySpool) + AutopilotMonitor.Agent.V2.Core.Transport.Telemetry.ITelemetrySpool? telemetrySpool, + TimelineEventStream? timelineEvents = null) { CapturedPatternIds = whiteGloveSealingPatternIds; return new CollectorSurfaces(Array.Empty()); diff --git a/src/Agent/AutopilotMonitor.Agent.V2.Core.Tests/Orchestration/GatherRuleExecutorHostTimelineFeedTests.cs b/src/Agent/AutopilotMonitor.Agent.V2.Core.Tests/Orchestration/GatherRuleExecutorHostTimelineFeedTests.cs new file mode 100644 index 00000000..ccf29d73 --- /dev/null +++ b/src/Agent/AutopilotMonitor.Agent.V2.Core.Tests/Orchestration/GatherRuleExecutorHostTimelineFeedTests.cs @@ -0,0 +1,189 @@ +#nullable enable +using System; +using System.Collections.Generic; +using System.Diagnostics; +using System.Threading; +using AutopilotMonitor.Agent.V2.Core.Logging; +using AutopilotMonitor.Agent.V2.Core.Orchestration; +using AutopilotMonitor.Agent.V2.Core.Tests.Harness; +using AutopilotMonitor.DecisionCore.Engine; +using AutopilotMonitor.DecisionCore.Signals; +using AutopilotMonitor.Shared.Models; +using Xunit; + +namespace AutopilotMonitor.Agent.V2.Core.Tests.Orchestration +{ + /// + /// GatherRuleExecutorHost feeds its phase_change / phase_exit / on_event triggers from the + /// POST-REDUCE (emitted timeline events), not from raw + /// signals. Regression anchor: session 32312a32 (rsneuffen.de) — a phase_change rule on + /// FinalizingSetup fired at the raw EspPhaseChanged(FinalizingSetup) signal (ESP exit), + /// 7 minutes before the engine's RealmJoin-gated phase_transition(FinalizingSetup), and read + /// a registry key before the RealmJoin package wrote it. With the timeline feed the rule + /// fires only when the phase declaration is actually emitted. + /// Rules use the absent-registry-path pattern so no real system state is touched; execution + /// is observed via the InformationalEvent posts the host sends through its ingress sink. + /// + [Collection("SerialThreading")] // rules execute on the shared ThreadPool + public sealed class GatherRuleExecutorHostTimelineFeedTests : IDisposable + { + private const string AbsentPath = + "HKLM\\SOFTWARE\\AutopilotMonitorTests\\DefinitelyAbsent_5b1f03a2-0000-0000-0000-000000000000"; + + private readonly TempDirectory _tmp = new TempDirectory(); + private readonly CapturingSink _sink = new CapturingSink(); + private readonly TimelineEventStream _stream = new TimelineEventStream(); + private GatherRuleExecutorHost? _host; + + public void Dispose() + { + _host?.Dispose(); + _tmp.Dispose(); + } + + private sealed class CapturingSink : ISignalIngressSink + { + private readonly List _eventTypes = new List(); + private readonly object _gate = new object(); + + public int CountOf(string eventType) + { + lock (_gate) + { + var n = 0; + foreach (var t in _eventTypes) + if (string.Equals(t, eventType, StringComparison.Ordinal)) n++; + return n; + } + } + + public void Post( + DecisionSignalKind kind, + DateTime occurredAtUtc, + string sourceOrigin, + Evidence evidence, + IReadOnlyDictionary? payload = null, + int kindSchemaVersion = 1, + object? typedPayload = null) + { + if (kind != DecisionSignalKind.InformationalEvent || payload == null) return; + if (!payload.TryGetValue(SignalPayloadKeys.EventType, out var eventType)) return; + lock (_gate) _eventTypes.Add(eventType); + } + } + + private void BuildHost(params GatherRule[] rules) + { + var logger = new AgentLogger(_tmp.Path, AgentLogLevel.Info); + _host = new GatherRuleExecutorHost( + sessionId: "sess", + tenantId: "tenant", + ingress: _sink, + clock: new SystemClock(), + logger: logger, + rules: new List(rules), + imeLogPathOverride: null, + unrestrictedMode: true, + gatherDebugLogPath: null, + timelineEvents: _stream); + _host.Start(); + } + + private bool WaitForCount(string eventType, int count, int timeoutMs = 10000) + { + var sw = Stopwatch.StartNew(); + while (sw.ElapsedMilliseconds < timeoutMs) + { + if (_sink.CountOf(eventType) >= count) return true; + Thread.Sleep(25); + } + return _sink.CountOf(eventType) >= count; + } + + private static GatherRule Rule(string id, string trigger, string? triggerPhase = null, string? triggerEventType = null) + => new GatherRule + { + RuleId = id, + Title = id, + CollectorType = "registry", + Target = AbsentPath, + Trigger = trigger, + TriggerPhase = triggerPhase ?? string.Empty, + TriggerEventType = triggerEventType ?? string.Empty, + OutputEventType = "gather_host_test", + Enabled = true, + }; + + [Fact] + public void PhaseChangeRule_FiresOnEmittedPhaseDeclaration_NotOnPhaselessEvents() + { + BuildHost(Rule("GATHER-HOST-001", "phase_change", triggerPhase: "FinalizingSetup")); + + // The raw-world sequence of session 32312a32 as it reaches the emitted timeline: + // esp_exiting and hello_wizard_started are emitted WITHOUT a phase declaration + // (the engine holds FinalizingSetup back behind the RealmJoin gate) — no fire. + _stream.Publish("esp_exiting", EnrollmentPhase.Unknown); + _stream.Publish("hello_wizard_started", EnrollmentPhase.Unknown); + Thread.Sleep(300); + Assert.Equal(0, _sink.CountOf("gather_host_test")); + + // Only the engine's actual phase declaration fires the rule. + _stream.Publish("phase_transition", EnrollmentPhase.FinalizingSetup); + Assert.True(WaitForCount("gather_host_test", 1)); + } + + [Fact] + public void OnEventRule_FiresForEngineEmittedEventTypes() + { + // enrollment_complete is an engine-emitted effect event; it never existed on the raw + // InformationalEvent signal stream, so the docs-promised "On Event with + // enrollment_complete" trigger was dead under the previous SignalPosted feed. + BuildHost(Rule("GATHER-HOST-002", "on_event", triggerEventType: "enrollment_complete")); + + _stream.Publish("enrollment_complete", EnrollmentPhase.Unknown); + Assert.True(WaitForCount("gather_host_test", 1)); + } + + [Fact] + public void PhaseFeed_DedupsConsecutiveSamePhaseDeclarations() + { + // "Any phase" phase_change rule fires on every phase TRANSITION — a re-declared + // identical phase (e.g. duplicate esp_phase_changed) must not count as one. + BuildHost(Rule("GATHER-HOST-003", "phase_change", triggerPhase: "")); + + _stream.Publish("esp_phase_changed", EnrollmentPhase.DeviceSetup); + Assert.True(WaitForCount("gather_host_test", 1)); + + _stream.Publish("esp_phase_changed", EnrollmentPhase.DeviceSetup); + Thread.Sleep(300); + Assert.Equal(1, _sink.CountOf("gather_host_test")); + + _stream.Publish("esp_phase_changed", EnrollmentPhase.AccountSetup); + Assert.True(WaitForCount("gather_host_test", 2)); + } + + [Fact] + public void PhaseExitRule_FiresWhenEmittedTimelineLeavesThePhase() + { + BuildHost(Rule("GATHER-HOST-004", "phase_exit", triggerPhase: "AccountSetup")); + + _stream.Publish("esp_phase_changed", EnrollmentPhase.AccountSetup); + Thread.Sleep(300); + Assert.Equal(0, _sink.CountOf("gather_host_test")); + + _stream.Publish("phase_transition", EnrollmentPhase.FinalizingSetup); + Assert.True(WaitForCount("gather_host_test", 1)); + } + + [Fact] + public void StoppedHost_NoLongerReactsToTheStream() + { + BuildHost(Rule("GATHER-HOST-005", "on_event", triggerEventType: "enrollment_complete")); + + _host!.Stop(); + _stream.Publish("enrollment_complete", EnrollmentPhase.Unknown); + Thread.Sleep(300); + Assert.Equal(0, _sink.CountOf("gather_host_test")); + } + } +} diff --git a/src/Agent/AutopilotMonitor.Agent.V2.Core.Tests/Telemetry/Events/EventTimelineEmitterTests.cs b/src/Agent/AutopilotMonitor.Agent.V2.Core.Tests/Telemetry/Events/EventTimelineEmitterTests.cs index 45de3b74..9823fbba 100644 --- a/src/Agent/AutopilotMonitor.Agent.V2.Core.Tests/Telemetry/Events/EventTimelineEmitterTests.cs +++ b/src/Agent/AutopilotMonitor.Agent.V2.Core.Tests/Telemetry/Events/EventTimelineEmitterTests.cs @@ -153,6 +153,58 @@ namespace AutopilotMonitor.Agent.V2.Core.Tests.Telemetry.Events Assert.Equal("Device Setup", (string?)parsed["PhaseName"]); } + [Fact] + public void Emit_publishes_eventType_and_parsed_phase_to_TimelineEventStream() + { + // Post-reduce observer surface: gather-rule phase/event triggers subscribe here so + // they fire in step with what the timeline shows (session 32312a32 regression). + using var r = new Rig(); + var stream = new AutopilotMonitor.Agent.V2.Core.Orchestration.TimelineEventStream(); + var published = new List<(string EventType, EnrollmentPhase Phase)>(); + stream.EventEmitted += (eventType, phase) => published.Add((eventType, phase)); + var sut = new EventTimelineEmitter(r.Inner, stream); + + sut.Emit( + new Dictionary + { + ["eventType"] = "phase_transition", + ["phase"] = "FinalizingSetup", + }, + State(), + At); + sut.Emit( + new Dictionary { ["eventType"] = "enrollment_complete" }, + State(), + At); + + Assert.Equal(2, published.Count); + Assert.Equal(("phase_transition", EnrollmentPhase.FinalizingSetup), published[0]); + Assert.Equal(("enrollment_complete", EnrollmentPhase.Unknown), published[1]); + // Publish happens AFTER the transport enqueue — wire ordering is authoritative. + Assert.Equal(2, r.Transport.EnqueueCount); + } + + [Fact] + public void TimelineEventStream_subscriber_exception_does_not_break_the_emit_path() + { + using var r = new Rig(); + var stream = new AutopilotMonitor.Agent.V2.Core.Orchestration.TimelineEventStream(); + var secondSubscriberCalls = 0; + stream.EventEmitted += (_, _) => throw new InvalidOperationException("subscriber bug"); + stream.EventEmitted += (_, _) => secondSubscriberCalls++; + var sut = new EventTimelineEmitter(r.Inner, stream); + + sut.Emit( + new Dictionary { ["eventType"] = "enrollment_complete" }, + State(), + At); + + // Emit neither threw (would surface as an EffectRunner retry = duplicated event) + // nor skipped the remaining subscribers. + Assert.Equal(1, r.Transport.EnqueueCount); + Assert.Equal(1, secondSubscriberCalls); + } + [Fact] public void Phase_parameter_with_invalid_value_falls_back_to_Unknown() { diff --git a/src/Agent/AutopilotMonitor.Agent.V2.Core/Orchestration/DefaultComponentFactory.cs b/src/Agent/AutopilotMonitor.Agent.V2.Core/Orchestration/DefaultComponentFactory.cs index 343fd976..90413b27 100644 --- a/src/Agent/AutopilotMonitor.Agent.V2.Core/Orchestration/DefaultComponentFactory.cs +++ b/src/Agent/AutopilotMonitor.Agent.V2.Core/Orchestration/DefaultComponentFactory.cs @@ -97,7 +97,8 @@ namespace AutopilotMonitor.Agent.V2.Core.Orchestration IReadOnlyCollection whiteGloveSealingPatternIds, ISignalIngressSink ingress, IClock clock, - Transport.Telemetry.ITelemetrySpool? telemetrySpool) + Transport.Telemetry.ITelemetrySpool? telemetrySpool, + TimelineEventStream? timelineEvents = null) { if (string.IsNullOrEmpty(sessionId)) throw new ArgumentException("SessionId required.", nameof(sessionId)); if (string.IsNullOrEmpty(tenantId)) throw new ArgumentException("TenantId required.", nameof(tenantId)); @@ -471,8 +472,10 @@ namespace AutopilotMonitor.Agent.V2.Core.Orchestration if (officeHost != null) hosts.Add(officeHost); // M4.6.δ — Gather-rules runtime executor. Runs "startup" rules once the agent is up - // and "interval" rules on their timers; "phase_change" / "on_event" rules are driven - // by the host's SignalIngress.SignalPosted subscription (MON-A1). + // and "interval" rules on their timers; "phase_change" / "phase_exit" / "on_event" + // rules are driven by the post-reduce TimelineEventStream (emitted timeline events), + // NOT the raw SignalPosted stream — the raw phase can run minutes ahead of the + // engine-declared one (RealmJoin gate, session 32312a32). See TimelineEventStream doc. if (_remoteConfig.GatherRules != null && _remoteConfig.GatherRules.Count > 0) { hosts.Add(new GatherRuleExecutorHost( @@ -484,7 +487,8 @@ namespace AutopilotMonitor.Agent.V2.Core.Orchestration rules: _remoteConfig.GatherRules, imeLogPathOverride: _agentConfig.ImeLogPathOverride, unrestrictedMode: _agentConfig.UnrestrictedMode, - gatherDebugLogPath: _agentConfig.GatherRuleDebugLogPath)); + gatherDebugLogPath: _agentConfig.GatherRuleDebugLogPath, + timelineEvents: timelineEvents)); } else if (!string.IsNullOrEmpty(_agentConfig.GatherRuleDebugLogPath)) { diff --git a/src/Agent/AutopilotMonitor.Agent.V2.Core/Orchestration/EnrollmentOrchestrator.cs b/src/Agent/AutopilotMonitor.Agent.V2.Core/Orchestration/EnrollmentOrchestrator.cs index 0cef88d2..3dcaf78e 100644 --- a/src/Agent/AutopilotMonitor.Agent.V2.Core/Orchestration/EnrollmentOrchestrator.cs +++ b/src/Agent/AutopilotMonitor.Agent.V2.Core/Orchestration/EnrollmentOrchestrator.cs @@ -107,6 +107,7 @@ namespace AutopilotMonitor.Agent.V2.Core.Orchestration private TelemetrySignalEmitter? _signalEmitter; private TelemetryTransitionEmitter? _transitionEmitter; private EventTimelineEmitter? _timelineEmitter; + private TimelineEventStream? _timelineEvents; private BackPressureEventObserver? _backPressureObserver; private IDeadlineScheduler? _scheduler; private ClassifierRegistry? _classifierRegistry; @@ -413,7 +414,12 @@ namespace AutopilotMonitor.Agent.V2.Core.Orchestration _transport, _eventSequenceCounter, _sessionId, _tenantId, _traceEventsEnabled); _signalEmitter = new TelemetrySignalEmitter(_transport, _sessionId, _tenantId); _transitionEmitter = new TelemetryTransitionEmitter(_transport, _sessionId, _tenantId); - _timelineEmitter = new EventTimelineEmitter(eventEmitter); + // Post-reduce timeline observer: EventTimelineEmitter publishes every emitted event's + // (eventType, phase) here. Fed to the collector-host factory at step 14 so gather-rule + // phase_change/on_event triggers key off the ENGINE-REDUCED timeline (RealmJoin gate + // et al. respected) instead of raw pre-reduce signals — see TimelineEventStream doc. + _timelineEvents = new TimelineEventStream(); + _timelineEmitter = new EventTimelineEmitter(eventEmitter, _timelineEvents); _backPressureObserver = new BackPressureEventObserver(eventEmitter, _clock); // 5) Deadlines + Classifiers. @@ -653,7 +659,8 @@ namespace AutopilotMonitor.Agent.V2.Core.Orchestration { var surfaces = _componentFactory.CreateCollectorHosts( _sessionId, _tenantId, _logger, _whiteGloveSealingPatternIds, - ingress: _ingress, clock: _clock, telemetrySpool: _spool); + ingress: _ingress, clock: _clock, telemetrySpool: _spool, + timelineEvents: _timelineEvents); CollectorSurfaces = surfaces; _collectorHosts = surfaces.Hosts; foreach (var host in _collectorHosts) diff --git a/src/Agent/AutopilotMonitor.Agent.V2.Core/Orchestration/Hosts/GatherRuleExecutorHost.cs b/src/Agent/AutopilotMonitor.Agent.V2.Core/Orchestration/Hosts/GatherRuleExecutorHost.cs index a9ed0f7c..ead5da24 100644 --- a/src/Agent/AutopilotMonitor.Agent.V2.Core/Orchestration/Hosts/GatherRuleExecutorHost.cs +++ b/src/Agent/AutopilotMonitor.Agent.V2.Core/Orchestration/Hosts/GatherRuleExecutorHost.cs @@ -4,7 +4,6 @@ using System.Collections.Generic; using System.Threading; using AutopilotMonitor.Agent.V2.Core.Logging; using AutopilotMonitor.DecisionCore.Engine; -using AutopilotMonitor.DecisionCore.Signals; using AutopilotMonitor.Shared.Models; namespace AutopilotMonitor.Agent.V2.Core.Orchestration @@ -19,17 +18,24 @@ namespace AutopilotMonitor.Agent.V2.Core.Orchestration private readonly bool _unrestrictedMode; private int _disposed; - // MON-A1: drive phase_change / on_event gather triggers from the central signal stream. - // V1 fired these from MonitoringService on every EnrollmentEvent (OnPhaseChanged(evt.Phase) - // on phase change + OnEvent(evt.EventType) per event); the V2 host previously only called - // UpdateRules, so 5 of 10 shipped rules (incl. the only enabled one, dsregcmd at - // FinalizingSetup) silently never fired. In V2 every event is an InformationalEvent signal - // carrying eventType + (optional) phase in its payload. Null on test fakes / non-SignalIngress - // sinks — then signal-triggers degrade off (startup + interval rules still run). - private readonly SignalIngress? _observableIngress; - private Action?>? _signalPostedHandler; + // MON-A1 (revised): drive phase_change / phase_exit / on_event gather triggers from the + // POST-REDUCE emitted timeline (TimelineEventStream), not from the raw signal stream. + // V1 fired these from MonitoringService on every emitted EnrollmentEvent; the first V2 + // wiring subscribed to SignalIngress.SignalPosted instead, which fires at enqueue time + // with the RAW collector payload — before the reducer has applied gates like the + // RealmJoin completion gate. Consequence (session 32312a32, rsneuffen.de): a phase_change + // rule on FinalizingSetup fired at the raw EspPhaseChanged(FinalizingSetup) signal (ESP + // exit / Hello wizard), 7 minutes before the engine declared phase_transition(FinalizingSetup) + // on the timeline — and before the RealmJoin package wrote the registry key the rule was + // built to read. The raw feed also never carried engine-emitted event types at all, so + // on_event rules on enrollment_complete (documented) could never fire. The emitted-event + // feed restores V1 semantics and matches what the UI timeline shows ("collect once when + // the enrollment reaches this phase"). Null on test fakes / standalone configurations — + // then phase/event triggers degrade off (startup + interval rules still run). + private readonly TimelineEventStream? _timelineEvents; + private Action? _timelineHandler; private readonly object _sync = new object(); - private string? _lastPhaseName; + private EnrollmentPhase _lastPhase = EnrollmentPhase.Unknown; public GatherRuleExecutorHost( string sessionId, @@ -40,7 +46,8 @@ namespace AutopilotMonitor.Agent.V2.Core.Orchestration List rules, string? imeLogPathOverride, bool unrestrictedMode = false, - string? gatherDebugLogPath = null) + string? gatherDebugLogPath = null, + TimelineEventStream? timelineEvents = null) { if (ingress == null) throw new ArgumentNullException(nameof(ingress)); if (clock == null) throw new ArgumentNullException(nameof(clock)); @@ -60,7 +67,7 @@ namespace AutopilotMonitor.Agent.V2.Core.Orchestration _executor = new Monitoring.Telemetry.Gather.GatherRuleExecutor( sessionId, tenantId, evt => post.Emit(evt), logger, imeLogPathOverride, debugLogPath: gatherDebugLogPath); - _observableIngress = ingress as SignalIngress; + _timelineEvents = timelineEvents; } public void Start() @@ -71,46 +78,43 @@ namespace AutopilotMonitor.Agent.V2.Core.Orchestration _executor.UnrestrictedMode = _unrestrictedMode; _executor.UpdateRules(_rules); - // MON-A1: observe the central signal stream so phase_change / on_event rules fire. - if (_observableIngress != null && _signalPostedHandler == null) + // MON-A1 (revised): observe the emitted timeline so phase/event triggers fire in + // step with what the timeline actually shows. + if (_timelineEvents != null && _timelineHandler == null) { - _signalPostedHandler = OnSignalPosted; - _observableIngress.SignalPosted += _signalPostedHandler; + _timelineHandler = OnTimelineEventEmitted; + _timelineEvents.EventEmitted += _timelineHandler; } _logger.Info( - $"GatherRuleExecutorHost: started with {_rules.Count} rule(s), unrestrictedMode={_unrestrictedMode}, signalTriggers={(_observableIngress != null)}."); + $"GatherRuleExecutorHost: started with {_rules.Count} rule(s), unrestrictedMode={_unrestrictedMode}, timelineTriggers={(_timelineEvents != null)}."); } /// - /// Translates posted signals into the executor's phase_change / on_event triggers (MON-A1). - /// Phase: fire when - /// the observed phase changes and parses to an (raw collector - /// phase strings that aren't enum names are ignored — they carry no gather-rule meaning). - /// Event: fire for every - /// InformationalEvent's eventType. The executor dispatches rule execution on the ThreadPool, - /// so this stays off the signal-posting hot path; it also dedups phase rules per (rule, phase). + /// Translates emitted timeline events into the executor's triggers. Phase: fire + /// when a + /// phase-declaration event (Phase != Unknown) moves the timeline to a new phase — this + /// is by construction the engine-reduced phase, so deferred transitions (RealmJoin gate) + /// fire the rules exactly when the timeline shows them. Event: fire + /// for every emitted + /// event's type — including engine-emitted types (enrollment_complete, phase_transition, + /// realmjoin_resolved, …) that never existed on the raw signal stream. The executor + /// dispatches rule execution on the ThreadPool, so this stays off the ingress worker's + /// effect path; it also dedups phase rules per (rule, phase). /// - private void OnSignalPosted(DecisionSignalKind kind, IReadOnlyDictionary? payload) + private void OnTimelineEventEmitted(string eventType, EnrollmentPhase phase) { - if (payload == null) return; - try { lock (_sync) { - if (payload.TryGetValue(SignalPayloadKeys.EspPhase, out var phaseName) - && !string.IsNullOrEmpty(phaseName) - && !string.Equals(phaseName, _lastPhaseName, StringComparison.OrdinalIgnoreCase) - && Enum.TryParse(phaseName, ignoreCase: true, out var phase)) + if (phase != EnrollmentPhase.Unknown && phase != _lastPhase) { - _lastPhaseName = phaseName; + _lastPhase = phase; _executor.OnPhaseChanged(phase); } - if (kind == DecisionSignalKind.InformationalEvent - && payload.TryGetValue(SignalPayloadKeys.EventType, out var eventType) - && !string.IsNullOrEmpty(eventType)) + if (!string.IsNullOrEmpty(eventType)) { _executor.OnEvent(eventType); } @@ -118,17 +122,17 @@ namespace AutopilotMonitor.Agent.V2.Core.Orchestration } catch (Exception ex) { - _logger.Verbose($"GatherRuleExecutorHost: signal-trigger dispatch failed: {ex.Message}"); + _logger.Verbose($"GatherRuleExecutorHost: timeline-trigger dispatch failed: {ex.Message}"); } } public void Stop() { - if (_observableIngress != null && _signalPostedHandler != null) + if (_timelineEvents != null && _timelineHandler != null) { - try { _observableIngress.SignalPosted -= _signalPostedHandler; } + try { _timelineEvents.EventEmitted -= _timelineHandler; } catch { /* best-effort unsubscribe during shutdown */ } - _signalPostedHandler = null; + _timelineHandler = null; } // GatherRuleExecutor is IDisposable; no explicit Stop beyond unsubscribe. Rely on Dispose. } diff --git a/src/Agent/AutopilotMonitor.Agent.V2.Core/Orchestration/IComponentFactory.cs b/src/Agent/AutopilotMonitor.Agent.V2.Core/Orchestration/IComponentFactory.cs index 66d0ca64..19d3aece 100644 --- a/src/Agent/AutopilotMonitor.Agent.V2.Core/Orchestration/IComponentFactory.cs +++ b/src/Agent/AutopilotMonitor.Agent.V2.Core/Orchestration/IComponentFactory.cs @@ -39,6 +39,13 @@ namespace AutopilotMonitor.Agent.V2.Core.Orchestration /// spool.pendingItemCount / spool.fileSizeBytes. Nullable — fakes and /// spool-less configurations pass null and the metrics fields are simply absent. /// + /// + /// flows into the GatherRuleExecutorHost so + /// phase_change / phase_exit / on_event gather triggers key off the + /// ENGINE-REDUCED emitted timeline (post-reduce, RealmJoin gate respected) instead of raw + /// pre-reduce signals — see . Nullable — fakes pass null + /// and the host's phase/event triggers degrade off (startup + interval rules still run). + /// /// CollectorSurfaces CreateCollectorHosts( string sessionId, @@ -47,6 +54,7 @@ namespace AutopilotMonitor.Agent.V2.Core.Orchestration IReadOnlyCollection whiteGloveSealingPatternIds, ISignalIngressSink ingress, IClock clock, - Transport.Telemetry.ITelemetrySpool? telemetrySpool); + Transport.Telemetry.ITelemetrySpool? telemetrySpool, + TimelineEventStream? timelineEvents = null); } } diff --git a/src/Agent/AutopilotMonitor.Agent.V2.Core/Orchestration/TimelineEventStream.cs b/src/Agent/AutopilotMonitor.Agent.V2.Core/Orchestration/TimelineEventStream.cs new file mode 100644 index 00000000..7b6daa3b --- /dev/null +++ b/src/Agent/AutopilotMonitor.Agent.V2.Core/Orchestration/TimelineEventStream.cs @@ -0,0 +1,55 @@ +#nullable enable +using System; +using AutopilotMonitor.Shared.Models; + +namespace AutopilotMonitor.Agent.V2.Core.Orchestration +{ + /// + /// Post-reduce observer surface for emitted timeline events: + /// publishes every it emits as an (eventType, phase) pair, + /// AFTER the decision engine has reduced the originating signal and its effects have been + /// executed. + /// + /// This is deliberately different from , which fires + /// at enqueue time on the producer thread and therefore only ever carries the RAW collector + /// payload — before the reducer has applied gates like the RealmJoin completion gate that can + /// defer or suppress a phase transition. Consumers that must agree with what the timeline + /// actually shows (the gather-rule phase/event triggers) subscribe here, not to the raw + /// signal stream. See session 32312a32: a phase_change gather rule on FinalizingSetup + /// fired at the raw EspPhaseChanged(FinalizingSetup) signal (ESP exit), 7 minutes + /// before the engine — deferred behind the RealmJoin gate — actually declared + /// phase_transition(FinalizingSetup) on the timeline. + /// + /// + /// Threading: runs on the ingress worker thread (the effect + /// execution path). Subscribers must stay fast and must not block; subscriber exceptions are + /// swallowed per handler so the emit path can never throw (an exception here would surface + /// as an EffectRunner transient retry and duplicate the emitted event). + /// + /// + public sealed class TimelineEventStream + { + /// + /// Raised once per emitted timeline event with its + /// and parsed ( + /// for the vast majority of events — only phase-declaration events carry a real phase, + /// which is exactly the set the UI timeline groups on). + /// + public event Action? EventEmitted; + + internal void Publish(string eventType, EnrollmentPhase phase) + { + var handler = EventEmitted; + if (handler == null) return; + + foreach (Action subscriber in handler.GetInvocationList()) + { + try { subscriber(eventType, phase); } + catch + { + // Subscriber isolation — see class doc: the emit path must never throw. + } + } + } + } +} diff --git a/src/Agent/AutopilotMonitor.Agent.V2.Core/Telemetry/Events/EventTimelineEmitter.cs b/src/Agent/AutopilotMonitor.Agent.V2.Core/Telemetry/Events/EventTimelineEmitter.cs index 75819481..74a397f6 100644 --- a/src/Agent/AutopilotMonitor.Agent.V2.Core/Telemetry/Events/EventTimelineEmitter.cs +++ b/src/Agent/AutopilotMonitor.Agent.V2.Core/Telemetry/Events/EventTimelineEmitter.cs @@ -81,10 +81,12 @@ namespace AutopilotMonitor.Agent.V2.Core.Telemetry.Events internal const string ReasonParamKey = "reason"; private readonly TelemetryEventEmitter _emitter; + private readonly TimelineEventStream? _timelineEvents; - public EventTimelineEmitter(TelemetryEventEmitter emitter) + public EventTimelineEmitter(TelemetryEventEmitter emitter, TimelineEventStream? timelineEvents = null) { _emitter = emitter ?? throw new ArgumentNullException(nameof(emitter)); + _timelineEvents = timelineEvents; } public void Emit( @@ -142,6 +144,11 @@ namespace AutopilotMonitor.Agent.V2.Core.Telemetry.Events }; _emitter.Emit(evt); + + // Post-reduce observer surface: published AFTER the transport enqueue so subscribers + // (gather-rule phase/event triggers) see exactly the ordering that reaches the wire. + // Publish never throws (subscriber isolation inside the stream). + _timelineEvents?.Publish(evt.EventType, evt.Phase); } ///