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 <noreply@anthropic.com>
2026-10-24 04:32:32 +00:00 · 2026-08-04 14:44:35 +02:00
parent ed5fcf3dc3
commit eb94c612e7
12 changed files with 417 additions and 52 deletions
@@ -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).
@@ -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)
@@ -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<string> 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));
@@ -27,7 +27,8 @@ namespace AutopilotMonitor.Agent.V2.Core.Tests.Orchestration
                IReadOnlyCollection<string> 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<ICollectorHost>());
@@ -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
+{
+    /// <summary>
+    /// GatherRuleExecutorHost feeds its phase_change / phase_exit / on_event triggers from the
+    /// POST-REDUCE <see cref="TimelineEventStream"/> (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.
+    /// </summary>
+    [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<string> _eventTypes = new List<string>();
+            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<string, string>? 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<GatherRule>(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"));
+        }
+    }
+}
@@ -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<string, string>
+                {
+                    ["eventType"] = "phase_transition",
+                    ["phase"] = "FinalizingSetup",
+                },
+                State(),
+                At);
+            sut.Emit(
+                new Dictionary<string, string> { ["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<string, string> { ["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()
        {
@@ -97,7 +97,8 @@ namespace AutopilotMonitor.Agent.V2.Core.Orchestration
            IReadOnlyCollection<string> 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))
            {
@@ -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)
@@ -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<DecisionSignalKind, IReadOnlyDictionary<string, string>?>? _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<string, EnrollmentPhase>? _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<GatherRule> 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)}.");
        }

        /// <summary>
-        /// Translates posted signals into the executor's phase_change / on_event triggers (MON-A1).
-        /// Phase: fire <see cref="Monitoring.Telemetry.Gather.GatherRuleExecutor.OnPhaseChanged"/> when
-        /// the observed phase changes and parses to an <see cref="EnrollmentPhase"/> (raw collector
-        /// phase strings that aren't enum names are ignored — they carry no gather-rule meaning).
-        /// Event: fire <see cref="Monitoring.Telemetry.Gather.GatherRuleExecutor.OnEvent"/> for every
-        /// InformationalEvent's eventType. The executor dispatches rule execution on the ThreadPool,
-        /// so this stays off the signal-posting hot path; it also dedups phase rules per (rule, phase).
+        /// Translates emitted timeline events into the executor's triggers. Phase: fire
+        /// <see cref="Monitoring.Telemetry.Gather.GatherRuleExecutor.OnPhaseChanged"/> 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
+        /// <see cref="Monitoring.Telemetry.Gather.GatherRuleExecutor.OnEvent"/> 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).
        /// </summary>
-        private void OnSignalPosted(DecisionSignalKind kind, IReadOnlyDictionary<string, string>? 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<EnrollmentPhase>(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.
        }
@@ -39,6 +39,13 @@ namespace AutopilotMonitor.Agent.V2.Core.Orchestration
        /// <c>spool.pendingItemCount</c> / <c>spool.fileSizeBytes</c>. Nullable — fakes and
        /// spool-less configurations pass null and the metrics fields are simply absent.
        /// </para>
+        /// <para>
+        /// <paramref name="timelineEvents"/> flows into the <c>GatherRuleExecutorHost</c> so
+        /// <c>phase_change</c> / <c>phase_exit</c> / <c>on_event</c> gather triggers key off the
+        /// ENGINE-REDUCED emitted timeline (post-reduce, RealmJoin gate respected) instead of raw
+        /// pre-reduce signals — see <see cref="TimelineEventStream"/>. Nullable — fakes pass null
+        /// and the host's phase/event triggers degrade off (startup + interval rules still run).
+        /// </para>
        /// </summary>
        CollectorSurfaces CreateCollectorHosts(
            string sessionId,
@@ -47,6 +54,7 @@ namespace AutopilotMonitor.Agent.V2.Core.Orchestration
            IReadOnlyCollection<string> whiteGloveSealingPatternIds,
            ISignalIngressSink ingress,
            IClock clock,
-            Transport.Telemetry.ITelemetrySpool? telemetrySpool);
+            Transport.Telemetry.ITelemetrySpool? telemetrySpool,
+            TimelineEventStream? timelineEvents = null);
    }
 }
@@ -0,0 +1,55 @@
+#nullable enable
+using System;
+using AutopilotMonitor.Shared.Models;
+
+namespace AutopilotMonitor.Agent.V2.Core.Orchestration
+{
+    /// <summary>
+    /// Post-reduce observer surface for emitted timeline events: <see cref="Telemetry.Events.EventTimelineEmitter"/>
+    /// publishes every <see cref="EnrollmentEvent"/> it emits as an <c>(eventType, phase)</c> pair,
+    /// AFTER the decision engine has reduced the originating signal and its effects have been
+    /// executed.
+    /// <para>
+    /// This is deliberately different from <see cref="SignalIngress.SignalPosted"/>, 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 <c>phase_change</c> gather rule on FinalizingSetup
+    /// fired at the raw <c>EspPhaseChanged(FinalizingSetup)</c> signal (ESP exit), 7 minutes
+    /// before the engine — deferred behind the RealmJoin gate — actually declared
+    /// <c>phase_transition(FinalizingSetup)</c> on the timeline.
+    /// </para>
+    /// <para>
+    /// <b>Threading</b>: <see cref="Publish"/> 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).
+    /// </para>
+    /// </summary>
+    public sealed class TimelineEventStream
+    {
+        /// <summary>
+        /// Raised once per emitted timeline event with its <see cref="EnrollmentEvent.EventType"/>
+        /// and parsed <see cref="EnrollmentEvent.Phase"/> (<see cref="EnrollmentPhase.Unknown"/>
+        /// for the vast majority of events — only phase-declaration events carry a real phase,
+        /// which is exactly the set the UI timeline groups on).
+        /// </summary>
+        public event Action<string, EnrollmentPhase>? EventEmitted;
+
+        internal void Publish(string eventType, EnrollmentPhase phase)
+        {
+            var handler = EventEmitted;
+            if (handler == null) return;
+
+            foreach (Action<string, EnrollmentPhase> subscriber in handler.GetInvocationList())
+            {
+                try { subscriber(eventType, phase); }
+                catch
+                {
+                    // Subscriber isolation — see class doc: the emit path must never throw.
+                }
+            }
+        }
+    }
+}
@@ -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);
        }

        /// <summary>