From a72e7168627ee08f6af33a905614663d439c70ea Mon Sep 17 00:00:00 2001 From: Oliver Kieselbach Date: Mon, 20 Apr 2026 20:41:59 +0200 Subject: [PATCH] refactor(v2core): M4.4.5.c collector-host seam + event-bridge wiring MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ICollectorHost + IComponentFactory als Seam fuer Collector-Produktion: Factory liefert fertig-konstruierte Hosts mit der telemetryEmitter.Emit- Callback bereits als onEnrollmentEvent-Sink gebunden. Orchestrator ruft Start/Stop/Dispose in korrekter Reihenfolge auf. Die 5 Collectors (HelloTracker, ShellCoreTracker, ProvisioningStatusTracker, StallProbeCollector, ModernDeploymentTracker) haben heavy Production-Deps (Event-Log-Watcher, WMI, Registry) — die Factory-Seam erlaubt Prod-Impl in Program.cs / Tests nutzen Fakes. Exception-Swallow-Wrapper im Orchestrator: ein Emitter-Fail (z.B. Transport disposed) darf den Collector-Thread nicht killen. ModernDeploymentTracker explizit als bridge-only dokumentiert: kein SignalAdapter, nur Event-Bridge — validiert durch dedizierten Test, der prueft dass Events im Transport landen aber SignalLog leer bleibt. Factory optional (default null): Orchestrator ohne Factory laeuft weiter ohne Collectors (backwards-compatible zu M4.4.5.b). +7 Agent.V2.Core Tests (253 → 260 gruen). Co-Authored-By: Claude Opus 4.7 (1M context) --- .../EnrollmentOrchestratorEventBridgeTests.cs | 282 ++++++++++++++++++ .../Orchestration/EnrollmentOrchestrator.cs | 67 ++++- .../Orchestration/ICollectorHost.cs | 34 +++ .../Orchestration/IComponentFactory.cs | 40 +++ 4 files changed, 421 insertions(+), 2 deletions(-) create mode 100644 src/Agent/AutopilotMonitor.Agent.V2.Core.Tests/Orchestration/EnrollmentOrchestratorEventBridgeTests.cs create mode 100644 src/Agent/AutopilotMonitor.Agent.V2.Core/Orchestration/ICollectorHost.cs create mode 100644 src/Agent/AutopilotMonitor.Agent.V2.Core/Orchestration/IComponentFactory.cs 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); + } +}