fix(agent): recovery crash-loop on duplicated SignalLog line (session b9b92d89)

A runtime self-update restart (runtime_hash_mismatch, force) killed the agent
mid-append: the final agent_shutting_down line landed byte-identically twice
(FileStream 4096-buffer re-flushed by Dispose after a faulted flush). The next
start's ReducerReplay threw non-monotonic ordinal (27 <= 27) and the uncaught
exception turned one crash artifact into a PERMANENT startup crash-loop.

Three surgical fixes:
- SignalLogWriter.Append: bufferSize 1 -> unbuffered write; Dispose can never
  re-flush the payload (at-most-once physical append)
- SignalLogWriter.ReadAll: skip byte-identical consecutive duplicate lines
  (crash artifact, not tamper); non-identical duplicate ordinals stay visible
- RecoveryCoordinator.Recover: catch InvalidOperationException from replay ->
  quarantine all reducer segments + snapshot, reseed from Initial state
  (mirrors the existing log-head-corrupt path) instead of fatal exit

E2E-verified against the real customer signal-log.jsonl: 29 lines -> 28
distinct signals -> replay OK (LastAppliedOrdinal=27, zero signal loss).
Suites green: Agent.V2.Core 1719, DecisionCore 423.

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

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]

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()
         {

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(
+            try
-                engine: replayEngine,
+            {
-                seed: seed,
+                initialState = ReducerReplay.Replay(
-                signals: signalsToReplay,
+                    engine: replayEngine,
-                onTransition: journal.Append);
+                    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<DecisionSignal>();
+                initialState = seed;
+                branchTag += "+replay-quarantined";
+                wasStartupQuarantine = true;
+            }
 
             logger.Info(
                 $"EnrollmentOrchestrator: recovery branch={branchTag}, " +

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;