diff --git a/src/Agent/AutopilotMonitor.Agent.V2.Core.Tests/Orchestration/EnrollmentOrchestratorEventBridgeTests.cs b/src/Agent/AutopilotMonitor.Agent.V2.Core.Tests/Orchestration/EnrollmentOrchestratorEventBridgeTests.cs
new file mode 100644
index 00000000..979246fc
--- /dev/null
+++ b/src/Agent/AutopilotMonitor.Agent.V2.Core.Tests/Orchestration/EnrollmentOrchestratorEventBridgeTests.cs
@@ -0,0 +1,282 @@
+using System;
+using System.Collections.Generic;
+using System.IO;
+using System.Threading;
+using AutopilotMonitor.Agent.V2.Core.Logging;
+using AutopilotMonitor.Agent.V2.Core.Orchestration;
+using AutopilotMonitor.Agent.V2.Core.Tests.Harness;
+using AutopilotMonitor.Agent.V2.Core.Tests.Transport;
+using AutopilotMonitor.Agent.V2.Core.Transport.Telemetry;
+using AutopilotMonitor.DecisionCore.Classifiers;
+using AutopilotMonitor.Shared.Models;
+using Xunit;
+
+#pragma warning disable xUnit1031
+
+namespace AutopilotMonitor.Agent.V2.Core.Tests.Orchestration
+{
+ public sealed class EnrollmentOrchestratorEventBridgeTests
+ {
+ private static DateTime At => new DateTime(2026, 4, 20, 10, 0, 0, DateTimeKind.Utc);
+
+ /// Test host that lets a test manually fire enrollment events.
+ private sealed class FakeCollectorHost : ICollectorHost
+ {
+ private readonly Action _sink;
+ public string Name { get; }
+ public int StartCalls { get; private set; }
+ public int StopCalls { get; private set; }
+ public int DisposeCalls { get; private set; }
+
+ public FakeCollectorHost(string name, Action sink)
+ {
+ Name = name;
+ _sink = sink;
+ }
+
+ public void Start() => StartCalls++;
+ public void Stop() => StopCalls++;
+ public void Dispose() => DisposeCalls++;
+
+ public void Emit(EnrollmentEvent evt) => _sink(evt);
+ }
+
+ private sealed class FakeComponentFactory : IComponentFactory
+ {
+ public Action? CapturedSink { get; private set; }
+ public readonly List Hosts = new List();
+ public readonly IReadOnlyList HostNames;
+
+ public FakeComponentFactory(params string[] hostNames)
+ {
+ HostNames = hostNames.Length == 0
+ ? new[] { "HelloTracker", "ShellCoreTracker", "ProvisioningStatusTracker", "StallProbeCollector", "ModernDeploymentTracker" }
+ : hostNames;
+ }
+
+ public IReadOnlyList CreateCollectorHosts(
+ string sessionId,
+ string tenantId,
+ AgentLogger logger,
+ Action onEnrollmentEvent)
+ {
+ CapturedSink = onEnrollmentEvent;
+ foreach (var name in HostNames)
+ {
+ Hosts.Add(new FakeCollectorHost(name, onEnrollmentEvent));
+ }
+ return Hosts;
+ }
+ }
+
+ private sealed class Rig : IDisposable
+ {
+ public TempDirectory Tmp { get; } = new TempDirectory();
+ public VirtualClock Clock { get; } = new VirtualClock(At);
+ public AgentLogger Logger { get; }
+ public FakeBackendTelemetryUploader Uploader { get; } = new FakeBackendTelemetryUploader();
+ public List Classifiers { get; } = new List();
+ public FakeComponentFactory Factory { get; }
+ public string StateDir { get; }
+ public string TransportDir { get; }
+
+ public Rig(FakeComponentFactory? factory = null)
+ {
+ Logger = new AgentLogger(Tmp.Path, AgentLogLevel.Info);
+ StateDir = Path.Combine(Tmp.Path, "State");
+ TransportDir = Path.Combine(Tmp.Path, "Transport");
+ Factory = factory ?? new FakeComponentFactory();
+ }
+
+ public EnrollmentOrchestrator Build() =>
+ new EnrollmentOrchestrator(
+ sessionId: "S1",
+ tenantId: "T1",
+ stateDirectory: StateDir,
+ transportDirectory: TransportDir,
+ clock: Clock,
+ logger: Logger,
+ uploader: Uploader,
+ classifiers: Classifiers,
+ componentFactory: Factory,
+ drainInterval: TimeSpan.FromDays(1),
+ terminalDrainTimeout: TimeSpan.FromSeconds(2));
+
+ public void Dispose() => Tmp.Dispose();
+ }
+
+ private static EnrollmentEvent Evt(string eventType) => new EnrollmentEvent
+ {
+ EventType = eventType,
+ Severity = EventSeverity.Info,
+ Source = "test",
+ Phase = EnrollmentPhase.Unknown,
+ Timestamp = At,
+ };
+
+ // ========================================================================= Wiring
+
+ [Fact]
+ public void Start_without_factory_does_not_throw_and_spawns_no_hosts()
+ {
+ using var tmp = new TempDirectory();
+ var logger = new AgentLogger(tmp.Path, AgentLogLevel.Info);
+ var uploader = new FakeBackendTelemetryUploader();
+ var sut = new EnrollmentOrchestrator(
+ sessionId: "S1",
+ tenantId: "T1",
+ stateDirectory: Path.Combine(tmp.Path, "State"),
+ transportDirectory: Path.Combine(tmp.Path, "Transport"),
+ clock: new VirtualClock(At),
+ logger: logger,
+ uploader: uploader,
+ classifiers: new List(),
+ componentFactory: null,
+ drainInterval: TimeSpan.FromDays(1),
+ terminalDrainTimeout: TimeSpan.FromSeconds(2));
+
+ sut.Start();
+ sut.Stop(); // no exception
+ }
+
+ [Fact]
+ public void Start_with_factory_creates_all_hosts_and_captures_sink()
+ {
+ using var rig = new Rig();
+ var sut = rig.Build();
+
+ sut.Start();
+
+ Assert.NotNull(rig.Factory.CapturedSink);
+ Assert.Equal(5, rig.Factory.Hosts.Count);
+ Assert.All(rig.Factory.Hosts, h => Assert.Equal(1, h.StartCalls));
+
+ sut.Stop();
+ }
+
+ [Fact]
+ public void Emit_from_host_reaches_telemetry_transport()
+ {
+ using var rig = new Rig();
+ var sut = rig.Build();
+ sut.Start();
+
+ var hello = rig.Factory.Hosts.Find(h => h.Name == "HelloTracker")!;
+ hello.Emit(Evt("hello_completed"));
+
+ var spool = GetSpool(sut);
+ var pending = spool.Peek(10);
+ Assert.Single(pending);
+ Assert.Equal(TelemetryItemKind.Event, pending[0].Kind);
+ Assert.Contains("hello_completed", pending[0].PayloadJson);
+
+ sut.Stop();
+ }
+
+ [Fact]
+ public void Emit_from_each_of_five_hosts_produces_five_transport_items()
+ {
+ using var rig = new Rig();
+ var sut = rig.Build();
+ sut.Start();
+
+ foreach (var host in rig.Factory.Hosts)
+ {
+ host.Emit(Evt($"event_from_{host.Name}"));
+ }
+
+ var pending = GetSpool(sut).Peek(10);
+ Assert.Equal(5, pending.Count);
+ foreach (var host in rig.Factory.Hosts)
+ {
+ Assert.Contains(pending, p => p.PayloadJson.Contains($"event_from_{host.Name}"));
+ }
+
+ sut.Stop();
+ }
+
+ [Fact]
+ public void Modern_deployment_host_emits_event_bridge_only_no_decision_signal()
+ {
+ using var rig = new Rig();
+ var sut = rig.Build();
+ sut.Start();
+
+ var modern = rig.Factory.Hosts.Find(h => h.Name == "ModernDeploymentTracker")!;
+ modern.Emit(Evt("modern_deployment_log"));
+
+ // Event reached transport (bridge works).
+ var pending = GetSpool(sut).Peek(10);
+ Assert.Single(pending);
+ Assert.Contains("modern_deployment_log", pending[0].PayloadJson);
+
+ // No decision-signal was posted — signal-log is empty (no SignalAdapter wraps
+ // ModernDeploymentTracker by design, Plan §4.x M4.3-Erkenntnis).
+ var signalLog = GetSignalLog(sut);
+ Assert.Empty(signalLog.ReadAll());
+
+ sut.Stop();
+ }
+
+ [Fact]
+ public void Emitter_exception_does_not_propagate_to_host_thread()
+ {
+ // Simulate: TelemetryEventEmitter throws (e.g. transport disposed mid-emit).
+ // Build a shared sink that throws on demand; wrap into a host that uses it directly.
+ // Easier path: dispose the transport post-Start so Enqueue throws, then fire an event.
+ using var rig = new Rig();
+ var sut = rig.Build();
+ sut.Start();
+
+ // Force a transport-level exception by disposing it mid-run.
+ GetTransport(sut).Dispose();
+
+ var hello = rig.Factory.Hosts.Find(h => h.Name == "HelloTracker")!;
+
+ // Must not throw — orchestrator wraps the emitter call in try/catch.
+ var ex = Record.Exception(() => hello.Emit(Evt("hello_after_dispose")));
+ Assert.Null(ex);
+
+ sut.Stop();
+ }
+
+ [Fact]
+ public void Stop_stops_and_disposes_all_hosts_in_creation_order()
+ {
+ using var rig = new Rig();
+ var sut = rig.Build();
+ sut.Start();
+
+ sut.Stop();
+
+ Assert.All(rig.Factory.Hosts, h => Assert.Equal(1, h.StopCalls));
+ Assert.All(rig.Factory.Hosts, h => Assert.Equal(1, h.DisposeCalls));
+ }
+
+ // ========================================================================= Helpers
+
+ private static TelemetrySpool GetSpool(EnrollmentOrchestrator sut)
+ {
+ var field = typeof(EnrollmentOrchestrator).GetField(
+ "_spool",
+ System.Reflection.BindingFlags.Instance | System.Reflection.BindingFlags.NonPublic);
+ return (TelemetrySpool)field!.GetValue(sut)!;
+ }
+
+ private static TelemetryUploadOrchestrator GetTransport(EnrollmentOrchestrator sut)
+ {
+ var field = typeof(EnrollmentOrchestrator).GetField(
+ "_transport",
+ System.Reflection.BindingFlags.Instance | System.Reflection.BindingFlags.NonPublic);
+ return (TelemetryUploadOrchestrator)field!.GetValue(sut)!;
+ }
+
+ private static AutopilotMonitor.Agent.V2.Core.Persistence.ISignalLogWriter GetSignalLog(EnrollmentOrchestrator sut)
+ {
+ var field = typeof(EnrollmentOrchestrator).GetField(
+ "_signalLog",
+ System.Reflection.BindingFlags.Instance | System.Reflection.BindingFlags.NonPublic);
+ return (AutopilotMonitor.Agent.V2.Core.Persistence.ISignalLogWriter)field!.GetValue(sut)!;
+ }
+ }
+}
diff --git a/src/Agent/AutopilotMonitor.Agent.V2.Core/Orchestration/EnrollmentOrchestrator.cs b/src/Agent/AutopilotMonitor.Agent.V2.Core/Orchestration/EnrollmentOrchestrator.cs
index e08a53e3..22c19088 100644
--- a/src/Agent/AutopilotMonitor.Agent.V2.Core/Orchestration/EnrollmentOrchestrator.cs
+++ b/src/Agent/AutopilotMonitor.Agent.V2.Core/Orchestration/EnrollmentOrchestrator.cs
@@ -70,6 +70,7 @@ namespace AutopilotMonitor.Agent.V2.Core.Orchestration
private readonly AgentLogger _logger;
private readonly IBackendTelemetryUploader _uploader;
private readonly IReadOnlyList _classifiers;
+ private readonly IComponentFactory? _componentFactory;
private readonly int _channelCapacity;
private readonly int _quarantineThreshold;
private readonly TimeSpan _drainInterval;
@@ -94,6 +95,7 @@ namespace AutopilotMonitor.Agent.V2.Core.Orchestration
private DecisionStepProcessor? _processor;
private SessionTraceOrdinalProvider? _traceCounter;
private SignalIngress? _ingress;
+ private IReadOnlyList? _collectorHosts;
private EventHandler? _deadlineBridge;
@@ -119,6 +121,7 @@ namespace AutopilotMonitor.Agent.V2.Core.Orchestration
AgentLogger logger,
IBackendTelemetryUploader uploader,
IEnumerable classifiers,
+ IComponentFactory? componentFactory = null,
int channelCapacity = SignalIngress.DefaultChannelCapacity,
int quarantineThreshold = DecisionStepProcessor.DefaultQuarantineThreshold,
TimeSpan? drainInterval = null,
@@ -145,6 +148,7 @@ namespace AutopilotMonitor.Agent.V2.Core.Orchestration
list.Add(c);
}
_classifiers = list;
+ _componentFactory = componentFactory;
_channelCapacity = channelCapacity;
_quarantineThreshold = quarantineThreshold;
@@ -265,10 +269,31 @@ namespace AutopilotMonitor.Agent.V2.Core.Orchestration
_deadlineBridge = OnDeadlineFired;
_scheduler.Fired += _deadlineBridge;
- // 13) Ingress-Worker starten.
+ // 13) Ingress-Worker starten (vor Collectors — sonst race-prone).
_ingress.Start();
- // 14) Periodischer Drain-Loop (Fire-and-forget).
+ // 14) Collector-Hosts via Factory — onEnrollmentEvent bridged Telemetry-Events
+ // an den zentralen TelemetryEventEmitter. Exception-Swallow damit ein
+ // Collector-Thread nicht am Emitter-Fehler stirbt (Plan §2.7a).
+ if (_componentFactory != null)
+ {
+ var safeSink = BuildTelemetryEventSink();
+ _collectorHosts = _componentFactory.CreateCollectorHosts(_sessionId, _tenantId, _logger, safeSink);
+ foreach (var host in _collectorHosts)
+ {
+ try
+ {
+ host.Start();
+ _logger.Debug($"EnrollmentOrchestrator: started collector host '{host.Name}'.");
+ }
+ catch (Exception ex)
+ {
+ _logger.Error($"EnrollmentOrchestrator: failed to start collector host '{host.Name}'.", ex);
+ }
+ }
+ }
+
+ // 15) Periodischer Drain-Loop (Fire-and-forget).
_drainCts = new CancellationTokenSource();
_drainTask = Task.Run(() => DrainLoopAsync(_drainCts.Token));
@@ -285,6 +310,16 @@ namespace AutopilotMonitor.Agent.V2.Core.Orchestration
_logger.Info("EnrollmentOrchestrator: stopping.");
+ // 0) Collector-Hosts stoppen — keine neuen Events / DecisionSignals aus dem Feld.
+ if (_collectorHosts != null)
+ {
+ foreach (var host in _collectorHosts)
+ {
+ try { host.Stop(); }
+ catch (Exception ex) { _logger.Warning($"EnrollmentOrchestrator: host '{host.Name}' stop failed: {ex.Message}"); }
+ }
+ }
+
// 1) Scheduler.Fired-Handler abmelden — keine neuen DeadlineFired-Signals mehr.
try
{
@@ -333,6 +368,13 @@ namespace AutopilotMonitor.Agent.V2.Core.Orchestration
catch (Exception ex) { _logger.Warning($"EnrollmentOrchestrator: final snapshot save failed: {ex.Message}"); }
// 7) Dispose aller Disposables.
+ if (_collectorHosts != null)
+ {
+ foreach (var host in _collectorHosts)
+ {
+ try { host.Dispose(); } catch { /* best-effort */ }
+ }
+ }
try { _transport?.Dispose(); } catch { /* best-effort */ }
try { _ingress?.Dispose(); } catch { /* best-effort */ }
try { _drainCts?.Dispose(); } catch { /* best-effort */ }
@@ -376,6 +418,27 @@ namespace AutopilotMonitor.Agent.V2.Core.Orchestration
if (!Directory.Exists(_transportDirectory)) Directory.CreateDirectory(_transportDirectory);
}
+ ///
+ /// Shared callback that collector hosts use for onEventCollected. Exception-safe:
+ /// an emitter failure (disk hiccup, transport closed) must never kill a collector thread.
+ ///
+ private Action BuildTelemetryEventSink()
+ {
+ return evt =>
+ {
+ if (evt == null || _eventEmitter == null) return;
+ try
+ {
+ _eventEmitter.Emit(evt);
+ }
+ catch (Exception ex)
+ {
+ _logger.Error(
+ $"EnrollmentOrchestrator: telemetry emit failed for event '{evt.EventType}'.", ex);
+ }
+ };
+ }
+
private void OnDeadlineFired(object? sender, DeadlineFiredEventArgs e)
{
if (Volatile.Read(ref _stopRequested) == 1 || _ingress == null) return;
diff --git a/src/Agent/AutopilotMonitor.Agent.V2.Core/Orchestration/ICollectorHost.cs b/src/Agent/AutopilotMonitor.Agent.V2.Core/Orchestration/ICollectorHost.cs
new file mode 100644
index 00000000..2f8cbdd8
--- /dev/null
+++ b/src/Agent/AutopilotMonitor.Agent.V2.Core/Orchestration/ICollectorHost.cs
@@ -0,0 +1,34 @@
+#nullable enable
+using System;
+
+namespace AutopilotMonitor.Agent.V2.Core.Orchestration
+{
+ ///
+ /// Lifecycle-Abstraktion für einen einzelnen Collector im V2-Runtime.
+ /// Plan §2.1 / §4.x M4.4.5.c.
+ ///
+ /// Ein Host wrappt einen konkreten Collector (z.B. HelloTracker,
+ /// ShellCoreTracker, ModernDeploymentTracker) und kapselt Start/Stop.
+ /// Die Action<EnrollmentEvent>-Callback für onEventCollected wird bei
+ /// der Host-Erstellung über die injiziert — der
+ /// Orchestrator übergibt dort
+ /// als Sink.
+ ///
+ ///
+ /// SignalAdapter-Wiring: separat — kann die
+ /// SignalAdapter-Instanzen mit-konstruieren und selbst an ihre Collector-Events
+ /// subscriben. Der Orchestrator muss die Adapter-Seam nicht kennen.
+ ///
+ ///
+ public interface ICollectorHost : IDisposable
+ {
+ /// Stabiler Name für Observability/Logs (z.B. "HelloTracker").
+ string Name { get; }
+
+ /// Startet Event-Subscriptions + Background-Worker des gewrappten Collectors.
+ void Start();
+
+ /// Stoppt und gibt Ressourcen frei. Idempotent.
+ void Stop();
+ }
+}
diff --git a/src/Agent/AutopilotMonitor.Agent.V2.Core/Orchestration/IComponentFactory.cs b/src/Agent/AutopilotMonitor.Agent.V2.Core/Orchestration/IComponentFactory.cs
new file mode 100644
index 00000000..af98ceaa
--- /dev/null
+++ b/src/Agent/AutopilotMonitor.Agent.V2.Core/Orchestration/IComponentFactory.cs
@@ -0,0 +1,40 @@
+#nullable enable
+using System;
+using System.Collections.Generic;
+using AutopilotMonitor.Agent.V2.Core.Logging;
+using AutopilotMonitor.Shared.Models;
+
+namespace AutopilotMonitor.Agent.V2.Core.Orchestration
+{
+ ///
+ /// Factory-Seam für Collector-Hosts. Plan §4.x M4.4.5.c.
+ ///
+ /// Der delegiert an diese Factory, weil die konkreten
+ /// Collector-Ctoren heavy Production-Deps (Event-Log-Watcher, WMI, Registry) haben — Tests
+ /// liefern eine Fake-Factory, Prod liefert eine Default-Implementation, die HelloTracker,
+ /// ShellCoreTracker, ProvisioningStatusTracker, StallProbeCollector und
+ /// ModernDeploymentTracker baut und deren onEventCollected-Callbacks an den
+ /// übergebenen bindet.
+ ///
+ ///
+ /// ModernDeploymentTracker-Spezifikum: der einzige Kollektor ohne SignalAdapter —
+ /// bridged nur Events, keine DecisionSignals. Default-Factory-Impls dürfen ihn aufbauen,
+ /// Tests brauchen ihn nicht zwingend.
+ ///
+ ///
+ public interface IComponentFactory
+ {
+ ///
+ /// Baut alle Collector-Hosts und bindet deren Event-Emission an
+ /// (typischerweise
+ /// ).
+ /// Exceptions im Callback darf die Factory NICHT weiterreichen — der Orchestrator
+ /// wrappt das mit einem Exception-Swallow, damit ein Collector-Thread nicht stirbt.
+ ///
+ IReadOnlyList CreateCollectorHosts(
+ string sessionId,
+ string tenantId,
+ AgentLogger logger,
+ Action onEnrollmentEvent);
+ }
+}