diff --git a/src/Agent/AutopilotMonitor.Agent.V2.Core.Tests/Orchestration/RecoveryCoordinatorTests.cs b/src/Agent/AutopilotMonitor.Agent.V2.Core.Tests/Orchestration/RecoveryCoordinatorTests.cs index 419d6584..631fb761 100644 --- a/src/Agent/AutopilotMonitor.Agent.V2.Core.Tests/Orchestration/RecoveryCoordinatorTests.cs +++ b/src/Agent/AutopilotMonitor.Agent.V2.Core.Tests/Orchestration/RecoveryCoordinatorTests.cs @@ -292,6 +292,74 @@ namespace AutopilotMonitor.Agent.V2.Core.Tests.Orchestration Assert.True(new FileInfo(rig.StatePath("signal-log.jsonl")).Length > 0); } + // ================================================================= Replay failure → quarantine + + [Fact] + public void Byte_identical_duplicate_log_line_replays_clean_without_quarantine() + { + // Field case (session b9b92d89, 2026-07-09): the dying process double-flushed + // its final Append → identical JSONL line twice. ReadAll dedupes the artifact, + // so recovery replays the distinct signals and does NOT quarantine anything. + using var rig = new Rig(); + + var sigs = new[] + { + MakeSignal(0, DecisionSignalKind.SessionStarted), + MakeSignal(1, DecisionSignalKind.AppInstallCompleted), + }; + SeedSignalLog(rig, sigs); + + var logPath = rig.StatePath("signal-log.jsonl"); + var lines = File.ReadAllLines(logPath, Encoding.UTF8); + File.AppendAllText(logPath, lines[lines.Length - 1] + "\n", Encoding.UTF8); + + var expected = ReducerReplay.Replay( + new DecisionEngine(), DecisionState.CreateInitial(SessionId, TenantId), sigs); + + var result = rig.Recover(); + + Assert.Equal(expected.StepIndex, result.InitialState.StepIndex); + Assert.Equal(expected.LastAppliedSignalOrdinal, result.InitialState.LastAppliedSignalOrdinal); + Assert.False(Directory.Exists(Path.Combine(rig.StateDir, ".quarantine"))); + } + + [Fact] + public void Non_monotonic_log_is_quarantined_and_reseeds_fresh_instead_of_throwing() + { + // A duplicate ordinal on a NON-identical line (true corruption) must not escape + // Recover as a fatal exception — pre-fix this produced a permanent startup + // crash-loop (every restart replayed the same corrupt log and died again). + // Contract: quarantine all reducer segments, reseed from Initial. + using var rig = new Rig(); + + var sigs = new[] + { + MakeSignal(0, DecisionSignalKind.SessionStarted), + MakeSignal(1, DecisionSignalKind.AppInstallCompleted), + }; + SeedSignalLog(rig, sigs); + + // Same ordinal 1, different kind → different bytes → survives ReadAll dedup. + var rogue = AutopilotMonitor.DecisionCore.Serialization.SignalSerializer.Serialize( + MakeSignal(1, DecisionSignalKind.DesktopArrived)); + File.AppendAllText(rig.StatePath("signal-log.jsonl"), rogue + "\n", Encoding.UTF8); + + var result = rig.Recover(); + + Assert.True(result.WasStartupQuarantine); + Assert.Equal(SessionStage.SessionStarted, result.InitialState.Stage); + Assert.Equal(0, result.InitialState.StepIndex); + Assert.Equal(-1, result.InitialState.LastAppliedSignalOrdinal); + + // Corrupt stream preserved for forensics; live writers are fresh. + Assert.True(Directory.Exists(Path.Combine(rig.StateDir, ".quarantine"))); + var quarantinedLogs = Directory.GetFiles( + Path.Combine(rig.StateDir, ".quarantine"), "signal-log.jsonl", SearchOption.AllDirectories); + Assert.Single(quarantinedLogs); + Assert.Equal(-1, result.SignalLog.LastOrdinal); + Assert.Equal(-1, result.Journal.LastStepIndex); + } + // ================================================================= Journal alignment [Fact] diff --git a/src/Agent/AutopilotMonitor.Agent.V2.Core.Tests/Persistence/SignalLogWriterTests.cs b/src/Agent/AutopilotMonitor.Agent.V2.Core.Tests/Persistence/SignalLogWriterTests.cs index 9a67f2b6..2704c521 100644 --- a/src/Agent/AutopilotMonitor.Agent.V2.Core.Tests/Persistence/SignalLogWriterTests.cs +++ b/src/Agent/AutopilotMonitor.Agent.V2.Core.Tests/Persistence/SignalLogWriterTests.cs @@ -105,6 +105,55 @@ namespace AutopilotMonitor.Agent.V2.Core.Tests.Persistence Assert.Equal(2, reread.ReadAll().Count); } + [Fact] + public void ReadAll_skips_byte_identical_duplicate_line_crash_artifact() + { + // Field case (session b9b92d89, 2026-07-09): a process kill during the final + // Append (self-update restart) double-flushed the same buffered line → the + // identical JSONL line landed twice. That is a crash artifact, not tampering — + // ReadAll must dedupe it so recovery replay sees a strictly monotonic stream. + using var tmp = new TempDirectory(); + var log = tmp.File("signal-log.jsonl"); + + var writer = new SignalLogWriter(log); + writer.Append(TestSignals.Raw(0)); + writer.Append(TestSignals.Raw(1)); + + // Duplicate the last line verbatim — simulates the double-flush. + var lines = File.ReadAllLines(log, Encoding.UTF8); + File.AppendAllText(log, lines[lines.Length - 1] + "\n", Encoding.UTF8); + + var reread = new SignalLogWriter(log); + Assert.Equal(1, reread.LastOrdinal); + + var signals = reread.ReadAll(); + Assert.Equal(2, signals.Count); + Assert.Equal(0, signals[0].SessionSignalOrdinal); + Assert.Equal(1, signals[1].SessionSignalOrdinal); + } + + [Fact] + public void ReadAll_does_not_dedupe_non_identical_duplicate_ordinal() + { + // A duplicate ordinal carried by a DIFFERENT line (true corruption / tamper) + // must NOT be silently dropped — it stays in the stream so the replay ordinal + // check can reject it and trigger quarantine. + using var tmp = new TempDirectory(); + var log = tmp.File("signal-log.jsonl"); + + var writer = new SignalLogWriter(log); + writer.Append(TestSignals.Raw(0)); + writer.Append(TestSignals.Raw(1)); + + // Re-serialize ordinal 1 with a different kind → same ordinal, different bytes. + var rogue = AutopilotMonitor.DecisionCore.Serialization.SignalSerializer.Serialize( + TestSignals.Raw(1, DecisionSignalKind.DesktopArrived)); + File.AppendAllText(log, rogue + "\n", Encoding.UTF8); + + var reread = new SignalLogWriter(log); + Assert.Equal(3, reread.ReadAll().Count); + } + [Fact] public void ReadAll_on_missing_file_returns_empty() { diff --git a/src/Agent/AutopilotMonitor.Agent.V2.Core/Orchestration/RecoveryCoordinator.cs b/src/Agent/AutopilotMonitor.Agent.V2.Core/Orchestration/RecoveryCoordinator.cs index 5946c096..7658cd01 100644 --- a/src/Agent/AutopilotMonitor.Agent.V2.Core/Orchestration/RecoveryCoordinator.cs +++ b/src/Agent/AutopilotMonitor.Agent.V2.Core/Orchestration/RecoveryCoordinator.cs @@ -406,11 +406,47 @@ namespace AutopilotMonitor.Agent.V2.Core.Orchestration // 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); + try + { + initialState = ReducerReplay.Replay( + engine: replayEngine, + seed: seed, + signals: signalsToReplay, + onTransition: journal.Append); + } + catch (InvalidOperationException ex) + { + // ReducerReplay (or the Journal backfill callback) rejected the persisted + // stream — non-monotonic ordinal, null signal, or a journal monotonicity + // violation. The exception contract says "caller should quarantine rather + // than trust this stream": do exactly that instead of letting the throw + // escape as a fatal startup failure. Field case (session b9b92d89, + // 2026-07-09): a self-update restart killed the agent mid-append and left a + // duplicated final line in the SignalLog; the uncaught throw here turned + // that single crash artifact into a PERMANENT crash-loop — every restart + // replayed the same corrupt log and died again. Fail-closed to a fresh + // Initial seed; the quarantine bucket keeps the stream for forensics. + logger.Error( + "EnrollmentOrchestrator: SignalLog replay failed — quarantining all reducer " + + "segments and reseeding from Initial state.", ex); + + snapshot.Quarantine("replay-failed: " + ex.Message); + SegmentQuarantine.QuarantineAll( + stateDirectory, "replay-failed: " + ex.Message, () => clock.UtcNow); + + // Same pattern as the log-head-corrupt quarantine above: writers hold paths, + // not handles, but their in-memory counters are stale after the move — + // recreate to reset them to -1. + signalLog = new SignalLogWriter(signalLogPath); + journal = new JournalWriter(journalPath, () => clock.UtcNow); + eventSequence = new EventSequencePersistence(eventSequencePath); + + seed = DecisionState.CreateInitial(sessionId, tenantId, agentBootUtc); + signalsToReplay = Array.Empty(); + initialState = seed; + branchTag += "+replay-quarantined"; + wasStartupQuarantine = true; + } logger.Info( $"EnrollmentOrchestrator: recovery branch={branchTag}, " + diff --git a/src/Agent/AutopilotMonitor.Agent.V2.Core/Persistence/SignalLogWriter.cs b/src/Agent/AutopilotMonitor.Agent.V2.Core/Persistence/SignalLogWriter.cs index bbff7740..3a3f2243 100644 --- a/src/Agent/AutopilotMonitor.Agent.V2.Core/Persistence/SignalLogWriter.cs +++ b/src/Agent/AutopilotMonitor.Agent.V2.Core/Persistence/SignalLogWriter.cs @@ -76,12 +76,21 @@ namespace AutopilotMonitor.Agent.V2.Core.Persistence // WriteThrough + Flush(true) — belt-and-suspenders per L.12: OS cache // bypass plus explicit disk flush. Any return from Append means on-disk. + // + // bufferSize: 1 — at-most-once physical write. With the default 4096 + // buffer a line shorter than the buffer stays in FileStream's internal + // buffer until Flush; if that flush faults AFTER the data landed on disk + // (observed 2026-07-09, session b9b92d89: process killed by the self-update + // restart mid-append), the buffer position is not reset and the using- + // Dispose flushes the SAME buffer again → byte-identical duplicate line → + // non-monotonic ordinal on the next recovery replay. An unbuffered write + // hands the payload to the OS exactly once; Dispose has nothing to re-flush. using (var fs = new FileStream( _path, FileMode.Append, FileAccess.Write, FileShare.Read, - bufferSize: 4096, + bufferSize: 1, options: FileOptions.WriteThrough)) { fs.Write(bytes, 0, bytes.Length); @@ -106,10 +115,19 @@ namespace AutopilotMonitor.Agent.V2.Core.Persistence return signals; } + string? previousRawLine = null; foreach (var rawLine in File.ReadAllLines(_path, Encoding.UTF8)) { if (string.IsNullOrWhiteSpace(rawLine)) continue; + // Byte-identical consecutive duplicate = known crash artifact, not + // tampering: a pre-fix agent killed mid-append (self-update restart, + // session b9b92d89 2026-07-09) could flush the same buffered line twice. + // Skipping it preserves every distinct signal; anything non-monotonic + // that ISN'T an exact duplicate still reaches the replay ordinal check + // (and its quarantine handling) untouched. + if (rawLine == previousRawLine) continue; + DecisionSignal parsed; try { @@ -124,6 +142,7 @@ namespace AutopilotMonitor.Agent.V2.Core.Persistence } signals.Add(parsed); + previousRawLine = rawLine; } return signals;