refactor(v2core): M4.4.5.f recovery paths (WG part-1 resume + segment quarantine + transport resume)

Plan §2.7 Sonderfaelle 1-3 umgesetzt:

Sonderfall 1 — WG Part-1 → Reboot → Part-2:
Orchestrator erkennt state.Stage == WhiteGloveSealed beim snapshot.Load
und setzt IsWhiteGlovePart2Resume-Flag. Nach ingress.Start (vor
Collectors/Adapters) wird ein SessionRecovered-Signal in den Ingress
gepostet — der Reducer bridged ueber HandleWhiteGlovePart1To2Bridge zu
WhiteGloveAwaitingUserSignIn. Self-Destruct-Disable bleibt Caller-Sache
(Info via IsWhiteGlovePart2Resume exponiert).

Sonderfall 2 — Beschaedigtes Segment:
Snapshot-File existiert aber Load liefert null (Checksum-Mismatch oder
Deserialize-Failure) → SnapshotPersistence.Quarantine + neuer
SegmentQuarantine-Helper bewegen Snapshot/SignalLog/Journal/
EventSequence nach .quarantine/{timestamp}/. Writer werden danach
reinstantiiert (stale LastOrdinal/LastStepIndex → reset auf -1). Fresh
start mit DecisionState.CreateInitial. Observability via
WasStartupQuarantine.

Sonderfall 3 — Transport-Resume:
Kein Code-Aenderung noetig: UploadCursorPersistence + TelemetrySpool
lesen persistierten Cursor im Ctor, Orchestrator resumed automatisch
ab LastUploadedItemId + 1. Wire-Format (PK, RK) garantiert Backend-
Idempotenz. Test verifiziert: seeded Spool mit 3 Items + Cursor@0 →
Resume liefert nur Items 1+2.

+5 RecoveryPathTests (268 → 273 gruen). Helper-Klasse
Persistence/SegmentQuarantine (parallel zu SnapshotPersistence.Quarantine)
als oeffentliche API fuer zukuenftige Recovery-Szenarien.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

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<IClassifier>(),
+                    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<DrainResult>)method!.Invoke(sut, new object?[] { CancellationToken.None })!;
+            return task.GetAwaiter().GetResult();
+        }
+    }
+}

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
         /// <summary>Letzter Quarantine-Reason oder <c>null</c>.</summary>
         public string? QuarantineReason => _quarantineReason;
 
+        /// <summary>
+        /// <c>true</c> wenn der Start mit einem persistierten <c>WhiteGloveSealed</c>-State
+        /// gelaufen ist und die Part-2-Bridge via <c>SessionRecovered</c>-Signal gezündet wurde.
+        /// Plan §2.7 Sonderfall 1.
+        /// </summary>
+        public bool IsWhiteGlovePart2Resume => _isWhiteGlovePart2Resume;
+
+        /// <summary>
+        /// <c>true</c> wenn der Start auf einen korrupten State-Segment traf und
+        /// Snapshot + Log-Segmente nach <c>.quarantine/{ts}/</c> bewegt wurden.
+        /// Plan §2.7 Sonderfall 2.
+        /// </summary>
+        public bool WasStartupQuarantine => _wasStartupQuarantine;
+
         /// <summary>Exposed für Sub-c-Wiring (SignalAdapters + Collector-Callbacks).</summary>
         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<string, string>(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).

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
+{
+    /// <summary>
+    /// Verschiebt beschädigte Persistenz-Segmente (SignalLog, Journal, EventSequence) in einen
+    /// timestamped <c>.quarantine/{ts}/</c>-Bucket. Plan §2.7 Sonderfall 2 / §4.x M4.4.5.f.
+    /// <para>
+    /// Parallel zu <see cref="SnapshotPersistence.Quarantine"/> — der Caller
+    /// (<see cref="Orchestration.EnrollmentOrchestrator"/>) bewegt den Snapshot via
+    /// <c>SnapshotPersistence.Quarantine</c> und die Log-Segmente via diesem Helper.
+    /// Beide Aufrufe schreiben in <c>&lt;stateDir&gt;/.quarantine/&lt;timestamp&gt;/</c>.
+    /// </para>
+    /// <para>
+    /// <b>Invariante</b>: wenn die Methode erfolgreich zurückkehrt, sind die bekannten
+    /// Log-Files aus <paramref name="stateDirectory"/> verschwunden — der nächste Append
+    /// legt frische Dateien an, der Reducer startet im Initial-State.
+    /// </para>
+    /// </summary>
+    public static class SegmentQuarantine
+    {
+        /// <summary>Die bekannten Segment-Dateien im State-Verzeichnis (neben snapshot.json).</summary>
+        private static readonly string[] KnownSegmentFiles = new[]
+        {
+            "signal-log.jsonl",
+            "journal.jsonl",
+            "event-sequence.json",
+        };
+
+        /// <summary>
+        /// Bewegt alle existierenden Segment-Files in einen timestamped Bucket. Kein-Op
+        /// wenn das State-Verzeichnis nicht existiert.
+        /// </summary>
+        public static void QuarantineAll(string stateDirectory, string reason, Func<DateTime> 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 */ }
+            }
+        }
+    }
+}