diff --git a/src/Agent/AutopilotMonitor.Agent.V2.Core.Tests/Orchestration/RecoveryPathTests.cs b/src/Agent/AutopilotMonitor.Agent.V2.Core.Tests/Orchestration/RecoveryPathTests.cs new file mode 100644 index 00000000..ac84e132 --- /dev/null +++ b/src/Agent/AutopilotMonitor.Agent.V2.Core.Tests/Orchestration/RecoveryPathTests.cs @@ -0,0 +1,249 @@ +using System; +using System.Collections.Generic; +using System.IO; +using System.Linq; +using System.Text; +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.Agent.V2.Core.Transport.Telemetry; +using AutopilotMonitor.DecisionCore.Classifiers; +using AutopilotMonitor.DecisionCore.Engine; +using AutopilotMonitor.DecisionCore.Signals; +using AutopilotMonitor.DecisionCore.State; +using Xunit; + +#pragma warning disable xUnit1031 + +namespace AutopilotMonitor.Agent.V2.Core.Tests.Orchestration +{ + public sealed class RecoveryPathTests + { + 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 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() => + new EnrollmentOrchestrator( + sessionId: "S1", + tenantId: "T1", + stateDirectory: StateDir, + transportDirectory: TransportDir, + clock: Clock, + logger: Logger, + uploader: Uploader, + classifiers: new List(), + drainInterval: TimeSpan.FromDays(1), + terminalDrainTimeout: TimeSpan.FromSeconds(2)); + + public void Dispose() => Tmp.Dispose(); + } + + // ================================================================= Sonderfall 1: WG Part-1 Resume + + [Fact] + public void Whiteglove_part1_snapshot_triggers_session_recovered_bridge_on_start() + { + using var rig = new Rig(); + Directory.CreateDirectory(rig.StateDir); + + // Seed a snapshot with stage = WhiteGloveSealed (Part 1 complete pre-reboot). + var sealedState = DecisionState.CreateInitial("S1", "T1") + .ToBuilder() + .WithStage(SessionStage.WhiteGloveSealed) + .WithStepIndex(5) + .Build(); + new SnapshotPersistence(Path.Combine(rig.StateDir, "snapshot.json")).Save(sealedState); + + var sut = rig.Build(); + sut.Start(); + + Assert.True(sut.IsWhiteGlovePart2Resume); + Assert.False(sut.WasStartupQuarantine); + + // After ingress start, the orchestrator posts SessionRecovered — the reducer bridges + // to WhiteGloveAwaitingUserSignIn. Wait for the signal-log to contain the signal. + var signalLog = GetSignalLog(sut); + Assert.True(SpinWait.SpinUntil( + () => signalLog.ReadAll().Any(s => s.Kind == DecisionSignalKind.SessionRecovered), + 3000)); + + // State must have advanced past WhiteGloveSealed to WhiteGloveAwaitingUserSignIn. + Assert.True(SpinWait.SpinUntil( + () => sut.CurrentState.Stage == SessionStage.WhiteGloveAwaitingUserSignIn, + 3000)); + + sut.Stop(); + } + + [Fact] + public void Non_whiteglove_snapshot_does_not_set_part2_resume_flag() + { + using var rig = new Rig(); + Directory.CreateDirectory(rig.StateDir); + + // Seed snapshot with a non-WG-sealed stage. + var midState = DecisionState.CreateInitial("S1", "T1") + .ToBuilder() + .WithStage(SessionStage.EspDeviceSetup) + .WithStepIndex(3) + .Build(); + new SnapshotPersistence(Path.Combine(rig.StateDir, "snapshot.json")).Save(midState); + + var sut = rig.Build(); + sut.Start(); + + Assert.False(sut.IsWhiteGlovePart2Resume); + + // No SessionRecovered should have been posted. + Thread.Sleep(100); + var signalLog = GetSignalLog(sut); + Assert.DoesNotContain(signalLog.ReadAll(), s => s.Kind == DecisionSignalKind.SessionRecovered); + + sut.Stop(); + } + + // ================================================================= Sonderfall 2: Segment-Quarantine + + [Fact] + public void Corrupt_snapshot_file_triggers_quarantine_and_fresh_start() + { + using var rig = new Rig(); + Directory.CreateDirectory(rig.StateDir); + + // Write a snapshot file that claims a valid checksum but has mismatching content. + var snapshotPath = Path.Combine(rig.StateDir, "snapshot.json"); + File.WriteAllText( + snapshotPath, + "{\"Checksum\":\"deadbeef\",\"State\":{\"SessionId\":\"S1\",\"TenantId\":\"T1\"}}", + Encoding.UTF8); + + // Seed a non-empty signal-log too so we can verify it also gets quarantined. + File.WriteAllText( + Path.Combine(rig.StateDir, "signal-log.jsonl"), + "{\"fake\": \"stale signal\"}\n", + Encoding.UTF8); + + var sut = rig.Build(); + sut.Start(); + + Assert.True(sut.WasStartupQuarantine); + Assert.False(sut.IsWhiteGlovePart2Resume); + + // Original snapshot + signal-log were moved aside. + Assert.False(File.Exists(snapshotPath)); + var signalLogPath = Path.Combine(rig.StateDir, "signal-log.jsonl"); + if (File.Exists(signalLogPath)) + { + // Fresh file should be empty (writer reinstantiated). + Assert.True(new FileInfo(signalLogPath).Length == 0); + } + + var quarantineRoot = Path.Combine(rig.StateDir, ".quarantine"); + Assert.True(Directory.Exists(quarantineRoot)); + var buckets = Directory.GetDirectories(quarantineRoot); + Assert.NotEmpty(buckets); + + // Bucket contains reason + moved segment(s). + var bucket = buckets[0]; + Assert.True(File.Exists(Path.Combine(bucket, "reason.txt"))); + + // Fresh initial state. + Assert.Equal(SessionStage.SessionStarted, sut.CurrentState.Stage); + Assert.Equal(0, sut.CurrentState.StepIndex); + + sut.Stop(); + } + + [Fact] + public void Missing_snapshot_does_not_trigger_quarantine() + { + using var rig = new Rig(); + var sut = rig.Build(); + + sut.Start(); + + Assert.False(sut.WasStartupQuarantine); + Assert.Equal(SessionStage.SessionStarted, sut.CurrentState.Stage); + + sut.Stop(); + } + + // ================================================================= Sonderfall 3: Transport-Resume + + [Fact] + public void Transport_resumes_from_persisted_cursor_after_restart() + { + using var rig = new Rig(); + Directory.CreateDirectory(rig.TransportDir); + + // Seed spool with 3 items (ItemIds 0/1/2, zero-based) and mark ItemId 0 as + // already uploaded. Remaining pending: ItemIds 1 + 2. + var spoolDir = rig.TransportDir; + var seedSpool = new TelemetrySpool(spoolDir, rig.Clock); + for (int i = 1; i <= 3; i++) + { + seedSpool.Enqueue(new TelemetryItemDraft( + kind: TelemetryItemKind.Event, + partitionKey: "T1_S1", + rowKey: $"row-{i}", + payloadJson: $"{{\"n\":{i}}}", + isSessionScoped: true, + requiresImmediateFlush: false)); + } + seedSpool.MarkUploaded(0); // pretend ItemId 0 already uploaded → 2 pending + + // Re-open via Orchestrator — uploader is scripted OK so the remaining 2 items drain. + rig.Uploader.QueueOk(10); + var sut = rig.Build(); + sut.Start(); + + // Explicit drain — tests don't wait for the periodic loop. + var drainResult = InvokeDrain(sut); + Assert.True(drainResult.Success); + Assert.Equal(2, drainResult.UploadedItems); + + // Uploader received one batch containing only items 2 + 3 (cursor was 1 on startup). + Assert.Single(rig.Uploader.Received); + Assert.Equal(2, rig.Uploader.Received[0].Count); + + sut.Stop(); + } + + // ================================================================= Helpers + + 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)!; + } + + private static DrainResult InvokeDrain(EnrollmentOrchestrator sut) + { + var method = typeof(EnrollmentOrchestrator).GetMethod( + "DrainAsync", + System.Reflection.BindingFlags.Instance | System.Reflection.BindingFlags.NonPublic); + var task = (System.Threading.Tasks.Task)method!.Invoke(sut, new object?[] { CancellationToken.None })!; + return task.GetAwaiter().GetResult(); + } + } +} diff --git a/src/Agent/AutopilotMonitor.Agent.V2.Core/Orchestration/EnrollmentOrchestrator.cs b/src/Agent/AutopilotMonitor.Agent.V2.Core/Orchestration/EnrollmentOrchestrator.cs index 9045c145..dac97082 100644 --- a/src/Agent/AutopilotMonitor.Agent.V2.Core/Orchestration/EnrollmentOrchestrator.cs +++ b/src/Agent/AutopilotMonitor.Agent.V2.Core/Orchestration/EnrollmentOrchestrator.cs @@ -109,10 +109,14 @@ namespace AutopilotMonitor.Agent.V2.Core.Orchestration private int _stopRequested; private int _disposed; - // Quarantine flag — M4.4.5.f reads this on next start. + // Quarantine flag — set mid-run, read on next start. private bool _quarantineRequested; private string? _quarantineReason; + // Recovery flags (populated during Start()). + private bool _isWhiteGlovePart2Resume; + private bool _wasStartupQuarantine; + public EnrollmentOrchestrator( string sessionId, string tenantId, @@ -171,6 +175,20 @@ namespace AutopilotMonitor.Agent.V2.Core.Orchestration /// Letzter Quarantine-Reason oder null. public string? QuarantineReason => _quarantineReason; + /// + /// true wenn der Start mit einem persistierten WhiteGloveSealed-State + /// gelaufen ist und die Part-2-Bridge via SessionRecovered-Signal gezündet wurde. + /// Plan §2.7 Sonderfall 1. + /// + public bool IsWhiteGlovePart2Resume => _isWhiteGlovePart2Resume; + + /// + /// true wenn der Start auf einen korrupten State-Segment traf und + /// Snapshot + Log-Segmente nach .quarantine/{ts}/ bewegt wurden. + /// Plan §2.7 Sonderfall 2. + /// + public bool WasStartupQuarantine => _wasStartupQuarantine; + /// Exposed für Sub-c-Wiring (SignalAdapters + Collector-Callbacks). public ISignalIngressSink IngressSink => (ISignalIngressSink?)_ingress ?? throw new InvalidOperationException("Orchestrator not started."); @@ -195,22 +213,61 @@ namespace AutopilotMonitor.Agent.V2.Core.Orchestration EnsureDirectories(); - // 1) Persistenz. + // 1) Persistenz. Writer scannen bestehende Files im Ctor (LastOrdinal / LastStepIndex). + var snapshotPath = Path.Combine(_stateDirectory, "snapshot.json"); _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); + _snapshot = new SnapshotPersistence(snapshotPath, () => _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) + // 2) Recovery (Plan §2.7 Sonderfälle 1+2). + var snapshotFileExistsPreLoad = File.Exists(snapshotPath); + var loadedState = _snapshot.Load(); + DecisionState initialState; + + if (loadedState == null && snapshotFileExistsPreLoad) + { + // Sonderfall 2: Snapshot-File existiert, Load lieferte null → Checksum-Mismatch + // oder Deserialize-Failure. Plan §2.7c: Snapshot ist Cache, nicht Wahrheit — + // Snapshot + Log-Segmente in Quarantäne, frisch starten. + _logger.Error( + "EnrollmentOrchestrator: snapshot present but Load returned null (checksum mismatch or parse error) — quarantining state."); + const string reason = "checksum-mismatch-on-startup"; + _snapshot.Quarantine(reason); + SegmentQuarantine.QuarantineAll(_stateDirectory, reason, () => _clock.UtcNow); + + // Writer halten Pfade, nicht Handles — aber ihre in-memory Counter (LastOrdinal, + // LastStepIndex) sind jetzt stale gegenüber den entfernten Files. Recreate, damit + // Counter bei -1 starten. + _signalLog = new SignalLogWriter(Path.Combine(_stateDirectory, "signal-log.jsonl")); + _journal = new JournalWriter(Path.Combine(_stateDirectory, "journal.jsonl")); + _eventSequencePersistence = new EventSequencePersistence(Path.Combine(_stateDirectory, "event-sequence.json")); + + initialState = DecisionState.CreateInitial(_sessionId, _tenantId); + _wasStartupQuarantine = true; + } + else if (loadedState != null) + { + initialState = loadedState; + _logger.Info( + $"EnrollmentOrchestrator: recovered state from snapshot (stage={loadedState.Stage}, stepIndex={loadedState.StepIndex})."); + } + else { initialState = DecisionState.CreateInitial(_sessionId, _tenantId); _logger.Info($"EnrollmentOrchestrator: fresh initial state for session {_sessionId}."); } - else + + // Sonderfall 1 Detection: WG Part 1 -> Reboot -> Part 2 Resume. + // Self-destruct bleibt AUS (Caller ist zuständig — die Info wird via + // IsWhiteGlovePart2Resume exponiert); Session-Recovered-Signal wird gepostet + // nachdem Ingress gestartet ist. + if (initialState.Stage == SessionStage.WhiteGloveSealed) { - _logger.Info($"EnrollmentOrchestrator: recovered state from snapshot (stage={initialState.Stage}, stepIndex={initialState.StepIndex})."); + _isWhiteGlovePart2Resume = true; + _logger.Info( + "EnrollmentOrchestrator: White-Glove Part-1 resume detected — SessionRecovered signal " + + "will be posted after ingress start; self-destruct handling is caller-owned."); } // 3) Telemetry-Transport. @@ -275,6 +332,34 @@ namespace AutopilotMonitor.Agent.V2.Core.Orchestration // 13) Ingress-Worker starten (vor Collectors — sonst race-prone). _ingress.Start(); + // 13a) Recovery Sonderfall 1: WG Part-1 → Part-2 bridge zünden. Das Signal muss + // nach _ingress.Start laufen, aber VOR Collectors/Adapters — so sieht der + // Reducer-Worker zuerst die Bridge-Transition (WhiteGloveSealed → AwaitingUserSignIn) + // und erst danach kommende Real-Signals. + if (_isWhiteGlovePart2Resume) + { + try + { + _ingress.Post( + kind: DecisionSignalKind.SessionRecovered, + occurredAtUtc: _clock.UtcNow, + sourceOrigin: "EnrollmentOrchestrator", + evidence: new Evidence( + kind: EvidenceKind.Synthetic, + identifier: "wg_part1_resume", + summary: "White-Glove Part-1 state recovered from snapshot; Part-2 bridge triggered."), + payload: new Dictionary(StringComparer.Ordinal) + { + ["from"] = SessionStage.WhiteGloveSealed.ToString(), + ["to"] = SessionStage.WhiteGloveAwaitingUserSignIn.ToString(), + }); + } + catch (Exception ex) + { + _logger.Error("EnrollmentOrchestrator: failed to post SessionRecovered for WG Part-2 resume.", ex); + } + } + // 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). diff --git a/src/Agent/AutopilotMonitor.Agent.V2.Core/Persistence/SegmentQuarantine.cs b/src/Agent/AutopilotMonitor.Agent.V2.Core/Persistence/SegmentQuarantine.cs new file mode 100644 index 00000000..7a8b4142 --- /dev/null +++ b/src/Agent/AutopilotMonitor.Agent.V2.Core/Persistence/SegmentQuarantine.cs @@ -0,0 +1,78 @@ +#nullable enable +using System; +using System.IO; +using System.Text; + +namespace AutopilotMonitor.Agent.V2.Core.Persistence +{ + /// + /// Verschiebt beschädigte Persistenz-Segmente (SignalLog, Journal, EventSequence) in einen + /// timestamped .quarantine/{ts}/-Bucket. Plan §2.7 Sonderfall 2 / §4.x M4.4.5.f. + /// + /// Parallel zu — der Caller + /// () bewegt den Snapshot via + /// SnapshotPersistence.Quarantine und die Log-Segmente via diesem Helper. + /// Beide Aufrufe schreiben in <stateDir>/.quarantine/<timestamp>/. + /// + /// + /// Invariante: wenn die Methode erfolgreich zurückkehrt, sind die bekannten + /// Log-Files aus verschwunden — der nächste Append + /// legt frische Dateien an, der Reducer startet im Initial-State. + /// + /// + public static class SegmentQuarantine + { + /// Die bekannten Segment-Dateien im State-Verzeichnis (neben snapshot.json). + private static readonly string[] KnownSegmentFiles = new[] + { + "signal-log.jsonl", + "journal.jsonl", + "event-sequence.json", + }; + + /// + /// Bewegt alle existierenden Segment-Files in einen timestamped Bucket. Kein-Op + /// wenn das State-Verzeichnis nicht existiert. + /// + public static void QuarantineAll(string stateDirectory, string reason, Func utcNow) + { + if (string.IsNullOrEmpty(stateDirectory)) throw new ArgumentException("stateDirectory is mandatory.", nameof(stateDirectory)); + if (utcNow == null) throw new ArgumentNullException(nameof(utcNow)); + if (!Directory.Exists(stateDirectory)) return; + + var quarantineRoot = Path.Combine(stateDirectory, ".quarantine"); + if (!Directory.Exists(quarantineRoot)) + { + Directory.CreateDirectory(quarantineRoot); + } + + var bucket = Path.Combine( + quarantineRoot, + utcNow().ToString("yyyyMMdd'T'HHmmssfff'Z'")); + Directory.CreateDirectory(bucket); + + var movedAny = false; + foreach (var name in KnownSegmentFiles) + { + var src = Path.Combine(stateDirectory, name); + if (File.Exists(src)) + { + File.Move(src, Path.Combine(bucket, name)); + movedAny = true; + } + } + + // Reason-Sidecar immer schreiben, auch wenn keine Datei bewegt wurde — macht die + // Quarantäne-Historie nachvollziehbar. + File.WriteAllText(Path.Combine(bucket, "reason.txt"), reason ?? string.Empty, Encoding.UTF8); + + // Wenn nichts zu bewegen war → leeren Bucket entfernen, damit kein leeres + // Quarantine-Verzeichnis bleibt. + if (!movedAny) + { + try { File.Delete(Path.Combine(bucket, "reason.txt")); } catch { /* best-effort */ } + try { Directory.Delete(bucket); } catch { /* best-effort */ } + } + } + } +}