diff --git a/src/Agent/AutopilotMonitor.Agent.V2.Core.Tests/Orchestration/RecoveryCoordinatorTests.cs b/src/Agent/AutopilotMonitor.Agent.V2.Core.Tests/Orchestration/RecoveryCoordinatorTests.cs new file mode 100644 index 00000000..419d6584 --- /dev/null +++ b/src/Agent/AutopilotMonitor.Agent.V2.Core.Tests/Orchestration/RecoveryCoordinatorTests.cs @@ -0,0 +1,484 @@ +using System; +using System.Collections.Generic; +using System.IO; +using System.Linq; +using System.Text; +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.DecisionCore.Engine; +using AutopilotMonitor.DecisionCore.Signals; +using AutopilotMonitor.DecisionCore.State; +using Xunit; + +namespace AutopilotMonitor.Agent.V2.Core.Tests.Orchestration +{ + /// + /// Direct unit tests for — the crash-recovery + /// state machine extracted from EnrollmentOrchestrator.Start (ARCH-F5). These exercise + /// the recovery branches in isolation (no ingress worker, drain loop, collectors, or scheduler), + /// asserting directly on the returned and the on-disk side effects. + /// + /// The orchestrator-level RecoveryPathTests still cover the same branches end-to-end + /// (replay reaches the live pipeline, deadlines re-arm, the Terminated event fires). This class + /// pins the coordinator's own contract: which branch is taken, which flags it returns, and which + /// segment files it moves — including branches the end-to-end tests under-assert + /// (a-total-loss, the prior-run quarantine-marker flags). + /// + /// + public sealed class RecoveryCoordinatorTests + { + private static readonly DateTime At = new DateTime(2026, 4, 20, 10, 0, 0, DateTimeKind.Utc); + private const string SessionId = "S1"; + private const string TenantId = "T1"; + + private sealed class Rig : IDisposable + { + public TempDirectory Tmp { get; } = new TempDirectory(); + public VirtualClock Clock { get; } = new VirtualClock(At); + public AgentLogger Logger { get; } + public string StateDir { get; } + + public Rig() + { + Logger = new AgentLogger(Tmp.Path, AgentLogLevel.Info); + StateDir = Path.Combine(Tmp.Path, "State"); + Directory.CreateDirectory(StateDir); + } + + // The orchestrator runs EnsureDirectories() before Recover(); mirror that contract. + public RecoveryResult Recover() => + RecoveryCoordinator.Recover(StateDir, SessionId, TenantId, Clock, Logger); + + public string StatePath(string name) => System.IO.Path.Combine(StateDir, name); + + public void Dispose() => Tmp.Dispose(); + } + + // ----------------------------------------------------------------- Fixtures + + private static DecisionSignal MakeSignal( + long ordinal, + DecisionSignalKind kind = DecisionSignalKind.AppInstallCompleted) => + new DecisionSignal( + sessionSignalOrdinal: ordinal, + sessionTraceOrdinal: ordinal, + kind: kind, + kindSchemaVersion: 1, + occurredAtUtc: At.AddSeconds(ordinal), + sourceOrigin: "recovery-coordinator-tests", + evidence: new Evidence(EvidenceKind.Synthetic, $"recovery:ord-{ordinal}", "test signal")); + + private static DecisionTransition MakeTransition(int stepIndex, long signalRef) => + new DecisionTransition( + stepIndex: stepIndex, + sessionTraceOrdinal: stepIndex, + signalOrdinalRef: signalRef, + occurredAtUtc: At.AddSeconds(stepIndex), + trigger: "TestTrigger", + fromStage: SessionStage.SessionStarted, + toStage: SessionStage.SessionStarted, + taken: true, + deadEndReason: null, + reducerVersion: "2.0.0.0"); + + private void SeedSnapshot(Rig rig, DecisionState state) => + new SnapshotPersistence(rig.StatePath("snapshot.json"), () => rig.Clock.UtcNow).Save(state); + + private void SeedSignalLog(Rig rig, params DecisionSignal[] signals) + { + var writer = new SignalLogWriter(rig.StatePath("signal-log.jsonl")); + foreach (var s in signals) writer.Append(s); + } + + // ================================================================= Branch c1 — fresh start + + [Fact] + public void Empty_state_directory_yields_fresh_initial_state_no_flags() + { + using var rig = new Rig(); + + var result = rig.Recover(); + + Assert.Equal(SessionStage.SessionStarted, result.InitialState.Stage); + Assert.Equal(0, result.InitialState.StepIndex); + Assert.Equal(SessionId, result.InitialState.SessionId); + Assert.Equal(TenantId, result.InitialState.TenantId); + + Assert.False(result.IsWhiteGlovePart2); + Assert.False(result.WasStartupQuarantine); + Assert.False(result.PriorRunQuarantined); + Assert.Null(result.PriorRunQuarantineReason); + + // All four writers are returned and ready for the wiring steps. + Assert.NotNull(result.SignalLog); + Assert.NotNull(result.Journal); + Assert.NotNull(result.Snapshot); + Assert.NotNull(result.EventSequence); + } + + // ================================================================= Branch b — clean snapshot + + [Fact] + public void Clean_snapshot_without_tail_is_loaded_as_seed() + { + using var rig = new Rig(); + var saved = DecisionState.CreateInitial(SessionId, TenantId) + .ToBuilder() + .WithStage(SessionStage.EspDeviceSetup) + .WithStepIndex(3) + .WithLastAppliedSignalOrdinal(2) + .Build(); + SeedSnapshot(rig, saved); + + var result = rig.Recover(); + + Assert.False(result.WasStartupQuarantine); + Assert.False(result.IsWhiteGlovePart2); + Assert.Equal(SessionStage.EspDeviceSetup, result.InitialState.Stage); + Assert.Equal(3, result.InitialState.StepIndex); + Assert.Equal(2, result.InitialState.LastAppliedSignalOrdinal); + } + + [Fact] + public void Clean_snapshot_restamps_agent_boot_utc_to_current_run() + { + // Branch (b) re-stamps AgentBootUtc to "now" so rehydrated deadlines floor at the + // current run start rather than the prior run's (premature-fire safety). + using var rig = new Rig(); + var priorBoot = At.AddHours(-5); + var saved = DecisionState.CreateInitial(SessionId, TenantId, priorBoot) + .ToBuilder() + .WithStage(SessionStage.AwaitingHello) + .WithStepIndex(4) + .WithLastAppliedSignalOrdinal(3) + .Build(); + SeedSnapshot(rig, saved); + + var result = rig.Recover(); + + Assert.Equal(At, result.InitialState.AgentBootUtc); + Assert.Equal(SessionStage.AwaitingHello, result.InitialState.Stage); + } + + [Fact] + public void Snapshot_plus_pending_tail_replays_tail_onto_snapshot() + { + using var rig = new Rig(); + + var sigs = new[] + { + MakeSignal(0, DecisionSignalKind.SessionStarted), + MakeSignal(1, DecisionSignalKind.AppInstallCompleted), + }; + SeedSignalLog(rig, sigs); + + // Snapshot lags one step behind (only ordinal 0 consumed). + var engine = new DecisionEngine(); + var lagged = engine.Reduce(DecisionState.CreateInitial(SessionId, TenantId), sigs[0]).NewState; + SeedSnapshot(rig, lagged); + + var expected = ReducerReplay.Replay(engine, DecisionState.CreateInitial(SessionId, TenantId), sigs); + + var result = rig.Recover(); + + Assert.False(result.WasStartupQuarantine); + Assert.Equal(expected.StepIndex, result.InitialState.StepIndex); + Assert.Equal(expected.LastAppliedSignalOrdinal, result.InitialState.LastAppliedSignalOrdinal); + Assert.Equal(expected.Stage, result.InitialState.Stage); + } + + // ================================================================= Branch a — corrupt snapshot + + [Fact] + public void Corrupt_snapshot_with_intact_log_quarantines_snapshot_only_and_replays_log() + { + using var rig = new Rig(); + + File.WriteAllText( + rig.StatePath("snapshot.json"), + "{\"Checksum\":\"deadbeef\",\"State\":{\"SessionId\":\"S1\",\"TenantId\":\"T1\"}}", + Encoding.UTF8); + + var sigs = new[] + { + MakeSignal(0, DecisionSignalKind.SessionStarted), + MakeSignal(1, DecisionSignalKind.AppInstallCompleted), + MakeSignal(2, DecisionSignalKind.AppInstallCompleted), + }; + SeedSignalLog(rig, sigs); + + var expected = ReducerReplay.Replay( + new DecisionEngine(), DecisionState.CreateInitial(SessionId, TenantId), sigs); + + var result = rig.Recover(); + + Assert.True(result.WasStartupQuarantine); + Assert.False(result.IsWhiteGlovePart2); + Assert.Equal(expected.StepIndex, result.InitialState.StepIndex); + Assert.Equal(expected.LastAppliedSignalOrdinal, result.InitialState.LastAppliedSignalOrdinal); + + // Snapshot moved aside; SignalLog survived (NOT quarantined). + Assert.False(File.Exists(rig.StatePath("snapshot.json"))); + Assert.True(File.Exists(rig.StatePath("signal-log.jsonl"))); + Assert.True(new FileInfo(rig.StatePath("signal-log.jsonl")).Length > 0); + + Assert.True(Directory.Exists(Path.Combine(rig.StateDir, ".quarantine"))); + } + + [Fact] + public void Corrupt_snapshot_with_head_corrupt_log_escalates_to_total_loss_quarantine() + { + // Branch (a-total-loss): the snapshot is corrupt AND the SignalLog is unreadable from + // its first line (non-empty file, zero parseable signals). Both segment groups are + // quarantined and the run seeds from a fresh Initial state. This is the branch the + // end-to-end RecoveryPathTests do not directly exercise. + using var rig = new Rig(); + + File.WriteAllText( + rig.StatePath("snapshot.json"), + "{\"Checksum\":\"deadbeef\",\"State\":{\"SessionId\":\"S1\",\"TenantId\":\"T1\"}}", + Encoding.UTF8); + + // Non-empty but unparseable from line one → ReadAll yields 0 → head-corrupt. + File.WriteAllText(rig.StatePath("signal-log.jsonl"), "not-json-garbage\n", Encoding.UTF8); + + var result = rig.Recover(); + + Assert.True(result.WasStartupQuarantine); + Assert.Equal(SessionStage.SessionStarted, result.InitialState.Stage); + Assert.Equal(0, result.InitialState.StepIndex); + + // The garbage log was moved into the quarantine bucket; the live writer is fresh. + Assert.True(Directory.Exists(Path.Combine(rig.StateDir, ".quarantine"))); + if (File.Exists(rig.StatePath("signal-log.jsonl"))) + { + var fresh = File.ReadAllText(rig.StatePath("signal-log.jsonl")); + Assert.DoesNotContain("not-json-garbage", fresh); + } + + // Fresh writer's recovery counters reset to -1 (no stale ordinal carried over). + Assert.Equal(-1, result.SignalLog.LastOrdinal); + } + + // ================================================================= Branch c2 — no snapshot, intact log + + [Fact] + public void No_snapshot_with_intact_log_rebuilds_via_full_log_replay() + { + using var rig = new Rig(); + + var sigs = new[] + { + MakeSignal(0, DecisionSignalKind.SessionStarted), + MakeSignal(1, DecisionSignalKind.AppInstallCompleted), + MakeSignal(2, DecisionSignalKind.AppInstallCompleted), + }; + SeedSignalLog(rig, sigs); + Assert.False(File.Exists(rig.StatePath("snapshot.json"))); + + var expected = ReducerReplay.Replay( + new DecisionEngine(), DecisionState.CreateInitial(SessionId, TenantId), sigs); + + var result = rig.Recover(); + + Assert.True(result.WasStartupQuarantine); + Assert.Equal(expected.StepIndex, result.InitialState.StepIndex); + Assert.Equal(expected.LastAppliedSignalOrdinal, result.InitialState.LastAppliedSignalOrdinal); + + // Log was not quarantined — only the missing snapshot drove the replay. + Assert.True(File.Exists(rig.StatePath("signal-log.jsonl"))); + Assert.True(new FileInfo(rig.StatePath("signal-log.jsonl")).Length > 0); + } + + // ================================================================= Journal alignment + + [Fact] + public void Journal_ahead_of_seed_truncates_phantom_tail_and_backfills_in_lockstep() + { + using var rig = new Rig(); + + var sigs = new[] + { + MakeSignal(0, DecisionSignalKind.SessionStarted), + MakeSignal(1, DecisionSignalKind.AppInstallCompleted), + }; + SeedSignalLog(rig, sigs); + + var engine = new DecisionEngine(); + var step0 = engine.Reduce(DecisionState.CreateInitial(SessionId, TenantId), sigs[0]); + var step1 = engine.Reduce(step0.NewState, sigs[1]); + SeedSnapshot(rig, step0.NewState); // StepIndex=1 + + // Journal holds 3 entries — StepIndex 3 is a phantom (no backing signal). + var journalPath = rig.StatePath("journal.jsonl"); + var seedJournal = new JournalWriter(journalPath, () => rig.Clock.UtcNow); + seedJournal.Append(step0.Transition); // StepIndex=1 + seedJournal.Append(step1.Transition); // StepIndex=2 + seedJournal.Append(MakeTransition(3, signalRef: 99)); // StepIndex=3 phantom + + var expected = ReducerReplay.Replay(engine, step0.NewState, new[] { sigs[1] }); + + var result = rig.Recover(); + + Assert.Equal(expected.StepIndex, result.InitialState.StepIndex); + Assert.Equal(expected.LastAppliedSignalOrdinal, result.InitialState.LastAppliedSignalOrdinal); + + // Two entries beyond seed.StepIndex=1 went to the phantom bucket. + var phantomFiles = Directory.GetFiles( + Path.Combine(rig.StateDir, ".quarantine"), + "journal-phantom-tail.jsonl", + SearchOption.AllDirectories); + Assert.Single(phantomFiles); + Assert.Equal(2, File.ReadAllLines(phantomFiles[0]).Count(l => !string.IsNullOrWhiteSpace(l))); + + // Live journal rebuilt in lockstep: [1, 2]. + var live = new JournalWriter(journalPath, () => rig.Clock.UtcNow).ReadAll(); + Assert.Equal(2, live.Count); + Assert.Equal(1, live[0].StepIndex); + Assert.Equal(2, live[1].StepIndex); + } + + [Fact] + public void Signal_log_ahead_of_journal_backfills_missing_journal_entries() + { + using var rig = new Rig(); + + var sigs = new[] + { + MakeSignal(0, DecisionSignalKind.SessionStarted), + MakeSignal(1, DecisionSignalKind.AppInstallCompleted), + }; + SeedSignalLog(rig, sigs); + + var engine = new DecisionEngine(); + var step0 = engine.Reduce(DecisionState.CreateInitial(SessionId, TenantId), sigs[0]); + SeedSnapshot(rig, step0.NewState); // StepIndex=1, LastApplied=0 + + // Journal only has the sig0 transition — sig1's was never flushed. + var journalPath = rig.StatePath("journal.jsonl"); + new JournalWriter(journalPath, () => rig.Clock.UtcNow).Append(step0.Transition); + + var expected = ReducerReplay.Replay(engine, step0.NewState, new[] { sigs[1] }); + + var result = rig.Recover(); + + Assert.Equal(expected.StepIndex, result.InitialState.StepIndex); + Assert.Equal(expected.LastAppliedSignalOrdinal, result.InitialState.LastAppliedSignalOrdinal); + + var live = new JournalWriter(journalPath, () => rig.Clock.UtcNow).ReadAll(); + Assert.Equal(2, live.Count); + Assert.Equal(1, live[0].StepIndex); + Assert.Equal(2, live[1].StepIndex); + + // Nothing was beyond the seed boundary → no phantom quarantine. + var quarantineRoot = Path.Combine(rig.StateDir, ".quarantine"); + if (Directory.Exists(quarantineRoot)) + { + Assert.Empty(Directory.GetFiles( + quarantineRoot, "journal-phantom-tail.jsonl", SearchOption.AllDirectories)); + } + } + + // ================================================================= WhiteGlove Part-2 resume + + [Fact] + public void WhiteGloveSealed_snapshot_archives_state_and_sets_part2_flag() + { + using var rig = new Rig(); + + var sealedState = DecisionState.CreateInitial(SessionId, TenantId) + .ToBuilder() + .WithStage(SessionStage.WhiteGloveSealed) + .WithStepIndex(5) + .Build(); + SeedSnapshot(rig, sealedState); + File.WriteAllText(rig.StatePath("signal-log.jsonl"), "{\"part1-leftover\":true}\n", Encoding.UTF8); + File.WriteAllText(rig.StatePath("journal.jsonl"), "{\"part1-journal\":true}\n", Encoding.UTF8); + File.WriteAllText(rig.StatePath("event-sequence.json"), "{\"LastAssignedSequence\":17}", Encoding.UTF8); + + var result = rig.Recover(); + + Assert.True(result.IsWhiteGlovePart2); + Assert.False(result.WasStartupQuarantine); + + // Fresh live state — not the sealed stage. + Assert.Equal(0, result.InitialState.StepIndex); + Assert.NotEqual(SessionStage.WhiteGloveSealed, result.InitialState.Stage); + + // Reducer-state segments archived to .part1-*; event-sequence intentionally preserved. + var buckets = Directory.GetDirectories(rig.StateDir, ".part1-*"); + Assert.Single(buckets); + Assert.True(File.Exists(Path.Combine(buckets[0], "snapshot.json"))); + Assert.True(File.Exists(Path.Combine(buckets[0], "signal-log.jsonl"))); + Assert.True(File.Exists(Path.Combine(buckets[0], "journal.jsonl"))); + Assert.False(File.Exists(Path.Combine(buckets[0], "event-sequence.json"))); + Assert.Contains("17", File.ReadAllText(rig.StatePath("event-sequence.json"))); + } + + [Fact] + public void Non_whiteglove_snapshot_does_not_set_part2_flag_or_archive() + { + using var rig = new Rig(); + SeedSnapshot(rig, DecisionState.CreateInitial(SessionId, TenantId) + .ToBuilder() + .WithStage(SessionStage.EspDeviceSetup) + .WithStepIndex(3) + .Build()); + + var result = rig.Recover(); + + Assert.False(result.IsWhiteGlovePart2); + Assert.Empty(Directory.GetDirectories(rig.StateDir, ".part1-*")); + } + + // ================================================================= Prior-run quarantine marker + + [Fact] + public void Prior_run_quarantine_marker_quarantines_all_segments_and_sets_flag() + { + using var rig = new Rig(); + + // Seed a snapshot + log so we can prove they get quarantined. + SeedSnapshot(rig, DecisionState.CreateInitial(SessionId, TenantId) + .ToBuilder().WithStage(SessionStage.EspDeviceSetup).WithStepIndex(3).Build()); + SeedSignalLog(rig, MakeSignal(0, DecisionSignalKind.SessionStarted)); + + const string reason = "consecutive-journal-append-failures"; + File.WriteAllText( + rig.StatePath(RecoveryCoordinator.QuarantineRequestedMarkerFile), + "2026-04-20T09:59:00.0000000Z " + reason, + Encoding.UTF8); + + var result = rig.Recover(); + + Assert.True(result.PriorRunQuarantined); + Assert.NotNull(result.PriorRunQuarantineReason); + Assert.Contains(reason, result.PriorRunQuarantineReason!); + + // Fresh Initial state — the discarded snapshot did NOT seed it. + Assert.Equal(SessionStage.SessionStarted, result.InitialState.Stage); + Assert.Equal(0, result.InitialState.StepIndex); + + // Marker consumed (deleted after a successful quarantine). + Assert.False(File.Exists(rig.StatePath(RecoveryCoordinator.QuarantineRequestedMarkerFile))); + + // Old snapshot moved aside. + Assert.False(File.Exists(rig.StatePath("snapshot.json"))); + Assert.True(Directory.Exists(Path.Combine(rig.StateDir, ".quarantine"))); + } + + [Fact] + public void No_quarantine_marker_leaves_prior_run_flags_clear() + { + using var rig = new Rig(); + + var result = rig.Recover(); + + Assert.False(result.PriorRunQuarantined); + Assert.Null(result.PriorRunQuarantineReason); + Assert.False(File.Exists(rig.StatePath(RecoveryCoordinator.QuarantineRequestedMarkerFile))); + } + } +} diff --git a/src/Agent/AutopilotMonitor.Agent.V2.Core/Orchestration/EnrollmentOrchestrator.cs b/src/Agent/AutopilotMonitor.Agent.V2.Core/Orchestration/EnrollmentOrchestrator.cs index 059c6ffa..ba6b65bd 100644 --- a/src/Agent/AutopilotMonitor.Agent.V2.Core/Orchestration/EnrollmentOrchestrator.cs +++ b/src/Agent/AutopilotMonitor.Agent.V2.Core/Orchestration/EnrollmentOrchestrator.cs @@ -134,20 +134,12 @@ namespace AutopilotMonitor.Agent.V2.Core.Orchestration private bool _quarantineRequested; private string? _quarantineReason; - // TRACE-H2: marker file written by TriggerQuarantine (mid-run, when the journal can no - // longer be appended) and honoured on the NEXT Start — a running process must not pull the - // state files out from under itself. When the marker is found at startup, all reducer-state - // segments + snapshot are quarantined and the session starts from a clean Initial state. - private const string QuarantineRequestedMarkerFile = "quarantine-requested.marker"; + // TRACE-H2: a prior run that could no longer persist consistently leaves a marker + // (see RecoveryCoordinator.QuarantineRequestedMarkerFile) that the NEXT start honours. + // These flags capture whether that happened this start, for the step 13b-Q event. private bool _priorRunQuarantined; private string? _priorRunQuarantineReason; - // P2: set when a prior-run quarantine was REQUESTED (marker present) but did NOT fully - // succeed this start. The marker demands a clean reset, so we must NOT fall back to normal - // snapshot/log recovery (which could reload the very state we were told to discard) — this - // forces the run to seed from a fresh Initial state instead. - private bool _forceCleanStart; - // Recovery flags (populated during Start()). // V1-symmetric Part-2 hint (analog to V1 MonitoringService._isWhiteGlovePart2): // set when Start() detects a persisted WhiteGloveSealed snapshot and archives the @@ -334,251 +326,23 @@ namespace AutopilotMonitor.Agent.V2.Core.Orchestration EnsureDirectories(); - var snapshotPath = Path.Combine(_stateDirectory, "snapshot.json"); - var signalLogPath = Path.Combine(_stateDirectory, "signal-log.jsonl"); - var journalPath = Path.Combine(_stateDirectory, "journal.jsonl"); - var eventSequencePath = Path.Combine(_stateDirectory, "event-sequence.json"); - - // -1) TRACE-H2 — honour a prior run's quarantine request BEFORE the writers scan the - // (suspect) segment files. A prior run hit the consecutive-journal-failure threshold - // and could no longer persist consistently; it left the marker so THIS run resets - // cleanly. Quarantining here (not mid-run) preserves the "a running process never - // pulls its own state files out from under itself" invariant. event-sequence is - // quarantined too — a clean restart re-seeds it from 0 (backend dedups on RowKey). - var quarantineMarkerPath = Path.Combine(_stateDirectory, QuarantineRequestedMarkerFile); - if (File.Exists(quarantineMarkerPath)) - { - var quarantineReason = TryReadMarkerReason(quarantineMarkerPath); - _logger.Error( - "EnrollmentOrchestrator: prior-run quarantine marker found — quarantining all state " + - $"segments + snapshot before recovery. Reason: {quarantineReason}"); - try - { - SegmentQuarantine.QuarantineAll( - _stateDirectory, "quarantine-requested:" + quarantineReason, () => _clock.UtcNow); - new SnapshotPersistence(snapshotPath, () => _clock.UtcNow) - .Quarantine("quarantine-requested:" + quarantineReason); - - // P2: delete the marker ONLY after the quarantine fully succeeded. If QuarantineAll - // or the snapshot quarantine threw (ACL / lock / disk), the marker is LEFT so the - // next start retries fail-closed instead of proceeding on possibly-still-corrupt - // segments with the retry hint lost. - try { File.Delete(quarantineMarkerPath); } - catch (Exception ex) { _logger.Warning($"EnrollmentOrchestrator: quarantine marker delete failed: {ex.Message}"); } - - _priorRunQuarantined = true; - _priorRunQuarantineReason = quarantineReason; - } - catch (Exception ex) - { - // Marker intentionally retained — do NOT delete on failure (P2). AND force a - // clean Initial seed for THIS run: a partial failure (e.g. segments moved but the - // snapshot quarantine threw) must not let normal recovery reload the discarded - // snapshot. Fail-closed to fresh state; the retained marker drives a proper retry - // next start. - _forceCleanStart = true; - _logger.Error( - "EnrollmentOrchestrator: prior-run quarantine cleanup FAILED — marker retained for " - + "next-start retry; forcing a clean Initial seed for this run.", ex); - } - } - - // 0) WhiteGlove Part-2 resume detection. V1-symmetric Archive-and-Reset - // pattern: peek the persisted snapshot via the lock-free static reader; if - // the prior run sealed Part-1, move the reducer-state segment files - // (snapshot, signal-log, journal) aside to a timestamped - // .part1-<ts>/ bucket BEFORE the writers below open them for - // scanning. event-sequence.json is intentionally preserved so the - // session-wide event sequence stays monotonic across the resume boundary - // (the backend orders events by (SessionId, Sequence) and the - // Web UI splits on resumed.sequence - 1). The orchestrator then - // runs through the normal recovery branches as a fresh first-boot - // (branch c1), re-emits a whiteglove_resumed lifecycle event - // after the onIngressReady hook, and lets the standard Classic - // enrollment flow drive completion. - // Anti-loop: once archived, the WhiteGloveSealed snapshot is gone, so a - // subsequent restart sees no marker and continues as a Classic session. - // The hint bool is in-memory only — also matches V1 (which clears the - // on-disk marker file before emitting whiteglove_resumed). - if (Directory.Exists(_stateDirectory)) - { - var rawSnapshot = SnapshotPersistence.TryReadRaw(snapshotPath); - if (rawSnapshot != null && rawSnapshot.Stage == SessionStage.WhiteGloveSealed) - { - StateArchiver.ArchiveStateFolder( - stateDirectory: _stateDirectory, - reason: "wg_part1_resume_archive", - utcNow: () => _clock.UtcNow, - logger: _logger); - _isWhiteGlovePart2 = true; - _logger.Info( - "EnrollmentOrchestrator: WhiteGlove Part-1 resume detected — state archived; " + - "starting fresh Classic enrollment flow for Part-2."); - } - } - - // 1) Persistenz. Writer scannen bestehende Files im Ctor (LastOrdinal / LastStepIndex). - _signalLog = new SignalLogWriter(signalLogPath); - _journal = new JournalWriter(journalPath, () => _clock.UtcNow); - _snapshot = new SnapshotPersistence(snapshotPath, () => _clock.UtcNow); - _eventSequencePersistence = new EventSequencePersistence(eventSequencePath); - - // 2) Recovery (Plan §2.7 Sonderfälle 1+2, Codex follow-up #1). - // - // Invariant: the SignalLog is authoritative truth, the Snapshot is a cache. - // Therefore on every recovery we prefer reconstructing state from the log over - // trusting a stale/corrupt snapshot. ReducerReplay.Replay (Phase 1) performs the - // pure fold; this method decides the seed + which signals to feed it. - // - // Branches: - // a) Snapshot present but corrupt → quarantine the snapshot ONLY (not the log!) - // and rebuild from the full SignalLog. If the log itself is head-corrupt - // (file non-empty but ReadAll yields zero signals) we escalate to a total- - // loss quarantine of the log segments and start fresh. - // b) Snapshot loaded cleanly → replay any SignalLog tail past the snapshot's - // LastAppliedSignalOrdinal so a pre-crash Journal/Snapshot lag is closed. - // c) No snapshot file — two sub-cases (Codex follow-up post-#50 #A): - // c1) SignalLog also empty → genuinely fresh start. - // c2) SignalLog has entries → crash before the first snapshot save ever - // landed; rebuild from the full SignalLog like branch (a) does after - // snapshot quarantine. Anything else would keep stale log entries on - // disk while state resets to StepIndex=0 and break monotonicity on - // the next append. - // - // After the seed + replay-stream are selected the Journal is aligned AHEAD-OF - // replay (phantom suffix beyond seed.StepIndex-1 goes to the forensic bucket) - // and backfilled DURING replay via the onTransition callback. This closes two - // crash windows in one mechanism (Codex follow-up post-#50 #C): - // • AHEAD — Journal was flushed but Snapshot not; phantom transitions beyond - // the seed get quarantined before replay so live Append monotonicity holds. - // • BEHIND — SignalLog was flushed but Journal not; the replay callback - // rematerialises the missing StepIndex entries onto the Journal so it - // ends up in exact lockstep with initialState. - var snapshotFileExistsPreLoad = File.Exists(snapshotPath); - var loadedState = _snapshot.Load(); - DecisionState initialState; - - // Reducer is stateless — a transient instance is enough for the pure replay below. - // The live ingress pipeline creates its own instance further down the Start() body. - var replayEngine = new DecisionEngine(); - - // Dispatch: pick the seed + signal stream per branch. Null seed is impossible past - // this block (every path assigns it), but the compiler needs the explicit init. - // AgentBootUtc gets stamped from the live clock so reducer deadline-arming sites - // can floor replayed-signal timestamps at the current run's start (replay-safety). - var agentBootUtc = _clock.UtcNow; - DecisionState seed = DecisionState.CreateInitial(_sessionId, _tenantId, agentBootUtc); - IReadOnlyList signalsToReplay = Array.Empty(); - string branchTag; - - if (_forceCleanStart) - { - // P2 — a requested prior-run quarantine did not fully succeed. Fail-closed: ignore - // any (possibly-still-present) snapshot and SignalLog and seed from a fresh Initial - // state. seed/signalsToReplay are already CreateInitial/empty above. The retained - // marker drives a proper full quarantine + retry on the next start. - _logger.Warning( - "EnrollmentOrchestrator: forcing clean Initial seed (prior-run quarantine incomplete) — " - + "skipping snapshot/log recovery for this run."); - branchTag = "forced-clean-after-failed-quarantine"; - } - else if (loadedState == null && snapshotFileExistsPreLoad) - { - // Branch (a) — Snapshot corrupt. Quarantine the snapshot and attempt log replay. - _logger.Error( - "EnrollmentOrchestrator: snapshot present but Load returned null (checksum mismatch or parse error) — quarantining snapshot and attempting SignalLog replay."); - _snapshot.Quarantine("checksum-mismatch-on-startup"); - - var loggedSignals = _signalLog.ReadAll(); - var logFileInfo = new FileInfo(signalLogPath); - var logIsHeadCorrupt = - logFileInfo.Exists && logFileInfo.Length > 0 && loggedSignals.Count == 0; - - if (logIsHeadCorrupt) - { - _logger.Warning( - "EnrollmentOrchestrator: SignalLog unreadable from the first line — escalating to total-loss quarantine."); - SegmentQuarantine.QuarantineAll( - _stateDirectory, "log-head-corrupt-after-snapshot-loss", () => _clock.UtcNow); - - // Writer hold paths, not handles — but their in-memory counters are stale - // after the quarantine move. Recreate to reset them to -1. - _signalLog = new SignalLogWriter(signalLogPath); - _journal = new JournalWriter(journalPath, () => _clock.UtcNow); - _eventSequencePersistence = new EventSequencePersistence(eventSequencePath); - - branchTag = "a-total-loss"; - } - else - { - signalsToReplay = loggedSignals; - branchTag = "a-full-log-replay"; - } - - _wasStartupQuarantine = true; - } - else if (loadedState != null) - { - // Branch (b) — Snapshot loaded. Catch up any SignalLog tail past the snapshot. - // Re-stamp AgentBootUtc so the current run's deadlines floor at "now" — the - // persisted boot anchor is from the prior run and using it would let replayed - // tail signals arm deadlines that are already past-due (premature fire). - seed = loadedState.ToBuilder().WithAgentBootUtc(agentBootUtc).Build(); - signalsToReplay = CollectSignalLogTailAfter(loadedState.LastAppliedSignalOrdinal); - branchTag = signalsToReplay.Count > 0 ? "b-tail-replay" : "b-snapshot-current"; - } - else - { - // Branch (c) — no snapshot on disk. Two sub-cases: - // (c1) SignalLog also empty → genuinely fresh start. - // (c2) SignalLog has entries (crash BEFORE first snapshot save) → full - // log replay from CreateInitial (Codex follow-up post-#50 #A). - var loggedSignals = _signalLog.ReadAll(); - if (loggedSignals.Count == 0) - { - branchTag = "c1-fresh"; - } - else - { - signalsToReplay = loggedSignals; - _wasStartupQuarantine = true; - branchTag = "c2-full-log-replay"; - } - } - - // Align Journal AHEAD phantoms before replay. Engine semantics: a transition - // produced for state at StepIndex=K carries StepIndex=K+1 (see - // DecisionEngine.BuildTakenTransition); after K reduces from initial the journal - // holds entries [1..K] with LastStepIndex=K. So any on-disk entry beyond - // seed.StepIndex is a phantom from a crash between Journal.Append and - // Snapshot.Save. Truncating first lets the replay callback Append monotonically - // from seed.StepIndex+1 without colliding. - var journalBoundary = seed.StepIndex; - if (_journal.LastStepIndex > journalBoundary) - { - _logger.Warning( - $"EnrollmentOrchestrator: Journal ahead of seed state " + - $"(journal.LastStepIndex={_journal.LastStepIndex}, seed.StepIndex={seed.StepIndex}) — " + - $"truncating phantom transitions to boundary={journalBoundary}."); - _journal.TruncateAfter(journalBoundary); - } - - // Replay + Journal backfill via the onTransition callback (Codex follow-up - // post-#50 #C). For BEHIND crashes (SignalLog flushed, Journal not) this - // rematerialises every missing StepIndex. For no-replay branches the callback - // is simply never invoked; the journal stays aligned from the pre-replay - // truncate step above. - initialState = ReducerReplay.Replay( - engine: replayEngine, - seed: seed, - signals: signalsToReplay, - onTransition: _journal.Append); - - _logger.Info( - $"EnrollmentOrchestrator: recovery branch={branchTag}, " + - $"stage={initialState.Stage}, stepIndex={initialState.StepIndex}, " + - $"signalsReplayed={signalsToReplay.Count}, " + - $"journal.LastStepIndex={_journal.LastStepIndex}."); + // 1+2) Crash-recovery state machine (Plan §2.7). Extracted to RecoveryCoordinator + // (ARCH-F5): it honours a prior-run quarantine marker, performs the WhiteGlove + // Part-2 archive-and-reset, opens the persistence writers against the resulting + // clean directory, picks the recovery branch + replay stream, aligns the journal + // AHEAD-phantom boundary, and folds to the seed state. The writers + flags it + // returns drive the wiring steps below; field names (_signalLog, _journal, …) + // are preserved for the recovery tests that reflect on them. + var recovery = RecoveryCoordinator.Recover(_stateDirectory, _sessionId, _tenantId, _clock, _logger); + _signalLog = recovery.SignalLog; + _journal = recovery.Journal; + _snapshot = recovery.Snapshot; + _eventSequencePersistence = recovery.EventSequence; + _isWhiteGlovePart2 = recovery.IsWhiteGlovePart2; + _wasStartupQuarantine = recovery.WasStartupQuarantine; + _priorRunQuarantined = recovery.PriorRunQuarantined; + _priorRunQuarantineReason = recovery.PriorRunQuarantineReason; + var initialState = recovery.InitialState; // 3) Telemetry-Transport. Batch size flows from AgentConfiguration.MaxBatchSize // via Program.cs (P1 fix: previously the remote-config knob was merged but @@ -1040,26 +804,6 @@ namespace AutopilotMonitor.Agent.V2.Core.Orchestration catch (Exception ex) { _logger.Error("EnrollmentOrchestrator: Terminated handler threw.", ex); } } - /// - /// - /// Recovery helper — collect all SignalLog entries whose SessionSignalOrdinal - /// is strictly greater than the snapshot's last-consumed ordinal. Signals equal-or-less - /// are already baked into the seed state and must never be re-applied. - /// - private IReadOnlyList CollectSignalLogTailAfter(long lastConsumedOrdinal) - { - var all = _signalLog!.ReadAll(); - var tail = new List(); - for (int i = 0; i < all.Count; i++) - { - if (all[i].SessionSignalOrdinal > lastConsumedOrdinal) - { - tail.Add(all[i]); - } - } - return tail; - } - /// /// M4.6.β — DecisionStepProcessor callback: the engine has transitioned the session /// into a terminal . Fires the public @@ -1256,7 +1000,7 @@ namespace AutopilotMonitor.Agent.V2.Core.Orchestration // log-only behaviour (the BootTrigger / max-lifetime restart still re-runs recovery). try { - var markerPath = Path.Combine(_stateDirectory, QuarantineRequestedMarkerFile); + var markerPath = Path.Combine(_stateDirectory, RecoveryCoordinator.QuarantineRequestedMarkerFile); var body = _clock.UtcNow.ToString("o", System.Globalization.CultureInfo.InvariantCulture) + " " + (reason ?? string.Empty); File.WriteAllText(markerPath, body, System.Text.Encoding.UTF8); @@ -1267,13 +1011,6 @@ namespace AutopilotMonitor.Agent.V2.Core.Orchestration } } - /// Reads the reason text from a quarantine marker; returns a fallback on any error. - private static string TryReadMarkerReason(string markerPath) - { - try { return File.ReadAllText(markerPath, System.Text.Encoding.UTF8); } - catch { return "unknown (marker unreadable)"; } - } - /// /// TRACE-H1/P1 — turns a quarantined oversized item into a telemetry_upload_poisoned /// Warning so the drop is visible on the backend timeline (the item stays in spool.jsonl for diff --git a/src/Agent/AutopilotMonitor.Agent.V2.Core/Orchestration/RecoveryCoordinator.cs b/src/Agent/AutopilotMonitor.Agent.V2.Core/Orchestration/RecoveryCoordinator.cs new file mode 100644 index 00000000..d84a939c --- /dev/null +++ b/src/Agent/AutopilotMonitor.Agent.V2.Core/Orchestration/RecoveryCoordinator.cs @@ -0,0 +1,419 @@ +#nullable enable +using System; +using System.Collections.Generic; +using System.IO; +using AutopilotMonitor.Agent.V2.Core.Logging; +using AutopilotMonitor.Agent.V2.Core.Persistence; +using AutopilotMonitor.Agent.V2.Core.Telemetry.Events; +using AutopilotMonitor.DecisionCore.Engine; +using AutopilotMonitor.DecisionCore.Signals; +using AutopilotMonitor.DecisionCore.State; + +namespace AutopilotMonitor.Agent.V2.Core.Orchestration +{ + /// + /// Immutable outcome of : the freshly-opened + /// persistence writers, the reconstructed seed , and the + /// recovery flags the orchestrator surfaces for observability and downstream wiring. + /// + internal sealed class RecoveryResult + { + public RecoveryResult( + SignalLogWriter signalLog, + JournalWriter journal, + SnapshotPersistence snapshot, + EventSequencePersistence eventSequence, + DecisionState initialState, + bool isWhiteGlovePart2, + bool wasStartupQuarantine, + bool priorRunQuarantined, + string? priorRunQuarantineReason) + { + SignalLog = signalLog; + Journal = journal; + Snapshot = snapshot; + EventSequence = eventSequence; + InitialState = initialState; + IsWhiteGlovePart2 = isWhiteGlovePart2; + WasStartupQuarantine = wasStartupQuarantine; + PriorRunQuarantined = priorRunQuarantined; + PriorRunQuarantineReason = priorRunQuarantineReason; + } + + /// SignalLog writer opened against the (post-quarantine, post-archive) state dir. + public SignalLogWriter SignalLog { get; } + + /// Journal writer, aligned in lockstep with . + public JournalWriter Journal { get; } + + /// Snapshot persistence handle for the live pipeline's save path. + public SnapshotPersistence Snapshot { get; } + + /// Event-sequence persistence (intentionally never archived across WG resume). + public EventSequencePersistence EventSequence { get; } + + /// The seed state to start the live decision pipeline from. + public DecisionState InitialState { get; } + + /// + /// true when a persisted snapshot was + /// detected and the prior state folder archived to .part1-<ts>/. + /// + public bool IsWhiteGlovePart2 { get; } + + /// + /// true when a corrupt snapshot / head-corrupt or orphaned SignalLog forced a + /// startup quarantine of state segments. + /// + public bool WasStartupQuarantine { get; } + + /// true when a prior-run quarantine marker was honoured this start. + public bool PriorRunQuarantined { get; } + + /// Reason text from the prior-run quarantine marker, or null. + public string? PriorRunQuarantineReason { get; } + } + + /// + /// Owns the V2 agent's crash-recovery state machine (Plan §2.7). Extracted from + /// (ARCH-F5) so the recovery decision logic is + /// isolated from the orchestrator's component-wiring steps. + /// + /// Order is load-bearing and mirrors the original Start() body exactly: + /// + /// -1) Prior-run quarantine marker — honour a marker left by a prior run that + /// could no longer persist consistently, quarantining all state segments + snapshot BEFORE + /// the writers scan the suspect files. Preserves the "a running process never pulls its own + /// state files out from under itself" invariant. + /// 0) WhiteGlove Part-2 resume — when the persisted snapshot is + /// , archive the reducer-state segments aside and + /// begin a fresh Classic flow. event-sequence.json is intentionally preserved. + /// 1) Persistence writers — opened after -1/0 so they scan a clean directory. + /// 2) Recovery branches — pick the seed + replay stream (forced-clean / corrupt + /// snapshot / clean snapshot tail-replay / no-snapshot), align the journal AHEAD-phantom + /// boundary, then fold via with journal backfill. + /// + /// + /// + /// The invariant throughout: the SignalLog is authoritative truth, the Snapshot is a cache — + /// every recovery prefers reconstructing from the log over trusting a stale/corrupt snapshot. + /// + /// + internal static class RecoveryCoordinator + { + // TRACE-H2: marker file written by EnrollmentOrchestrator.TriggerQuarantine (mid-run, when + // the journal can no longer be appended) and honoured on the NEXT Recover — a running + // process must not pull the state files out from under itself. When the marker is found at + // startup, all reducer-state segments + snapshot are quarantined and the session starts + // from a clean Initial state. + public const string QuarantineRequestedMarkerFile = "quarantine-requested.marker"; + + /// + /// Runs the full recovery pipeline against and returns + /// the opened writers, seed state, and recovery flags. The directory is assumed to exist + /// (the orchestrator calls EnsureDirectories first). + /// + public static RecoveryResult Recover( + string stateDirectory, + string sessionId, + string tenantId, + IClock clock, + AgentLogger logger) + { + var snapshotPath = Path.Combine(stateDirectory, "snapshot.json"); + var signalLogPath = Path.Combine(stateDirectory, "signal-log.jsonl"); + var journalPath = Path.Combine(stateDirectory, "journal.jsonl"); + var eventSequencePath = Path.Combine(stateDirectory, "event-sequence.json"); + + var isWhiteGlovePart2 = false; + var wasStartupQuarantine = false; + var priorRunQuarantined = false; + string? priorRunQuarantineReason = null; + + // P2: set when a prior-run quarantine was REQUESTED (marker present) but did NOT fully + // succeed this start. The marker demands a clean reset, so we must NOT fall back to normal + // snapshot/log recovery (which could reload the very state we were told to discard) — this + // forces the run to seed from a fresh Initial state instead. + var forceCleanStart = false; + + // -1) TRACE-H2 — honour a prior run's quarantine request BEFORE the writers scan the + // (suspect) segment files. A prior run hit the consecutive-journal-failure threshold + // and could no longer persist consistently; it left the marker so THIS run resets + // cleanly. Quarantining here (not mid-run) preserves the "a running process never + // pulls its own state files out from under itself" invariant. event-sequence is + // quarantined too — a clean restart re-seeds it from 0 (backend dedups on RowKey). + var quarantineMarkerPath = Path.Combine(stateDirectory, QuarantineRequestedMarkerFile); + if (File.Exists(quarantineMarkerPath)) + { + var quarantineReason = TryReadMarkerReason(quarantineMarkerPath); + logger.Error( + "EnrollmentOrchestrator: prior-run quarantine marker found — quarantining all state " + + $"segments + snapshot before recovery. Reason: {quarantineReason}"); + try + { + SegmentQuarantine.QuarantineAll( + stateDirectory, "quarantine-requested:" + quarantineReason, () => clock.UtcNow); + new SnapshotPersistence(snapshotPath, () => clock.UtcNow) + .Quarantine("quarantine-requested:" + quarantineReason); + + // P2: delete the marker ONLY after the quarantine fully succeeded. If QuarantineAll + // or the snapshot quarantine threw (ACL / lock / disk), the marker is LEFT so the + // next start retries fail-closed instead of proceeding on possibly-still-corrupt + // segments with the retry hint lost. + try { File.Delete(quarantineMarkerPath); } + catch (Exception ex) { logger.Warning($"EnrollmentOrchestrator: quarantine marker delete failed: {ex.Message}"); } + + priorRunQuarantined = true; + priorRunQuarantineReason = quarantineReason; + } + catch (Exception ex) + { + // Marker intentionally retained — do NOT delete on failure (P2). AND force a + // clean Initial seed for THIS run: a partial failure (e.g. segments moved but the + // snapshot quarantine threw) must not let normal recovery reload the discarded + // snapshot. Fail-closed to fresh state; the retained marker drives a proper retry + // next start. + forceCleanStart = true; + logger.Error( + "EnrollmentOrchestrator: prior-run quarantine cleanup FAILED — marker retained for " + + "next-start retry; forcing a clean Initial seed for this run.", ex); + } + } + + // 0) WhiteGlove Part-2 resume detection. V1-symmetric Archive-and-Reset + // pattern: peek the persisted snapshot via the lock-free static reader; if + // the prior run sealed Part-1, move the reducer-state segment files + // (snapshot, signal-log, journal) aside to a timestamped + // .part1-<ts>/ bucket BEFORE the writers below open them for + // scanning. event-sequence.json is intentionally preserved so the + // session-wide event sequence stays monotonic across the resume boundary + // (the backend orders events by (SessionId, Sequence) and the + // Web UI splits on resumed.sequence - 1). The orchestrator then + // runs through the normal recovery branches as a fresh first-boot + // (branch c1), re-emits a whiteglove_resumed lifecycle event + // after the onIngressReady hook, and lets the standard Classic + // enrollment flow drive completion. + // Anti-loop: once archived, the WhiteGloveSealed snapshot is gone, so a + // subsequent restart sees no marker and continues as a Classic session. + // The hint bool is in-memory only — also matches V1 (which clears the + // on-disk marker file before emitting whiteglove_resumed). + if (Directory.Exists(stateDirectory)) + { + var rawSnapshot = SnapshotPersistence.TryReadRaw(snapshotPath); + if (rawSnapshot != null && rawSnapshot.Stage == SessionStage.WhiteGloveSealed) + { + StateArchiver.ArchiveStateFolder( + stateDirectory: stateDirectory, + reason: "wg_part1_resume_archive", + utcNow: () => clock.UtcNow, + logger: logger); + isWhiteGlovePart2 = true; + logger.Info( + "EnrollmentOrchestrator: WhiteGlove Part-1 resume detected — state archived; " + + "starting fresh Classic enrollment flow for Part-2."); + } + } + + // 1) Persistenz. Writer scannen bestehende Files im Ctor (LastOrdinal / LastStepIndex). + var signalLog = new SignalLogWriter(signalLogPath); + var journal = new JournalWriter(journalPath, () => clock.UtcNow); + var snapshot = new SnapshotPersistence(snapshotPath, () => clock.UtcNow); + var eventSequence = new EventSequencePersistence(eventSequencePath); + + // 2) Recovery (Plan §2.7 Sonderfälle 1+2, Codex follow-up #1). + // + // Invariant: the SignalLog is authoritative truth, the Snapshot is a cache. + // Therefore on every recovery we prefer reconstructing state from the log over + // trusting a stale/corrupt snapshot. ReducerReplay.Replay (Phase 1) performs the + // pure fold; this method decides the seed + which signals to feed it. + // + // Branches: + // a) Snapshot present but corrupt → quarantine the snapshot ONLY (not the log!) + // and rebuild from the full SignalLog. If the log itself is head-corrupt + // (file non-empty but ReadAll yields zero signals) we escalate to a total- + // loss quarantine of the log segments and start fresh. + // b) Snapshot loaded cleanly → replay any SignalLog tail past the snapshot's + // LastAppliedSignalOrdinal so a pre-crash Journal/Snapshot lag is closed. + // c) No snapshot file — two sub-cases (Codex follow-up post-#50 #A): + // c1) SignalLog also empty → genuinely fresh start. + // c2) SignalLog has entries → crash before the first snapshot save ever + // landed; rebuild from the full SignalLog like branch (a) does after + // snapshot quarantine. Anything else would keep stale log entries on + // disk while state resets to StepIndex=0 and break monotonicity on + // the next append. + // + // After the seed + replay-stream are selected the Journal is aligned AHEAD-OF + // replay (phantom suffix beyond seed.StepIndex-1 goes to the forensic bucket) + // and backfilled DURING replay via the onTransition callback. This closes two + // crash windows in one mechanism (Codex follow-up post-#50 #C): + // • AHEAD — Journal was flushed but Snapshot not; phantom transitions beyond + // the seed get quarantined before replay so live Append monotonicity holds. + // • BEHIND — SignalLog was flushed but Journal not; the replay callback + // rematerialises the missing StepIndex entries onto the Journal so it + // ends up in exact lockstep with initialState. + var snapshotFileExistsPreLoad = File.Exists(snapshotPath); + var loadedState = snapshot.Load(); + DecisionState initialState; + + // Reducer is stateless — a transient instance is enough for the pure replay below. + // The live ingress pipeline creates its own instance further down the Start() body. + var replayEngine = new DecisionEngine(); + + // Dispatch: pick the seed + signal stream per branch. Null seed is impossible past + // this block (every path assigns it), but the compiler needs the explicit init. + // AgentBootUtc gets stamped from the live clock so reducer deadline-arming sites + // can floor replayed-signal timestamps at the current run's start (replay-safety). + var agentBootUtc = clock.UtcNow; + DecisionState seed = DecisionState.CreateInitial(sessionId, tenantId, agentBootUtc); + IReadOnlyList signalsToReplay = Array.Empty(); + string branchTag; + + if (forceCleanStart) + { + // P2 — a requested prior-run quarantine did not fully succeed. Fail-closed: ignore + // any (possibly-still-present) snapshot and SignalLog and seed from a fresh Initial + // state. seed/signalsToReplay are already CreateInitial/empty above. The retained + // marker drives a proper full quarantine + retry on the next start. + logger.Warning( + "EnrollmentOrchestrator: forcing clean Initial seed (prior-run quarantine incomplete) — " + + "skipping snapshot/log recovery for this run."); + branchTag = "forced-clean-after-failed-quarantine"; + } + else if (loadedState == null && snapshotFileExistsPreLoad) + { + // Branch (a) — Snapshot corrupt. Quarantine the snapshot and attempt log replay. + logger.Error( + "EnrollmentOrchestrator: snapshot present but Load returned null (checksum mismatch or parse error) — quarantining snapshot and attempting SignalLog replay."); + snapshot.Quarantine("checksum-mismatch-on-startup"); + + var loggedSignals = signalLog.ReadAll(); + var logFileInfo = new FileInfo(signalLogPath); + var logIsHeadCorrupt = + logFileInfo.Exists && logFileInfo.Length > 0 && loggedSignals.Count == 0; + + if (logIsHeadCorrupt) + { + logger.Warning( + "EnrollmentOrchestrator: SignalLog unreadable from the first line — escalating to total-loss quarantine."); + SegmentQuarantine.QuarantineAll( + stateDirectory, "log-head-corrupt-after-snapshot-loss", () => clock.UtcNow); + + // Writer hold paths, not handles — but their in-memory counters are stale + // after the quarantine move. Recreate to reset them to -1. + signalLog = new SignalLogWriter(signalLogPath); + journal = new JournalWriter(journalPath, () => clock.UtcNow); + eventSequence = new EventSequencePersistence(eventSequencePath); + + branchTag = "a-total-loss"; + } + else + { + signalsToReplay = loggedSignals; + branchTag = "a-full-log-replay"; + } + + wasStartupQuarantine = true; + } + else if (loadedState != null) + { + // Branch (b) — Snapshot loaded. Catch up any SignalLog tail past the snapshot. + // Re-stamp AgentBootUtc so the current run's deadlines floor at "now" — the + // persisted boot anchor is from the prior run and using it would let replayed + // tail signals arm deadlines that are already past-due (premature fire). + seed = loadedState.ToBuilder().WithAgentBootUtc(agentBootUtc).Build(); + signalsToReplay = CollectSignalLogTailAfter(signalLog, loadedState.LastAppliedSignalOrdinal); + branchTag = signalsToReplay.Count > 0 ? "b-tail-replay" : "b-snapshot-current"; + } + else + { + // Branch (c) — no snapshot on disk. Two sub-cases: + // (c1) SignalLog also empty → genuinely fresh start. + // (c2) SignalLog has entries (crash BEFORE first snapshot save) → full + // log replay from CreateInitial (Codex follow-up post-#50 #A). + var loggedSignals = signalLog.ReadAll(); + if (loggedSignals.Count == 0) + { + branchTag = "c1-fresh"; + } + else + { + signalsToReplay = loggedSignals; + wasStartupQuarantine = true; + branchTag = "c2-full-log-replay"; + } + } + + // Align Journal AHEAD phantoms before replay. Engine semantics: a transition + // produced for state at StepIndex=K carries StepIndex=K+1 (see + // DecisionEngine.BuildTakenTransition); after K reduces from initial the journal + // holds entries [1..K] with LastStepIndex=K. So any on-disk entry beyond + // seed.StepIndex is a phantom from a crash between Journal.Append and + // Snapshot.Save. Truncating first lets the replay callback Append monotonically + // from seed.StepIndex+1 without colliding. + var journalBoundary = seed.StepIndex; + if (journal.LastStepIndex > journalBoundary) + { + logger.Warning( + $"EnrollmentOrchestrator: Journal ahead of seed state " + + $"(journal.LastStepIndex={journal.LastStepIndex}, seed.StepIndex={seed.StepIndex}) — " + + $"truncating phantom transitions to boundary={journalBoundary}."); + journal.TruncateAfter(journalBoundary); + } + + // Replay + Journal backfill via the onTransition callback (Codex follow-up + // post-#50 #C). For BEHIND crashes (SignalLog flushed, Journal not) this + // rematerialises every missing StepIndex. For no-replay branches the callback + // is simply never invoked; the journal stays aligned from the pre-replay + // truncate step above. + initialState = ReducerReplay.Replay( + engine: replayEngine, + seed: seed, + signals: signalsToReplay, + onTransition: journal.Append); + + logger.Info( + $"EnrollmentOrchestrator: recovery branch={branchTag}, " + + $"stage={initialState.Stage}, stepIndex={initialState.StepIndex}, " + + $"signalsReplayed={signalsToReplay.Count}, " + + $"journal.LastStepIndex={journal.LastStepIndex}."); + + return new RecoveryResult( + signalLog: signalLog, + journal: journal, + snapshot: snapshot, + eventSequence: eventSequence, + initialState: initialState, + isWhiteGlovePart2: isWhiteGlovePart2, + wasStartupQuarantine: wasStartupQuarantine, + priorRunQuarantined: priorRunQuarantined, + priorRunQuarantineReason: priorRunQuarantineReason); + } + + /// + /// Recovery helper — collect all SignalLog entries whose SessionSignalOrdinal + /// is strictly greater than the snapshot's last-consumed ordinal. Signals equal-or-less + /// are already baked into the seed state and must never be re-applied. + /// + private static IReadOnlyList CollectSignalLogTailAfter( + SignalLogWriter signalLog, long lastConsumedOrdinal) + { + var all = signalLog.ReadAll(); + var tail = new List(); + for (int i = 0; i < all.Count; i++) + { + if (all[i].SessionSignalOrdinal > lastConsumedOrdinal) + { + tail.Add(all[i]); + } + } + return tail; + } + + /// Reads the reason text from a quarantine marker; returns a fallback on any error. + private static string TryReadMarkerReason(string markerPath) + { + try { return File.ReadAllText(markerPath, System.Text.Encoding.UTF8); } + catch { return "unknown (marker unreadable)"; } + } + } +} diff --git a/src/Agent/AutopilotMonitor.Agent.V2/buildcounter.txt b/src/Agent/AutopilotMonitor.Agent.V2/buildcounter.txt index fc15416f..7d58aa7a 100644 --- a/src/Agent/AutopilotMonitor.Agent.V2/buildcounter.txt +++ b/src/Agent/AutopilotMonitor.Agent.V2/buildcounter.txt @@ -1 +1 @@ -1095 +1099