From 51f9db642929150bbaeebc553ea4ab562105c2fc Mon Sep 17 00:00:00 2001 From: Oliver Kieselbach Date: Mon, 20 Apr 2026 20:37:22 +0200 Subject: [PATCH] refactor(v2core): M4.4.5.b enrollment orchestrator skeleton + lifecycle MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Top-Level-Lifecycle-Owner. Start() verdrahtet die Kern-Pipeline (SignalLog/Journal/Snapshot, TelemetrySpool+Uploader, EventSequence+ Emitter+Timeline+BackPressure, DeadlineScheduler+ClassifierRegistry, EffectRunner, DecisionStepProcessor, SessionTraceOrdinalProvider, SignalIngress) und verbindet Scheduler.Fired → ingress.Post-Bridge. Zirkel EffectRunner-ISignalIngressSink ↔ SignalIngress-Processor via LazyIngressSinkRelay gebrochen. Snapshot-Recovery-Vorstufe bei Start; volle Checksum/Replay-Logik kommt in M4.4.5.f. Shutdown idempotent + reverse-order: Scheduler-Unsubscribe → Scheduler dispose → Ingress.Stop → Drain-Loop cancel → Terminal-Drain → Final- Snapshot → Dispose. IQuarantineSink-Impl setzt Flag (Bewegung erst beim naechsten Start, nicht mid-run). Drain-Loop nutzt Task.Delay (Wall-Clock) statt IClock.Delay, damit VirtualClock-Tests nicht in Tight-Loop laufen. +14 Agent.V2.Core Tests (239 → 253 gruen). Collectors + SignalAdapters folgen in M4.4.5.c, mTLS-HttpClient-Factory in .d, Pattern-Injection in .e, Recovery-Pfade in .f. Co-Authored-By: Claude Opus 4.7 (1M context) --- .../EnrollmentOrchestratorTests.cs | 330 ++++++++++++ .../Orchestration/EnrollmentOrchestrator.cs | 478 ++++++++++++++++++ 2 files changed, 808 insertions(+) create mode 100644 src/Agent/AutopilotMonitor.Agent.V2.Core.Tests/Orchestration/EnrollmentOrchestratorTests.cs create mode 100644 src/Agent/AutopilotMonitor.Agent.V2.Core/Orchestration/EnrollmentOrchestrator.cs diff --git a/src/Agent/AutopilotMonitor.Agent.V2.Core.Tests/Orchestration/EnrollmentOrchestratorTests.cs b/src/Agent/AutopilotMonitor.Agent.V2.Core.Tests/Orchestration/EnrollmentOrchestratorTests.cs new file mode 100644 index 00000000..7985fb3c --- /dev/null +++ b/src/Agent/AutopilotMonitor.Agent.V2.Core.Tests/Orchestration/EnrollmentOrchestratorTests.cs @@ -0,0 +1,330 @@ +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.Persistence; +using AutopilotMonitor.Agent.V2.Core.Tests.Harness; +using AutopilotMonitor.Agent.V2.Core.Tests.Transport; +using AutopilotMonitor.DecisionCore.Classifiers; +using AutopilotMonitor.DecisionCore.Engine; +using AutopilotMonitor.DecisionCore.Serialization; +using AutopilotMonitor.DecisionCore.Signals; +using AutopilotMonitor.DecisionCore.State; +using Xunit; + +#pragma warning disable xUnit1031 // Task.Wait + SpinWait for deterministic thread coordination + +namespace AutopilotMonitor.Agent.V2.Core.Tests.Orchestration +{ + public sealed class EnrollmentOrchestratorTests + { + private static DateTime At => new DateTime(2026, 4, 20, 10, 0, 0, DateTimeKind.Utc); + + 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 string StateDir { get; } + public string TransportDir { get; } + + public Rig() + { + Logger = new AgentLogger(Tmp.Path, AgentLogLevel.Info); + StateDir = Path.Combine(Tmp.Path, "State"); + TransportDir = Path.Combine(Tmp.Path, "Transport"); + } + + public EnrollmentOrchestrator Build(TimeSpan? drainInterval = null) => + new EnrollmentOrchestrator( + sessionId: "S1", + tenantId: "T1", + stateDirectory: StateDir, + transportDirectory: TransportDir, + clock: Clock, + logger: Logger, + uploader: Uploader, + classifiers: Classifiers, + // Default drain-interval = 1 day → periodic drain won't fire during short tests. + drainInterval: drainInterval ?? TimeSpan.FromDays(1), + terminalDrainTimeout: TimeSpan.FromSeconds(2)); + + public void Dispose() => Tmp.Dispose(); + } + + // ========================================================================= Ctor + + [Fact] + public void Ctor_rejects_empty_required_strings() + { + using var rig = new Rig(); + Assert.Throws(() => new EnrollmentOrchestrator( + "", "T1", rig.StateDir, rig.TransportDir, rig.Clock, rig.Logger, rig.Uploader, rig.Classifiers)); + Assert.Throws(() => new EnrollmentOrchestrator( + "S1", "", rig.StateDir, rig.TransportDir, rig.Clock, rig.Logger, rig.Uploader, rig.Classifiers)); + Assert.Throws(() => new EnrollmentOrchestrator( + "S1", "T1", "", rig.TransportDir, rig.Clock, rig.Logger, rig.Uploader, rig.Classifiers)); + Assert.Throws(() => new EnrollmentOrchestrator( + "S1", "T1", rig.StateDir, "", rig.Clock, rig.Logger, rig.Uploader, rig.Classifiers)); + } + + [Fact] + public void Ctor_rejects_null_dependencies() + { + using var rig = new Rig(); + Assert.Throws(() => new EnrollmentOrchestrator( + "S1", "T1", rig.StateDir, rig.TransportDir, null!, rig.Logger, rig.Uploader, rig.Classifiers)); + Assert.Throws(() => new EnrollmentOrchestrator( + "S1", "T1", rig.StateDir, rig.TransportDir, rig.Clock, null!, rig.Uploader, rig.Classifiers)); + Assert.Throws(() => new EnrollmentOrchestrator( + "S1", "T1", rig.StateDir, rig.TransportDir, rig.Clock, rig.Logger, null!, rig.Classifiers)); + Assert.Throws(() => new EnrollmentOrchestrator( + "S1", "T1", rig.StateDir, rig.TransportDir, rig.Clock, rig.Logger, rig.Uploader, null!)); + } + + [Fact] + public void Observability_before_start_throws() + { + using var rig = new Rig(); + var sut = rig.Build(); + Assert.Throws(() => sut.CurrentState); + Assert.Throws(() => sut.IngressSink); + Assert.Throws(() => sut.EventEmitter); + } + + // ========================================================================= Start + + [Fact] + public void Start_instantiates_pipeline_no_exception() + { + using var rig = new Rig(); + var sut = rig.Build(); + + sut.Start(); + + Assert.NotNull(sut.CurrentState); + Assert.Equal("S1", sut.CurrentState.SessionId); + Assert.Equal("T1", sut.CurrentState.TenantId); + Assert.NotNull(sut.IngressSink); + Assert.NotNull(sut.EventEmitter); + Assert.False(sut.IsQuarantineRequested); + + sut.Stop(); + } + + [Fact] + public void Start_creates_state_and_transport_directories() + { + using var rig = new Rig(); + Assert.False(Directory.Exists(rig.StateDir)); + Assert.False(Directory.Exists(rig.TransportDir)); + + var sut = rig.Build(); + sut.Start(); + + Assert.True(Directory.Exists(rig.StateDir)); + Assert.True(Directory.Exists(rig.TransportDir)); + + sut.Stop(); + } + + [Fact] + public void Start_creates_initial_state_when_no_snapshot() + { + using var rig = new Rig(); + var sut = rig.Build(); + sut.Start(); + + Assert.Equal(SessionStage.SessionStarted, sut.CurrentState.Stage); + Assert.Equal(0, sut.CurrentState.StepIndex); + + sut.Stop(); + } + + [Fact] + public void Start_loads_recovered_snapshot_when_file_exists() + { + using var rig = new Rig(); + Directory.CreateDirectory(rig.StateDir); + // Seed a snapshot with a non-initial stage so recovery is detectable. + var persistedState = DecisionState.CreateInitial("S1", "T1") + .ToBuilder() + .WithStage(SessionStage.EspDeviceSetup) + .WithStepIndex(7) + .Build(); + var snapshotPath = Path.Combine(rig.StateDir, "snapshot.json"); + var seed = new SnapshotPersistence(snapshotPath); + seed.Save(persistedState); + + var sut = rig.Build(); + sut.Start(); + + Assert.Equal(SessionStage.EspDeviceSetup, sut.CurrentState.Stage); + Assert.Equal(7, sut.CurrentState.StepIndex); + + sut.Stop(); + } + + [Fact] + public void Start_twice_throws() + { + using var rig = new Rig(); + var sut = rig.Build(); + sut.Start(); + + Assert.Throws(() => sut.Start()); + + sut.Stop(); + } + + // ========================================================================= Stop + + [Fact] + public void Stop_before_start_is_noop() + { + using var rig = new Rig(); + var sut = rig.Build(); + + sut.Stop(); // does not throw + } + + [Fact] + public void Stop_is_idempotent() + { + using var rig = new Rig(); + var sut = rig.Build(); + sut.Start(); + + sut.Stop(); + sut.Stop(); // no-op on second call + } + + [Fact] + public void Stop_flushes_final_snapshot_with_current_state() + { + using var rig = new Rig(); + var sut = rig.Build(); + sut.Start(); + + // Drive a state change — post a SessionStarted synthetic signal so the reducer + // advances StepIndex. Use the IngressSink to mirror the prod path. + sut.IngressSink.Post( + kind: DecisionSignalKind.SessionStarted, + occurredAtUtc: At, + sourceOrigin: "Test", + evidence: new Evidence(EvidenceKind.Synthetic, "test", "test")); + + // Wait until the worker applied it. + Assert.True(SpinWait.SpinUntil(() => sut.CurrentState.StepIndex > 0, 2000)); + var stepIndexBeforeStop = sut.CurrentState.StepIndex; + + sut.Stop(); + + // Snapshot file must exist + contain the final state. + var snapshotPath = Path.Combine(rig.StateDir, "snapshot.json"); + Assert.True(File.Exists(snapshotPath)); + var reloaded = new SnapshotPersistence(snapshotPath).Load(); + Assert.NotNull(reloaded); + Assert.Equal(stepIndexBeforeStop, reloaded!.StepIndex); + } + + // ========================================================================= Dispose + + [Fact] + public void Dispose_stops_orchestrator() + { + using var rig = new Rig(); + var sut = rig.Build(); + sut.Start(); + + sut.Dispose(); + + // Second Dispose is safe; Start after Dispose throws ObjectDisposedException. + sut.Dispose(); + Assert.Throws(() => sut.Start()); + } + + // ========================================================================= Deadline bridge + + [Fact] + public void Deadline_fired_posts_DeadlineFired_signal_to_ingress() + { + using var rig = new Rig(); + var sut = rig.Build(); + sut.Start(); + + // Schedule a past-due deadline → DeadlineScheduler fires via ThreadPool immediately. + var scheduler = GetScheduler(sut); + var pastDue = new ActiveDeadline( + name: "hello_safety", + dueAtUtc: At.AddMinutes(-1), + firesSignalKind: DecisionSignalKind.DeadlineFired); + scheduler.Schedule(pastDue); + + // Wait for the signal-log to record the synthetic DeadlineFired signal. We read + // via the SignalLogWriter's own thread-safe ReadAll() to avoid racing the worker's + // file-append lock (File.ReadAllLines opens without FileShare.ReadWrite on net48). + var signalLog = GetSignalLog(sut); + Assert.True(SpinWait.SpinUntil( + () => ContainsDeadlineFired(signalLog), + 3000)); + + sut.Stop(); + } + + // ========================================================================= Quarantine + + [Fact] + public void TriggerQuarantine_sets_flag_and_reason() + { + using var rig = new Rig(); + var sut = rig.Build(); + sut.Start(); + + Assert.False(sut.IsQuarantineRequested); + + sut.TriggerQuarantine("test-reason"); + + Assert.True(sut.IsQuarantineRequested); + Assert.Equal("test-reason", sut.QuarantineReason); + + sut.Stop(); + } + + // ========================================================================= Helpers + + private static bool ContainsDeadlineFired(ISignalLogWriter signalLog) + { + foreach (var sig in signalLog.ReadAll()) + { + if (sig.Kind == DecisionSignalKind.DeadlineFired) return true; + } + return false; + } + + private static ISignalLogWriter GetSignalLog(EnrollmentOrchestrator sut) + { + var field = typeof(EnrollmentOrchestrator).GetField( + "_signalLog", + System.Reflection.BindingFlags.Instance | System.Reflection.BindingFlags.NonPublic); + return (ISignalLogWriter)field!.GetValue(sut)!; + } + + /// + /// The orchestrator owns its as a private field; + /// tests need it to script past-due deadlines. Reflection access is acceptable for + /// a single integration-ish test; production code doesn't expose it. + /// + private static DeadlineScheduler GetScheduler(EnrollmentOrchestrator sut) + { + var field = typeof(EnrollmentOrchestrator).GetField( + "_scheduler", + System.Reflection.BindingFlags.Instance | System.Reflection.BindingFlags.NonPublic); + return (DeadlineScheduler)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 new file mode 100644 index 00000000..e08a53e3 --- /dev/null +++ b/src/Agent/AutopilotMonitor.Agent.V2.Core/Orchestration/EnrollmentOrchestrator.cs @@ -0,0 +1,478 @@ +#nullable enable +using System; +using System.Collections.Generic; +using System.IO; +using System.Threading; +using System.Threading.Tasks; +using AutopilotMonitor.Agent.V2.Core.Logging; +using AutopilotMonitor.Agent.V2.Core.Persistence; +using AutopilotMonitor.Agent.V2.Core.Telemetry.Events; +using AutopilotMonitor.Agent.V2.Core.Transport.Telemetry; +using AutopilotMonitor.DecisionCore.Classifiers; +using AutopilotMonitor.DecisionCore.Engine; +using AutopilotMonitor.DecisionCore.Signals; +using AutopilotMonitor.DecisionCore.State; + +namespace AutopilotMonitor.Agent.V2.Core.Orchestration +{ + /// + /// Top-Level-Lifecycle-Owner der V2-Agent-Runtime. Plan §2.1a / §2.7 / §4.x M4.4.5. + /// + /// Scope M4.4.5.b: verdrahtet die Kern-Pipeline — Persistenz (SignalLog + Journal + + /// Snapshot), Telemetry-Transport (Spool + Uploader), Event-Emitter-Kette, + /// DeadlineScheduler → Ingress-Bridge, ClassifierRegistry, EffectRunner, + /// DecisionStepProcessor, SignalIngress. Sub-c erweitert um Collectors + SignalAdapters. + /// + /// + /// Startup-Reihenfolge (ctor → ): + /// + /// Persistenz-Writer instanziieren (Sofort-Flush aktiv ab hier) + /// snapshot.Load() → initialState; fallback + /// TelemetrySpool + BackendUploader + TelemetryUploadOrchestrator + /// EventSequenceCounter → TelemetryEventEmitter → EventTimelineEmitter + BackPressureObserver + /// DeadlineScheduler + ClassifierRegistry + /// LazyIngressSinkRelay (für die EffectRunner-↔-SignalIngress-Zirkel-Dep) + /// EffectRunner (sink = Relay) + /// DecisionStepProcessor (initial state) + /// SessionTraceOrdinalProvider (TODO M4.4.5.f: seed aus persisted max) + /// SignalIngress — Relay.Target = ingress + /// Scheduler.Fired → Ingress.Post-Bridge subscriben + /// signalIngress.Start() — Worker-Thread läuft + /// Periodischer Drain-Loop starten (Fire-and-forget Task) + /// + /// + /// + /// Shutdown-Reihenfolge (, idempotent): + /// + /// Scheduler.Fired-Handler abmelden (keine neuen DeadlineFired-Signals) + /// DeadlineScheduler disposen (stoppt alle Timer) + /// SignalIngress.Stop — drainiert verbleibende Items + /// Drain-Loop Token cancellen, Task joinen + /// Terminaler Drain — finale Batches ans Backend + /// Snapshot.Save — letzter konsistenter State + /// Dispose aller Disposables + /// + /// + /// + public sealed class EnrollmentOrchestrator : IQuarantineSink, IDisposable + { + /// Default-Intervall zwischen periodischen Drain-Versuchen. + public static readonly TimeSpan DefaultDrainInterval = TimeSpan.FromSeconds(30); + + /// Default Timeout für den Terminal-Drain beim Stop. + public static readonly TimeSpan DefaultTerminalDrainTimeout = TimeSpan.FromSeconds(30); + + private readonly string _sessionId; + private readonly string _tenantId; + private readonly string _stateDirectory; + private readonly string _transportDirectory; + private readonly IClock _clock; + private readonly AgentLogger _logger; + private readonly IBackendTelemetryUploader _uploader; + private readonly IReadOnlyList _classifiers; + private readonly int _channelCapacity; + private readonly int _quarantineThreshold; + private readonly TimeSpan _drainInterval; + private readonly TimeSpan _terminalDrainTimeout; + + // Built in Start() + private DecisionEngine? _engine; + private SignalLogWriter? _signalLog; + private JournalWriter? _journal; + private SnapshotPersistence? _snapshot; + private EventSequencePersistence? _eventSequencePersistence; + private EventSequenceCounter? _eventSequenceCounter; + private TelemetrySpool? _spool; + private TelemetryUploadOrchestrator? _transport; + private TelemetryEventEmitter? _eventEmitter; + private EventTimelineEmitter? _timelineEmitter; + private BackPressureEventObserver? _backPressureObserver; + private DeadlineScheduler? _scheduler; + private ClassifierRegistry? _classifierRegistry; + private LazyIngressSinkRelay? _sinkRelay; + private EffectRunner? _effectRunner; + private DecisionStepProcessor? _processor; + private SessionTraceOrdinalProvider? _traceCounter; + private SignalIngress? _ingress; + + private EventHandler? _deadlineBridge; + + // Drain-Loop + private CancellationTokenSource? _drainCts; + private Task? _drainTask; + + // Lifecycle + private int _started; + private int _stopRequested; + private int _disposed; + + // Quarantine flag — M4.4.5.f reads this on next start. + private bool _quarantineRequested; + private string? _quarantineReason; + + public EnrollmentOrchestrator( + string sessionId, + string tenantId, + string stateDirectory, + string transportDirectory, + IClock clock, + AgentLogger logger, + IBackendTelemetryUploader uploader, + IEnumerable classifiers, + int channelCapacity = SignalIngress.DefaultChannelCapacity, + int quarantineThreshold = DecisionStepProcessor.DefaultQuarantineThreshold, + TimeSpan? drainInterval = null, + TimeSpan? terminalDrainTimeout = null) + { + if (string.IsNullOrEmpty(sessionId)) throw new ArgumentException("SessionId is mandatory.", nameof(sessionId)); + if (string.IsNullOrEmpty(tenantId)) throw new ArgumentException("TenantId is mandatory.", nameof(tenantId)); + if (string.IsNullOrEmpty(stateDirectory)) throw new ArgumentException("StateDirectory is mandatory.", nameof(stateDirectory)); + if (string.IsNullOrEmpty(transportDirectory)) throw new ArgumentException("TransportDirectory is mandatory.", nameof(transportDirectory)); + + _sessionId = sessionId; + _tenantId = tenantId; + _stateDirectory = stateDirectory; + _transportDirectory = transportDirectory; + _clock = clock ?? throw new ArgumentNullException(nameof(clock)); + _logger = logger ?? throw new ArgumentNullException(nameof(logger)); + _uploader = uploader ?? throw new ArgumentNullException(nameof(uploader)); + + if (classifiers == null) throw new ArgumentNullException(nameof(classifiers)); + var list = new List(); + foreach (var c in classifiers) + { + if (c == null) throw new ArgumentException("Classifier enumerable must not contain null.", nameof(classifiers)); + list.Add(c); + } + _classifiers = list; + + _channelCapacity = channelCapacity; + _quarantineThreshold = quarantineThreshold; + _drainInterval = drainInterval ?? DefaultDrainInterval; + _terminalDrainTimeout = terminalDrainTimeout ?? DefaultTerminalDrainTimeout; + } + + // ---------------------------------------------------------------- Observability + + /// Aktueller — nur nach verfügbar. + public DecisionState CurrentState => + _processor?.CurrentState ?? throw new InvalidOperationException("Orchestrator not started."); + + /// True wenn ein Processor-Callback Quarantine-Eskalation ausgelöst hat. M4.4.5.f liest das. + public bool IsQuarantineRequested => _quarantineRequested; + + /// Letzter Quarantine-Reason oder null. + public string? QuarantineReason => _quarantineReason; + + /// Exposed für Sub-c-Wiring (SignalAdapters + Collector-Callbacks). + public ISignalIngressSink IngressSink => + (ISignalIngressSink?)_ingress ?? throw new InvalidOperationException("Orchestrator not started."); + + /// Exposed für Sub-c-Wiring (Collector onEventCollected-Bridge). + public TelemetryEventEmitter EventEmitter => + _eventEmitter ?? throw new InvalidOperationException("Orchestrator not started."); + + // ---------------------------------------------------------------- Lifecycle + + /// + /// Wires all components and starts the Ingress-Worker + periodischen Drain-Loop. + /// Idempotent per — zweiter Aufruf wirft. + /// + public void Start() + { + if (Volatile.Read(ref _disposed) == 1) throw new ObjectDisposedException(nameof(EnrollmentOrchestrator)); + if (Interlocked.Exchange(ref _started, 1) == 1) + { + throw new InvalidOperationException("EnrollmentOrchestrator already started."); + } + + EnsureDirectories(); + + // 1) Persistenz. + _signalLog = new SignalLogWriter(Path.Combine(_stateDirectory, "signal-log.jsonl")); + _journal = new JournalWriter(Path.Combine(_stateDirectory, "journal.jsonl")); + _snapshot = new SnapshotPersistence(Path.Combine(_stateDirectory, "snapshot.json"), () => _clock.UtcNow); + _eventSequencePersistence = new EventSequencePersistence(Path.Combine(_stateDirectory, "event-sequence.json")); + + // 2) Initial state — Recovery-Vorstufe. Sub-f macht Checksum/Replay-Handling rund. + var initialState = _snapshot.Load(); + if (initialState == null) + { + initialState = DecisionState.CreateInitial(_sessionId, _tenantId); + _logger.Info($"EnrollmentOrchestrator: fresh initial state for session {_sessionId}."); + } + else + { + _logger.Info($"EnrollmentOrchestrator: recovered state from snapshot (stage={initialState.Stage}, stepIndex={initialState.StepIndex})."); + } + + // 3) Telemetry-Transport. + _spool = new TelemetrySpool(_transportDirectory, _clock); + _transport = new TelemetryUploadOrchestrator(_spool, _uploader, _clock); + + // 4) Event-Emitter-Kette. + _eventSequenceCounter = new EventSequenceCounter(_eventSequencePersistence); + _eventEmitter = new TelemetryEventEmitter(_transport, _eventSequenceCounter, _sessionId, _tenantId); + _timelineEmitter = new EventTimelineEmitter(_eventEmitter); + _backPressureObserver = new BackPressureEventObserver(_eventEmitter, _clock); + + // 5) Deadlines + Classifiers. + _scheduler = new DeadlineScheduler(_clock); + _classifierRegistry = new ClassifierRegistry(_classifiers); + + // 6) Relay — EffectRunner braucht ISignalIngressSink, aber Ingress wird erst später + // gebaut. Relay löst den Zirkel über einen setzbaren Target-Pointer. + _sinkRelay = new LazyIngressSinkRelay(); + + // 7) EffectRunner. + _effectRunner = new EffectRunner( + scheduler: _scheduler, + classifiers: _classifierRegistry, + ingress: _sinkRelay, + emitter: _timelineEmitter, + snapshot: _snapshot, + clock: _clock); + + // 8) Processor — owns the initial state + journal + snapshot + quarantine hook. + _processor = new DecisionStepProcessor( + initialState: initialState, + journal: _journal, + effectRunner: _effectRunner, + snapshot: _snapshot, + quarantineSink: this, + logger: _logger, + quarantineThreshold: _quarantineThreshold); + + // 9) Trace-Ordinal. TODO(M4.4.5.f): seed aus max(SignalLog.LastOrdinal, + // Journal.LastStepIndex, Spool.LastAssignedItemId via SessionTraceOrdinal). + _traceCounter = new SessionTraceOrdinalProvider(); + + // 10) DecisionEngine + SignalIngress. + _engine = new DecisionEngine(); + _ingress = new SignalIngress( + engine: _engine, + signalLog: _signalLog, + traceCounter: _traceCounter, + processor: _processor, + clock: _clock, + backPressureObserver: _backPressureObserver, + channelCapacity: _channelCapacity); + + // 11) Relay auf den echten Ingress umbiegen. + _sinkRelay.Target = _ingress; + + // 12) Scheduler.Fired → synthetic DeadlineFired-Signal. + _deadlineBridge = OnDeadlineFired; + _scheduler.Fired += _deadlineBridge; + + // 13) Ingress-Worker starten. + _ingress.Start(); + + // 14) Periodischer Drain-Loop (Fire-and-forget). + _drainCts = new CancellationTokenSource(); + _drainTask = Task.Run(() => DrainLoopAsync(_drainCts.Token)); + + _logger.Info("EnrollmentOrchestrator: started."); + } + + /// + /// Stop the pipeline. Idempotent; calls from as well. + /// + public void Stop() + { + if (Volatile.Read(ref _started) == 0) return; + if (Interlocked.Exchange(ref _stopRequested, 1) == 1) return; + + _logger.Info("EnrollmentOrchestrator: stopping."); + + // 1) Scheduler.Fired-Handler abmelden — keine neuen DeadlineFired-Signals mehr. + try + { + if (_scheduler != null && _deadlineBridge != null) + { + _scheduler.Fired -= _deadlineBridge; + } + } + catch (Exception ex) { _logger.Warning($"EnrollmentOrchestrator: unsubscribe scheduler failed: {ex.Message}"); } + + // 2) Scheduler disposen — stoppt alle aktiven Timer. + try { _scheduler?.Dispose(); } + catch (Exception ex) { _logger.Warning($"EnrollmentOrchestrator: scheduler dispose failed: {ex.Message}"); } + + // 3) SignalIngress stoppen — drainiert verbleibende Items, wartet auf Worker-Join. + try { _ingress?.Stop(); } + catch (Exception ex) { _logger.Warning($"EnrollmentOrchestrator: ingress stop failed: {ex.Message}"); } + + // 4) Drain-Loop Token cancellen + warten. + try + { + _drainCts?.Cancel(); + _drainTask?.Wait(TimeSpan.FromSeconds(5)); + } + catch (AggregateException) { /* wrapped OperationCanceledException */ } + catch (Exception ex) { _logger.Warning($"EnrollmentOrchestrator: drain loop join failed: {ex.Message}"); } + + // 5) Terminaler Drain — finale Batches. + try + { + using (var timeoutCts = new CancellationTokenSource(_terminalDrainTimeout)) + { + _transport?.DrainAllAsync(timeoutCts.Token).GetAwaiter().GetResult(); + } + } + catch (Exception ex) { _logger.Warning($"EnrollmentOrchestrator: terminal drain failed: {ex.Message}"); } + + // 6) Finales Snapshot. + try + { + if (_processor != null && _snapshot != null) + { + _snapshot.Save(_processor.CurrentState); + } + } + catch (Exception ex) { _logger.Warning($"EnrollmentOrchestrator: final snapshot save failed: {ex.Message}"); } + + // 7) Dispose aller Disposables. + try { _transport?.Dispose(); } catch { /* best-effort */ } + try { _ingress?.Dispose(); } catch { /* best-effort */ } + try { _drainCts?.Dispose(); } catch { /* best-effort */ } + + _logger.Info("EnrollmentOrchestrator: stopped."); + } + + public void Dispose() + { + if (Interlocked.Exchange(ref _disposed, 1) == 1) return; + try { Stop(); } + catch { /* Dispose darf nicht werfen. */ } + } + + // ---------------------------------------------------------------- IQuarantineSink + + public void TriggerQuarantine(string reason) + { + _quarantineRequested = true; + _quarantineReason = reason ?? string.Empty; + _logger.Error($"EnrollmentOrchestrator: quarantine requested — {reason}"); + // Actual segment-quarantine happens on next Start (M4.4.5.f), not mid-run. + } + + // ---------------------------------------------------------------- Test helpers + + /// + /// Explicit drain for tests. Returns the of one flush cycle. + /// + internal Task DrainAsync(CancellationToken cancellationToken = default) + { + if (_transport == null) throw new InvalidOperationException("Orchestrator not started."); + return _transport.DrainAllAsync(cancellationToken); + } + + // ---------------------------------------------------------------- Private + + private void EnsureDirectories() + { + if (!Directory.Exists(_stateDirectory)) Directory.CreateDirectory(_stateDirectory); + if (!Directory.Exists(_transportDirectory)) Directory.CreateDirectory(_transportDirectory); + } + + private void OnDeadlineFired(object? sender, DeadlineFiredEventArgs e) + { + if (Volatile.Read(ref _stopRequested) == 1 || _ingress == null) return; + + try + { + var evidence = new Evidence( + kind: EvidenceKind.Synthetic, + identifier: e.Deadline.Name, + summary: $"deadline '{e.Deadline.Name}' fired at {e.Deadline.DueAtUtc:O}"); + + var payload = new Dictionary(StringComparer.Ordinal) + { + ["deadlineName"] = e.Deadline.Name, + }; + + // OccurredAtUtc = DueAtUtc (not firedAt) — replay-determinism per DeadlineFiredEventArgs doc. + _ingress.Post( + kind: DecisionSignalKind.DeadlineFired, + occurredAtUtc: e.Deadline.DueAtUtc, + sourceOrigin: "DeadlineScheduler", + evidence: evidence, + payload: payload); + } + catch (Exception ex) + { + // Deadline-Bridge darf den Scheduler-Thread nicht killen. + _logger.Error($"EnrollmentOrchestrator: failed to post DeadlineFired for '{e.Deadline.Name}'.", ex); + } + } + + private async Task DrainLoopAsync(CancellationToken ct) + { + try + { + while (!ct.IsCancellationRequested) + { + try + { + // Wall-clock delay, NOT _clock.Delay: drain is a real network cadence + // independent of the decision-engine's logical time. VirtualClock.Delay + // returns immediately and would cause a tight loop in unit tests. + await Task.Delay(_drainInterval, ct).ConfigureAwait(false); + } + catch (OperationCanceledException) { return; } + + if (ct.IsCancellationRequested) return; + + try + { + var result = await _transport!.DrainAllAsync(ct).ConfigureAwait(false); + if (result.FailedBatches > 0) + { + _logger.Warning( + $"EnrollmentOrchestrator: periodic drain had {result.FailedBatches} failed batch(es); " + + $"last error: {result.LastErrorReason}"); + } + } + catch (OperationCanceledException) { return; } + catch (Exception ex) + { + _logger.Error("EnrollmentOrchestrator: periodic drain threw.", ex); + } + } + } + catch (Exception ex) + { + _logger.Error("EnrollmentOrchestrator: drain loop faulted.", ex); + } + } + + /// + /// Pass-through to break the Effect-↔-Ingress ctor + /// cycle: EffectRunner requires a sink, but the real Ingress wants the processor + + /// effect-runner to already exist. Relay is constructed first, wired to the real + /// Ingress last. + /// + private sealed class LazyIngressSinkRelay : ISignalIngressSink + { + internal ISignalIngressSink? Target { get; set; } + + public void Post( + DecisionSignalKind kind, + DateTime occurredAtUtc, + string sourceOrigin, + Evidence evidence, + IReadOnlyDictionary? payload = null, + int kindSchemaVersion = 1) + { + var t = Target; + if (t == null) + { + throw new InvalidOperationException( + "LazyIngressSinkRelay.Target is null — SignalIngress was not wired yet."); + } + t.Post(kind, occurredAtUtc, sourceOrigin, evidence, payload, kindSchemaVersion); + } + } + } +}