refactor(v2core): M4.4.5.c collector-host seam + event-bridge wiring

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) <noreply@anthropic.com>
This commit is contained in:
Oliver Kieselbach
2026-04-20 20:41:59 +02:00
parent 51f9db6429
commit a72e716862
4 changed files with 421 additions and 2 deletions
@@ -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);
/// <summary>Test host that lets a test manually fire enrollment events.</summary>
private sealed class FakeCollectorHost : ICollectorHost
{
private readonly Action<EnrollmentEvent> _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<EnrollmentEvent> 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<EnrollmentEvent>? CapturedSink { get; private set; }
public readonly List<FakeCollectorHost> Hosts = new List<FakeCollectorHost>();
public readonly IReadOnlyList<string> HostNames;
public FakeComponentFactory(params string[] hostNames)
{
HostNames = hostNames.Length == 0
? new[] { "HelloTracker", "ShellCoreTracker", "ProvisioningStatusTracker", "StallProbeCollector", "ModernDeploymentTracker" }
: hostNames;
}
public IReadOnlyList<ICollectorHost> CreateCollectorHosts(
string sessionId,
string tenantId,
AgentLogger logger,
Action<EnrollmentEvent> 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<IClassifier> Classifiers { get; } = new List<IClassifier>();
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<IClassifier>(),
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)!;
}
}
}
@@ -70,6 +70,7 @@ namespace AutopilotMonitor.Agent.V2.Core.Orchestration
private readonly AgentLogger _logger;
private readonly IBackendTelemetryUploader _uploader;
private readonly IReadOnlyList<IClassifier> _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<ICollectorHost>? _collectorHosts;
private EventHandler<DeadlineFiredEventArgs>? _deadlineBridge;
@@ -119,6 +121,7 @@ namespace AutopilotMonitor.Agent.V2.Core.Orchestration
AgentLogger logger,
IBackendTelemetryUploader uploader,
IEnumerable<IClassifier> 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);
}
/// <summary>
/// Shared callback that collector hosts use for <c>onEventCollected</c>. Exception-safe:
/// an emitter failure (disk hiccup, transport closed) must never kill a collector thread.
/// </summary>
private Action<AutopilotMonitor.Shared.Models.EnrollmentEvent> 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;
@@ -0,0 +1,34 @@
#nullable enable
using System;
namespace AutopilotMonitor.Agent.V2.Core.Orchestration
{
/// <summary>
/// Lifecycle-Abstraktion für einen einzelnen Collector im V2-Runtime.
/// Plan §2.1 / §4.x M4.4.5.c.
/// <para>
/// Ein Host wrappt einen konkreten Collector (z.B. <c>HelloTracker</c>,
/// <c>ShellCoreTracker</c>, <c>ModernDeploymentTracker</c>) und kapselt Start/Stop.
/// Die <c>Action&lt;EnrollmentEvent&gt;</c>-Callback für <c>onEventCollected</c> wird bei
/// der Host-Erstellung über die <see cref="IComponentFactory"/> injiziert — der
/// Orchestrator übergibt dort <see cref="Telemetry.Events.TelemetryEventEmitter.Emit"/>
/// als Sink.
/// </para>
/// <para>
/// <b>SignalAdapter-Wiring</b>: separat — <see cref="IComponentFactory"/> kann die
/// SignalAdapter-Instanzen mit-konstruieren und selbst an ihre Collector-Events
/// subscriben. Der Orchestrator muss die Adapter-Seam nicht kennen.
/// </para>
/// </summary>
public interface ICollectorHost : IDisposable
{
/// <summary>Stabiler Name für Observability/Logs (z.B. "HelloTracker").</summary>
string Name { get; }
/// <summary>Startet Event-Subscriptions + Background-Worker des gewrappten Collectors.</summary>
void Start();
/// <summary>Stoppt und gibt Ressourcen frei. Idempotent.</summary>
void Stop();
}
}
@@ -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
{
/// <summary>
/// Factory-Seam für Collector-Hosts. Plan §4.x M4.4.5.c.
/// <para>
/// Der <see cref="EnrollmentOrchestrator"/> 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 <c>onEventCollected</c>-Callbacks an den
/// übergebenen <paramref name="onEnrollmentEvent"/> bindet.
/// </para>
/// <para>
/// <b>ModernDeploymentTracker-Spezifikum</b>: der einzige Kollektor ohne SignalAdapter —
/// bridged nur Events, keine DecisionSignals. Default-Factory-Impls dürfen ihn aufbauen,
/// Tests brauchen ihn nicht zwingend.
/// </para>
/// </summary>
public interface IComponentFactory
{
/// <summary>
/// Baut alle Collector-Hosts und bindet deren Event-Emission an
/// <paramref name="onEnrollmentEvent"/> (typischerweise
/// <see cref="Telemetry.Events.TelemetryEventEmitter.Emit"/>).
/// Exceptions im Callback darf die Factory NICHT weiterreichen — der Orchestrator
/// wrappt das mit einem Exception-Swallow, damit ein Collector-Thread nicht stirbt.
/// </summary>
IReadOnlyList<ICollectorHost> CreateCollectorHosts(
string sessionId,
string tenantId,
AgentLogger logger,
Action<EnrollmentEvent> onEnrollmentEvent);
}
}