v2 codex-post-#50: recovery + abort durability follow-ups (#A #B #C #D)

Addresses Codex review against PR #50: #A (High) EnrollmentOrchestrator branch (c) now distinguishes empty-log (fresh start) from log-has-entries (full-log replay via ReducerReplay) so a crash-before-first-snapshot cannot regress state to StepIndex=0 while the SignalLog preserves older ordinals. #B (High) EffectInfrastructureFailure is now crash-durable. ApplyStep returns EffectRunResult; on SessionMustAbort the phantom snapshot is skipped, and SignalIngress synthesises + durably appends + reduces the terminal failure signal INLINE within the same worker iteration. The old in-memory PostEffectInfrastructureFailureSignal path in EffectRunner is removed. #C (Medium) ReducerReplay.Replay gained an Action<DecisionTransition>? onTransition overload. The orchestrator's recovery path truncates Journal AHEAD-of-seed phantoms before replay and rematerialises BEHIND entries bit-for-bit via the callback, closing the gap between SignalLog.Append and Journal.Append crashes. #D (Medium) ReducerVerifier.ReplaySemantically was seeding DecisionState with swapped (tenantId, sessionId) — tests mirrored the bug so it slipped through. Args fixed; defensive replay_identity_mismatch issue added so future swaps surface as an Error-severity verifier finding. Suite-Baseline all green, 0 warnings: DecisionCore 164/164, Agent.V2.Core 648/648, Functions 735/735. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
2026-10-24 12:42:38 +00:00 · 2026-04-24 13:39:47 +02:00
parent ed1c26f98b
commit bd3ec4e195
16 changed files with 608 additions and 173 deletions
@@ -196,8 +196,13 @@ namespace AutopilotMonitor.Agent.V2.Core.Tests.Orchestration
        // ========================================================================= EffectRunner behavior

        [Fact]
-        public void ApplyStep_effect_runner_SessionMustAbort_is_logged_but_step_committed()
+        public void ApplyStep_effect_runner_SessionMustAbort_skips_snapshot_and_returns_result()
        {
+            // Codex follow-up (post-#50 #B): when the effects signal SessionMustAbort,
+            // the processor MUST NOT snapshot the phantom state — a crash-and-recovery from
+            // that snapshot would re-arm the same failed deadline and hang the session.
+            // The caller (SignalIngress) sees the abort flag on the returned result and
+            // durably synthesises a terminal EffectInfrastructureFailure signal inline.
            using var rig = new Rig();
            var abortResult = new EffectRunResult(
                sessionMustAbort: true,
@@ -210,11 +215,14 @@ namespace AutopilotMonitor.Agent.V2.Core.Tests.Orchestration
            var sut = rig.Build();
            var (step, signal) = rig.ReduceSessionStarted();

-            sut.ApplyStep(step, signal);   // state still advances
+            var result = sut.ApplyStep(step, signal);

            Assert.Single(rig.Journal.Appended);
-            Assert.Same(step.NewState, sut.CurrentState);
-            Assert.Empty(rig.Quarantine.Reasons);   // effect-failures do NOT quarantine
+            Assert.Same(step.NewState, sut.CurrentState);   // state forwards so the synthetic signal reduces from here
+            Assert.Equal(0, rig.Snapshot.SaveCallCount);     // phantom snapshot suppressed
+            Assert.True(result.SessionMustAbort);
+            Assert.Equal(abortResult.AbortReason, result.AbortReason);
+            Assert.Empty(rig.Quarantine.Reasons);
        }

        [Fact]
@@ -91,11 +91,12 @@ namespace AutopilotMonitor.Agent.V2.Core.Tests.Orchestration
        }

        [Fact]
-        public async Task ScheduleDeadline_failure_posts_EffectInfrastructureFailure_signal_to_ingress()
+        public async Task ScheduleDeadline_failure_does_not_post_signal_from_EffectRunner()
        {
-            // Codex follow-up #2 — critical deadline failure must surface in the decision
-            // world, not just the log. EffectRunner posts a synthetic signal that the
-            // reducer picks up on the next step.
+            // Codex follow-up (post-#50 #B): responsibility for synthesising the durable
+            // EffectInfrastructureFailure signal moved from EffectRunner to SignalIngress
+            // (see SignalIngressTests.Abort_signal_is_synthesised_*). The EffectRunner's
+            // contract is now "return an abort result; do not touch the ingress queue".
            var rig = new Rig();
            rig.Scheduler.ThrowOnSchedule = new InvalidOperationException("timer-broken");
            var sut = rig.Build();
@@ -106,18 +107,12 @@ namespace AutopilotMonitor.Agent.V2.Core.Tests.Orchestration
            var result = await sut.RunAsync(new[] { effect }, InitialState(), At);

            Assert.True(result.SessionMustAbort);
-            Assert.Single(rig.Ingress.Posted);
-            var posted = rig.Ingress.Posted[0];
-            Assert.Equal(DecisionSignalKind.EffectInfrastructureFailure, posted.Kind);
-            Assert.Equal(At, posted.OccurredAtUtc);
-            Assert.Equal("effectrunner:critical:ScheduleDeadline", posted.SourceOrigin);
-            Assert.NotNull(posted.Payload);
-            Assert.Equal(result.AbortReason, posted.Payload!["reason"]);
-            Assert.Equal("ScheduleDeadline", posted.Payload["failingEffect"]);
+            Assert.Contains("timer_infrastructure_failure", result.AbortReason);
+            Assert.Empty(rig.Ingress.Posted); // no longer the EffectRunner's job
        }

        [Fact]
-        public async Task CancelDeadline_failure_posts_EffectInfrastructureFailure_signal_to_ingress()
+        public async Task CancelDeadline_failure_does_not_post_signal_from_EffectRunner()
        {
            var rig = new Rig();
            rig.Scheduler.ThrowOnCancel = new InvalidOperationException("timer-broken");
@@ -129,31 +124,7 @@ namespace AutopilotMonitor.Agent.V2.Core.Tests.Orchestration
            var result = await sut.RunAsync(new[] { effect }, InitialState(), At);

            Assert.True(result.SessionMustAbort);
-            Assert.Single(rig.Ingress.Posted);
-            var posted = rig.Ingress.Posted[0];
-            Assert.Equal(DecisionSignalKind.EffectInfrastructureFailure, posted.Kind);
-            Assert.Equal("effectrunner:critical:CancelDeadline", posted.SourceOrigin);
-            Assert.Equal("CancelDeadline", posted.Payload!["failingEffect"]);
-        }
-
-        [Fact]
-        public async Task Critical_failure_post_tolerates_ingress_exceptions()
-        {
-            // The max-lifetime watchdog is the last-resort safety net — if the synthetic
-            // post itself throws (e.g. channel closed), the EffectRunner must still
-            // return cleanly rather than crashing the ingress worker.
-            var rig = new Rig();
-            rig.Scheduler.ThrowOnSchedule = new InvalidOperationException("timer-broken");
-            rig.Ingress.ThrowOnPost = new InvalidOperationException("ingress-full");
-            var sut = rig.Build();
-            var effect = new DecisionEffect(
-                DecisionEffectKind.ScheduleDeadline,
-                deadline: Deadline("x", At.AddMinutes(1)));
-
-            var result = await sut.RunAsync(new[] { effect }, InitialState(), At);
-
-            Assert.True(result.SessionMustAbort);
-            // Ingress-Post throw was swallowed — no unhandled exception here.
+            Assert.Empty(rig.Ingress.Posted);
        }

        [Fact]
@@ -17,6 +17,7 @@ namespace AutopilotMonitor.Agent.V2.Core.Tests.Orchestration
    {
        private readonly object _lock = new object();
        private readonly Queue<Exception?> _script = new Queue<Exception?>();
+        private readonly Queue<EffectRunResult> _resultScript = new Queue<EffectRunResult>();
        private readonly List<ApplyCall> _calls = new List<ApplyCall>();
        private DecisionState _currentState;
        private int _failedApplyCount;
@@ -62,7 +63,20 @@ namespace AutopilotMonitor.Agent.V2.Core.Tests.Orchestration
            return this;
        }

-        public void ApplyStep(DecisionStep step, DecisionSignal signal)
+        /// <summary>
+        /// Script a specific <see cref="EffectRunResult"/> for the next ApplyStep call(s) —
+        /// lets tests drive the SignalIngress inline-abort path (Codex post-#50 #B).
+        /// </summary>
+        public FakeDecisionStepProcessor ScriptResult(EffectRunResult result, int count = 1)
+        {
+            lock (_lock)
+            {
+                for (int i = 0; i < count; i++) _resultScript.Enqueue(result);
+            }
+            return this;
+        }
+
+        public EffectRunResult ApplyStep(DecisionStep step, DecisionSignal signal)
        {
            BlockHandle?.Wait();

@@ -79,6 +93,7 @@ namespace AutopilotMonitor.Agent.V2.Core.Tests.Orchestration
                }
                _calls.Add(new ApplyCall(step, signal));
                _currentState = step.NewState;
+                return _resultScript.Count > 0 ? _resultScript.Dequeue() : EffectRunResult.Empty();
            }
        }

@@ -262,6 +262,50 @@ namespace AutopilotMonitor.Agent.V2.Core.Tests.Orchestration
            sut.Stop();
        }

+        [Fact]
+        public void No_snapshot_with_intact_log_rebuilds_state_via_full_log_replay()
+        {
+            // Codex follow-up (post-#50 #A): the snapshot was never written (either an
+            // abnormal first-session crash or post-quarantine wipe) BUT the SignalLog
+            // already has entries. Branch (c) must NOT fall through to CreateInitial —
+            // that would reset StepIndex=0 while the log continues at its original
+            // ordinals, breaking monotonicity on the next live append.
+            using var rig = new Rig();
+            Directory.CreateDirectory(rig.StateDir);
+
+            // Seed three signals on the log, NO snapshot file on disk.
+            var sigs = new[]
+            {
+                MakeSignal(0, DecisionSignalKind.SessionStarted),
+                MakeSignal(1, DecisionSignalKind.AppInstallCompleted),
+                MakeSignal(2, DecisionSignalKind.AppInstallCompleted),
+            };
+            var seedLog = new SignalLogWriter(Path.Combine(rig.StateDir, "signal-log.jsonl"));
+            foreach (var s in sigs) seedLog.Append(s);
+
+            Assert.False(File.Exists(Path.Combine(rig.StateDir, "snapshot.json")));
+
+            var engine = new DecisionEngine();
+            var expected = ReducerReplay.Replay(
+                engine, DecisionState.CreateInitial("S1", "T1"), sigs);
+
+            var sut = rig.Build();
+            sut.Start();
+
+            // Full-log replay produced the post-crash state — step/ordinal both at tail.
+            Assert.True(sut.WasStartupQuarantine);
+            Assert.Equal(expected.StepIndex, sut.CurrentState.StepIndex);
+            Assert.Equal(expected.LastAppliedSignalOrdinal, sut.CurrentState.LastAppliedSignalOrdinal);
+            Assert.Equal(expected.Stage, sut.CurrentState.Stage);
+
+            // SignalLog was NOT quarantined — only the absent snapshot path triggered replay.
+            var signalLogPath = Path.Combine(rig.StateDir, "signal-log.jsonl");
+            Assert.True(File.Exists(signalLogPath));
+            Assert.True(new FileInfo(signalLogPath).Length > 0);
+
+            sut.Stop();
+        }
+
        [Fact]
        public void Corrupt_snapshot_with_intact_log_replays_full_log_without_quarantining_log()
        {
@@ -308,15 +352,15 @@ namespace AutopilotMonitor.Agent.V2.Core.Tests.Orchestration
        [Fact]
        public void Journal_ahead_of_replayed_state_triggers_phantom_truncate()
        {
-            // Pathological crash scenario: the Journal flushed step N but the Snapshot saved
-            // only step N-1 and a later SignalLog append was never actually persisted (log
-            // tail physically shorter than journal). On recovery the journal suffix is a
-            // phantom — TruncateAfter moves it to .quarantine and realigns so the next live
-            // append does not violate monotonicity.
+            // AHEAD crash scenario: the Journal flushed step for sig1, then also a phantom
+            // step for a signal the SignalLog never persisted, while the Snapshot saved
+            // only step for sig0. On recovery everything beyond seed.StepIndex goes to the
+            // phantom bucket and the replay callback (Codex follow-up post-#50 #C) rebuilds
+            // the authoritative tail bit-for-bit in lockstep with state advancement.
            using var rig = new Rig();
            Directory.CreateDirectory(rig.StateDir);

-            // Step 1: SignalLog has ordinals 0 + 1 only.
+            // Step 1: SignalLog has sig0 + sig1.
            var sigs = new[]
            {
                MakeSignal(0, DecisionSignalKind.SessionStarted),
@@ -325,21 +369,25 @@ namespace AutopilotMonitor.Agent.V2.Core.Tests.Orchestration
            var seedLog = new SignalLogWriter(Path.Combine(rig.StateDir, "signal-log.jsonl"));
            foreach (var s in sigs) seedLog.Append(s);

-            // Step 2: Snapshot lags behind — state after only the first reduce.
+            // Step 2: Produce real engine transitions for sig0 + sig1 — engine semantics
+            // say their StepIndex is 1 and 2 (state.StepIndex + 1). Snapshot only captures
+            // post-sig0 state; the phantom transition for a non-existent sig beyond has
+            // StepIndex=3.
            var engine = new DecisionEngine();
-            var snapshotState = engine.Reduce(
-                DecisionState.CreateInitial("S1", "T1"), sigs[0]).NewState;
+            var step0 = engine.Reduce(DecisionState.CreateInitial("S1", "T1"), sigs[0]);
+            var step1 = engine.Reduce(step0.NewState, sigs[1]);
+            var snapshotState = step0.NewState; // StepIndex=1
            new SnapshotPersistence(
                Path.Combine(rig.StateDir, "snapshot.json"),
                () => rig.Clock.UtcNow).Save(snapshotState);

-            // Step 3: Journal has 3 transitions on disk (StepIndex 0, 1, 2) — the StepIndex-2
-            // entry is the phantom (its signal was never flushed to the log).
+            // Step 3: Journal has 3 transitions on disk — the StepIndex=3 one is phantom
+            // (no backing signal in the log).
            var journalPath = Path.Combine(rig.StateDir, "journal.jsonl");
            var seedJournal = new JournalWriter(journalPath, () => rig.Clock.UtcNow);
-            seedJournal.Append(MakeTransition(0, signalRef: 0));
-            seedJournal.Append(MakeTransition(1, signalRef: 1));
-            seedJournal.Append(MakeTransition(2, signalRef: 99)); // phantom
+            seedJournal.Append(step0.Transition);                // StepIndex=1
+            seedJournal.Append(step1.Transition);                // StepIndex=2 — pre-crash flushed
+            seedJournal.Append(MakeTransition(3, signalRef: 99)); // StepIndex=3 — phantom

            // Expected post-recovery state: replay tail [sig1] → StepIndex=2, LastApplied=1.
            var expected = ReducerReplay.Replay(engine, snapshotState, new[] { sigs[1] });
@@ -351,13 +399,80 @@ namespace AutopilotMonitor.Agent.V2.Core.Tests.Orchestration
            Assert.Equal(expected.StepIndex, sut.CurrentState.StepIndex);
            Assert.Equal(expected.LastAppliedSignalOrdinal, sut.CurrentState.LastAppliedSignalOrdinal);

-            // Phantom file captured in the quarantine bucket.
+            // Phantom bucket contains the two entries beyond seed.StepIndex=1 (StepIndex 2 + 3):
+            // the pre-crash-flushed real transition AND the signal-less phantom. The replay
+            // then rebuilt StepIndex=2 from sig1 via the onTransition callback so the live
+            // journal ends in lockstep with state at [1, 2].
            var quarantineRoot = Path.Combine(rig.StateDir, ".quarantine");
            Assert.True(Directory.Exists(quarantineRoot));
            var phantomFiles = Directory.GetFiles(
                quarantineRoot, "journal-phantom-tail.jsonl", SearchOption.AllDirectories);
            Assert.Single(phantomFiles);
-            Assert.Single(File.ReadAllLines(phantomFiles[0]), l => !string.IsNullOrWhiteSpace(l));
+            Assert.Equal(2, File.ReadAllLines(phantomFiles[0])
+                .Count(l => !string.IsNullOrWhiteSpace(l)));
+
+            var liveJournal = new JournalWriter(journalPath, () => rig.Clock.UtcNow).ReadAll();
+            Assert.Equal(2, liveJournal.Count);
+            Assert.Equal(1, liveJournal[0].StepIndex);
+            Assert.Equal(2, liveJournal[1].StepIndex);
+
+            sut.Stop();
+        }
+
+        [Fact]
+        public void Signal_log_ahead_of_journal_rebuilds_missing_journal_entries_during_replay()
+        {
+            // Codex follow-up (post-#50 #C): BEHIND scenario. A crash landed between
+            // SignalLog.Append(sig1) and Journal.Append(step1.Transition), so the log is one
+            // step ahead of the journal while the snapshot already captures the pre-sig1
+            // state. Recovery must NOT leave the journal with a gap — the onTransition
+            // callback on ReducerReplay.Replay backfills the missing StepIndex in lockstep.
+            using var rig = new Rig();
+            Directory.CreateDirectory(rig.StateDir);
+
+            var sigs = new[]
+            {
+                MakeSignal(0, DecisionSignalKind.SessionStarted),
+                MakeSignal(1, DecisionSignalKind.AppInstallCompleted),
+            };
+            var seedLog = new SignalLogWriter(Path.Combine(rig.StateDir, "signal-log.jsonl"));
+            foreach (var s in sigs) seedLog.Append(s);
+
+            var engine = new DecisionEngine();
+            var step0 = engine.Reduce(DecisionState.CreateInitial("S1", "T1"), sigs[0]);
+            var snapshotState = step0.NewState; // StepIndex=1, LastApplied=0
+            new SnapshotPersistence(
+                Path.Combine(rig.StateDir, "snapshot.json"),
+                () => rig.Clock.UtcNow).Save(snapshotState);
+
+            // Journal only has the sig0 transition — the sig1 transition was never flushed
+            // before the crash. LastStepIndex=1 matches snapshotState.StepIndex, so no
+            // AHEAD-truncation fires; recovery must backfill StepIndex=2 via the callback.
+            var journalPath = Path.Combine(rig.StateDir, "journal.jsonl");
+            var seedJournal = new JournalWriter(journalPath, () => rig.Clock.UtcNow);
+            seedJournal.Append(step0.Transition); // StepIndex=1
+
+            var expected = ReducerReplay.Replay(engine, snapshotState, new[] { sigs[1] });
+
+            var sut = rig.Build();
+            sut.Start();
+
+            Assert.Equal(expected.StepIndex, sut.CurrentState.StepIndex);
+            Assert.Equal(expected.LastAppliedSignalOrdinal, sut.CurrentState.LastAppliedSignalOrdinal);
+
+            var liveJournal = new JournalWriter(journalPath, () => rig.Clock.UtcNow).ReadAll();
+            Assert.Equal(2, liveJournal.Count);
+            Assert.Equal(1, liveJournal[0].StepIndex);
+            Assert.Equal(2, liveJournal[1].StepIndex);
+
+            // No phantom quarantine happened — nothing was beyond the seed boundary.
+            var quarantineRoot = Path.Combine(rig.StateDir, ".quarantine");
+            if (Directory.Exists(quarantineRoot))
+            {
+                var phantomFiles = Directory.GetFiles(
+                    quarantineRoot, "journal-phantom-tail.jsonl", SearchOption.AllDirectories);
+                Assert.Empty(phantomFiles);
+            }

            sut.Stop();
        }
@@ -233,10 +233,11 @@ namespace AutopilotMonitor.Agent.V2.Core.Tests.Orchestration

            public DecisionCore.State.DecisionState CurrentState => _state;

-            public void ApplyStep(DecisionStep step, DecisionSignal signal)
+            public EffectRunResult ApplyStep(DecisionStep step, DecisionSignal signal)
            {
                _inspect(step, signal, _log.LastOrdinal);
                _state = step.NewState;
+                return EffectRunResult.Empty();
            }
        }

@@ -283,6 +284,85 @@ namespace AutopilotMonitor.Agent.V2.Core.Tests.Orchestration
            Assert.Equal(2, rig.SignalLog.ReadAll().Count);
        }

+        // ============================================================ Codex follow-up post-#50 #B — durable inline abort
+
+        [Fact]
+        public void SessionMustAbort_synthesises_EffectInfrastructureFailure_signal_durably_before_worker_returns()
+        {
+            // When the processor returns SessionMustAbort=true, the ingress must:
+            //  (1) append a synthetic EffectInfrastructureFailure signal to the SignalLog,
+            //  (2) invoke the processor with that signal a second time (terminal transition),
+            // all within the current worker iteration — so a crash after the iteration
+            // returns can recover to a terminal state, not a phantom deadline state.
+            using var rig = new Rig();
+            var abort = new EffectRunResult(
+                sessionMustAbort: true,
+                abortReason: "timer_infrastructure_failure: scheduler-died",
+                failures: Array.Empty<EffectFailure>(),
+                classifierInvocations: 0,
+                classifierSkippedByAntiLoop: 0);
+            rig.Processor.ScriptResult(abort);
+
+            var ing = rig.Build();
+            ing.Start();
+            ing.Post(DecisionSignalKind.SessionStarted, At, "Collector", RawEvidence("original"));
+
+            // Wait for BOTH the original apply and the synthetic abort apply to land.
+            Assert.True(WaitFor(() => rig.Processor.ApplyCallCount >= 2, timeoutMs: 5000));
+            ing.Stop();
+
+            // SignalLog has [original, synthetic] — both durable, monotone ordinals.
+            var logged = rig.SignalLog.ReadAll();
+            Assert.Equal(2, logged.Count);
+            Assert.Equal(DecisionSignalKind.SessionStarted, logged[0].Kind);
+            Assert.Equal(DecisionSignalKind.EffectInfrastructureFailure, logged[1].Kind);
+            Assert.Equal(0L, logged[0].SessionSignalOrdinal);
+            Assert.Equal(1L, logged[1].SessionSignalOrdinal);
+
+            // Synthetic signal payload echoes the abort reason for forensics.
+            Assert.NotNull(logged[1].Payload);
+            Assert.Equal(abort.AbortReason, logged[1].Payload!["reason"]);
+            Assert.Equal("signal-ingress:inline-abort", logged[1].SourceOrigin);
+
+            // Processor received both signals, in order, with matching signals.
+            var calls = rig.Processor.Calls;
+            Assert.True(calls.Count >= 2);
+            Assert.Equal(DecisionSignalKind.SessionStarted, calls[0].Signal.Kind);
+            Assert.Equal(DecisionSignalKind.EffectInfrastructureFailure, calls[1].Signal.Kind);
+        }
+
+        [Fact]
+        public void Synthetic_abort_signal_itself_reporting_abort_does_not_recurse()
+        {
+            // Defence-in-depth against infinite recursion: if ApplyStep for the synthetic
+            // EffectInfrastructureFailure signal were (wrongly) to report another abort,
+            // ingress must NOT produce a third synthetic signal — the reducer's terminal
+            // handler is trusted as the end of the abort chain.
+            using var rig = new Rig();
+            var abortEverything = new EffectRunResult(
+                sessionMustAbort: true,
+                abortReason: "simulated: every step aborts",
+                failures: Array.Empty<EffectFailure>(),
+                classifierInvocations: 0,
+                classifierSkippedByAntiLoop: 0);
+            // Two scripted abort results: one for the original signal, one for the
+            // synthetic. The third call (if any) would use the default Empty result.
+            rig.Processor.ScriptResult(abortEverything, count: 2);
+
+            var ing = rig.Build();
+            ing.Start();
+            ing.Post(DecisionSignalKind.SessionStarted, At, "Collector", RawEvidence("original"));
+
+            Assert.True(WaitFor(() => rig.Processor.ApplyCallCount >= 2, timeoutMs: 5000));
+            // Give the worker a moment to potentially recurse (it must not).
+            Thread.Sleep(150);
+            ing.Stop();
+
+            // Exactly 2 apply calls — original + one synthetic. No recursion.
+            Assert.Equal(2, rig.Processor.ApplyCallCount);
+            Assert.Equal(2, rig.SignalLog.ReadAll().Count);
+        }
+
        // ============================================================ Codex Finding 4 — SignalPosted

        [Fact]
@@ -98,7 +98,7 @@ namespace AutopilotMonitor.Agent.V2.Core.Orchestration
        /// <summary>Test-Observability — anzahl aufeinanderfolgender Journal-Failures.</summary>
        public int ConsecutiveJournalFailureCount => _consecutiveJournalFailures;

-        public void ApplyStep(DecisionStep step, DecisionSignal signal)
+        public EffectRunResult ApplyStep(DecisionStep step, DecisionSignal signal)
        {
            if (step == null) throw new ArgumentNullException(nameof(step));
            if (signal == null) throw new ArgumentNullException(nameof(signal));
@@ -160,35 +160,43 @@ namespace AutopilotMonitor.Agent.V2.Core.Orchestration

            if (effectResult.SessionMustAbort)
            {
-                // Codex follow-up #2: EffectRunner has already posted a synthetic
-                // EffectInfrastructureFailure signal to the ingress. The reducer will
-                // transition the session to Failed / EnrollmentFailed on the next
-                // dispatched step — our job here is purely observability.
+                // Codex follow-up (post-#50 #B): the phantom state must NOT be snapshotted
+                // because its ActiveDeadline was never actually armed on the live scheduler.
+                // Recovery from a phantom snapshot would try to re-arm and re-fail, leaving
+                // the session dangling for the max-lifetime watchdog. Responsibility for
+                // flipping the session to EnrollmentFailed lives with the caller
+                // (SignalIngress): it synthesises + DURABLY appends the
+                // EffectInfrastructureFailure signal, re-enters ApplyStep, and the terminal
+                // step's snapshot replaces the stale N-1 snapshot.
                _logger.Error(
                    $"DecisionStepProcessor: effect run signaled session abort " +
                    $"(reason='{effectResult.AbortReason}') for signal ordinal={signal.SessionSignalOrdinal}; " +
-                    $"EffectInfrastructureFailure signal enqueued, reducer will terminate next step.");
+                    $"skipping snapshot of phantom state — caller will synthesise EffectInfrastructureFailure durably.");
            }
-            else if (effectResult.Failures.Count > 0)
+            else
            {
-                _logger.Warning(
-                    $"DecisionStepProcessor: effect run completed with {effectResult.Failures.Count} non-fatal failure(s) " +
-                    $"for signal ordinal={signal.SessionSignalOrdinal}.");
+                if (effectResult.Failures.Count > 0)
+                {
+                    _logger.Warning(
+                        $"DecisionStepProcessor: effect run completed with {effectResult.Failures.Count} non-fatal failure(s) " +
+                        $"for signal ordinal={signal.SessionSignalOrdinal}.");
+                }
+
+                // 3) Snapshot — best-effort. Journal ist die Wahrheit, Snapshot ist Cache.
+                try
+                {
+                    _snapshot.Save(step.NewState);
+                }
+                catch (Exception ex)
+                {
+                    _logger.Warning(
+                        $"DecisionStepProcessor: snapshot save failed (transient, not escalated) " +
+                        $"for signal ordinal={signal.SessionSignalOrdinal}: {ex.Message}");
+                }
            }

-            // 3) Snapshot — best-effort. Journal ist die Wahrheit, Snapshot ist Cache.
-            try
-            {
-                _snapshot.Save(step.NewState);
-            }
-            catch (Exception ex)
-            {
-                _logger.Warning(
-                    $"DecisionStepProcessor: snapshot save failed (transient, not escalated) " +
-                    $"for signal ordinal={signal.SessionSignalOrdinal}: {ex.Message}");
-            }
-
-            // 4) State-Forward + Counter-Reset.
+            // 4) State-Forward + Counter-Reset. Happens regardless of abort so the caller's
+            //    follow-up synthetic signal reduces FROM the correct pre-terminal state.
            _currentState = step.NewState;
            _consecutiveJournalFailures = 0;

@@ -208,6 +216,8 @@ namespace AutopilotMonitor.Agent.V2.Core.Orchestration
                        ex);
                }
            }
+
+            return effectResult;
        }

        private void TryTriggerQuarantine(string reason)
@@ -81,13 +81,11 @@ namespace AutopilotMonitor.Agent.V2.Core.Orchestration
                            failures.Add(criticalFail);
                            var abortReason = $"timer_infrastructure_failure: {criticalFail.ErrorReason}";

-                            // Codex follow-up #2: post a synthetic EffectInfrastructureFailure
-                            // signal so the reducer can transition cleanly to Failed /
-                            // EnrollmentFailed on the next step. Without this the flag would
-                            // only surface in logs and the session could hang on a phantom
-                            // deadline that was never actually armed.
-                            PostEffectInfrastructureFailureSignal(effect.Kind, abortReason, stepOccurredAtUtc);
-
+                            // Codex follow-up (post-#50 #B): responsibility for posting the
+                            // synthetic EffectInfrastructureFailure signal moved to SignalIngress
+                            // so it can DURABLY append + reduce inline within the same worker
+                            // iteration, closing the crash-window between in-memory post and
+                            // SignalLog persistence. EffectRunner just signals the abort here.
                            return new EffectRunResult(
                                sessionMustAbort: true,
                                abortReason: abortReason,
@@ -229,41 +227,6 @@ namespace AutopilotMonitor.Agent.V2.Core.Orchestration
            return new ClassifierEffectOutcome(invoked: true, skippedByAntiLoop: false, failure: null);
        }

-        /// <summary>
-        /// Codex follow-up #2 — synthesise an <see cref="DecisionSignalKind.EffectInfrastructureFailure"/>
-        /// signal when a critical deadline effect cannot be delivered. The reducer's
-        /// <c>HandleEffectInfrastructureFailureV1</c> picks this up next tick and transitions the
-        /// session to <see cref="SessionStage.Failed"/> / <see cref="SessionOutcome.EnrollmentFailed"/>.
-        /// Any exception from the ingress (full channel / disposed) is swallowed: the
-        /// max-lifetime watchdog remains the last-resort termination path.
-        /// </summary>
-        private void PostEffectInfrastructureFailureSignal(
-            DecisionEffectKind failingEffect,
-            string abortReason,
-            DateTime stepOccurredAtUtc)
-        {
-            try
-            {
-                _ingress.Post(
-                    kind: DecisionSignalKind.EffectInfrastructureFailure,
-                    occurredAtUtc: stepOccurredAtUtc,
-                    sourceOrigin: $"effectrunner:critical:{failingEffect}",
-                    evidence: new Evidence(
-                        kind: EvidenceKind.Synthetic,
-                        identifier: $"effect_infrastructure_failure:{failingEffect}",
-                        summary: $"Critical effect {failingEffect} failed: {abortReason}"),
-                    payload: new Dictionary<string, string>(StringComparer.Ordinal)
-                    {
-                        ["reason"] = abortReason,
-                        ["failingEffect"] = failingEffect.ToString(),
-                    });
-            }
-            catch
-            {
-                // Intentionally swallowed — see method doc.
-            }
-        }
-
        private void PostClassifierVerdictSignal(ClassifierVerdict verdict, string classifierId, DateTime stepOccurredAtUtc)
        {
            _ingress.Post(
@@ -279,13 +279,23 @@ namespace AutopilotMonitor.Agent.V2.Core.Orchestration
            //         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 → fresh initial state.
+            //      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 is chosen we ensure the Journal tail is not AHEAD of the
-            //    replayed state — if a crash left phantom transitions on disk (Journal flushed,
-            //    Snapshot not) the next live Append would throw a monotonicity violation and
-            //    eventually quarantine-escalate. TruncateAfter moves the phantoms to a
-            //    forensic bucket and realigns.
+            //    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;
@@ -294,6 +304,12 @@ namespace AutopilotMonitor.Agent.V2.Core.Orchestration
            // 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.
+            DecisionState seed = DecisionState.CreateInitial(_sessionId, _tenantId);
+            IReadOnlyList<DecisionSignal> signalsToReplay = Array.Empty<DecisionSignal>();
+            string branchTag;
+
            if (loadedState == null && snapshotFileExistsPreLoad)
            {
                // Branch (a) — Snapshot corrupt. Quarantine the snapshot and attempt log replay.
@@ -319,16 +335,12 @@ namespace AutopilotMonitor.Agent.V2.Core.Orchestration
                    _journal = new JournalWriter(journalPath, () => _clock.UtcNow);
                    _eventSequencePersistence = new EventSequencePersistence(eventSequencePath);

-                    initialState = DecisionState.CreateInitial(_sessionId, _tenantId);
+                    branchTag = "a-total-loss";
                }
                else
                {
-                    _logger.Info(
-                        $"EnrollmentOrchestrator: rebuilding state via ReducerReplay over {loggedSignals.Count} SignalLog entries.");
-                    initialState = ReducerReplay.Replay(
-                        engine: replayEngine,
-                        seed: DecisionState.CreateInitial(_sessionId, _tenantId),
-                        signals: loggedSignals);
+                    signalsToReplay = loggedSignals;
+                    branchTag = "a-full-log-replay";
                }

                _wasStartupQuarantine = true;
@@ -336,43 +348,63 @@ namespace AutopilotMonitor.Agent.V2.Core.Orchestration
            else if (loadedState != null)
            {
                // Branch (b) — Snapshot loaded. Catch up any SignalLog tail past the snapshot.
-                var tail = CollectSignalLogTailAfter(loadedState.LastAppliedSignalOrdinal);
-
-                if (tail.Count > 0)
-                {
-                    _logger.Info(
-                        $"EnrollmentOrchestrator: snapshot at ordinal={loadedState.LastAppliedSignalOrdinal}; " +
-                        $"replaying {tail.Count} SignalLog tail signal(s) to catch up.");
-                    initialState = ReducerReplay.Replay(replayEngine, loadedState, tail);
-                }
-                else
-                {
-                    initialState = loadedState;
-                    _logger.Info(
-                        $"EnrollmentOrchestrator: recovered state from snapshot (stage={loadedState.Stage}, stepIndex={loadedState.StepIndex}).");
-                }
+                seed = loadedState;
+                signalsToReplay = CollectSignalLogTailAfter(loadedState.LastAppliedSignalOrdinal);
+                branchTag = signalsToReplay.Count > 0 ? "b-tail-replay" : "b-snapshot-current";
            }
            else
            {
-                // Branch (c) — no prior state.
-                initialState = DecisionState.CreateInitial(_sessionId, _tenantId);
-                _logger.Info($"EnrollmentOrchestrator: fresh initial state for session {_sessionId}.");
+                // 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";
+                }
            }

-            // Post-recovery alignment (plan §5.1): truncate Journal phantom suffix if the file
-            // is ahead of the replayed state. After a normal Reduce the Journal holds entries
-            // with StepIndex = [0 .. state.StepIndex - 1]; anything beyond is a leftover from a
-            // pre-crash Journal-flushed-but-snapshot-not path.
-            if (_journal.LastStepIndex >= initialState.StepIndex)
+            // 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)
            {
-                var boundary = initialState.StepIndex - 1;
                _logger.Warning(
-                    $"EnrollmentOrchestrator: Journal tail ahead of replayed state " +
-                    $"(journal.LastStepIndex={_journal.LastStepIndex}, state.StepIndex={initialState.StepIndex}) — " +
-                    $"truncating phantom transitions to boundary={boundary}.");
-                _journal.TruncateAfter(boundary);
+                    $"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}.");
+
            // 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
@@ -34,10 +34,17 @@ namespace AutopilotMonitor.Agent.V2.Core.Orchestration
        /// <list type="number">
        ///   <item>Transition ins Journal appenden (Sofort-Flush)</item>
        ///   <item>Effekte mittels <see cref="IEffectRunner"/> ausführen</item>
+        ///   <item>Snapshot der <see cref="DecisionStep.NewState"/> speichern — ABER nur wenn
+        ///         <see cref="EffectRunResult.SessionMustAbort"/> false ist (sonst wäre der
+        ///         Snapshot ein phantom-Zustand mit einer Deadline, die der live Scheduler
+        ///         nie arm-en konnte — siehe Codex follow-up post-#50 #B).</item>
        ///   <item><see cref="CurrentState"/> auf <see cref="DecisionStep.NewState"/> fortsetzen</item>
        /// </list>
-        /// Darf werfen — Ingress fängt ab und loggt, verstummt aber nicht.
+        /// Gibt das <see cref="EffectRunResult"/> zurück, damit der Caller (typischerweise
+        /// <see cref="SignalIngress"/>) entscheiden kann, ob ein synthetisches
+        /// <see cref="DecisionSignalKind.EffectInfrastructureFailure"/> inline nachgeschoben
+        /// werden muss. Darf werfen — Ingress fängt ab und loggt, verstummt aber nicht.
        /// </summary>
-        void ApplyStep(DecisionStep step, DecisionSignal signal);
+        EffectRunResult ApplyStep(DecisionStep step, DecisionSignal signal);
    }
 }
@@ -341,9 +341,10 @@ namespace AutopilotMonitor.Agent.V2.Core.Orchestration
            // IDecisionEngine.Reduce ist fail-safe (DecisionEngine.cs:39-47): wirft nicht.
            var step = _engine.Reduce(state, signal);

+            EffectRunResult? effectResult = null;
            try
            {
-                _processor.ApplyStep(step, signal);
+                effectResult = _processor.ApplyStep(step, signal);
            }
            catch
            {
@@ -351,6 +352,96 @@ namespace AutopilotMonitor.Agent.V2.Core.Orchestration
                // M4.4.5 vom Orchestrator geloggt + ggf. Quarantine ausgelöst. M4.4.0
                // verschluckt bewusst, damit der Ingress-Worker nicht verstummt.
            }
+
+            // Codex follow-up (post-#50 #B): inline durable-abort path. When the step's
+            // effects signal a critical infrastructure failure (e.g. timer scheduler died)
+            // we must land a synthetic EffectInfrastructureFailure signal on the SignalLog
+            // BEFORE returning to the worker loop — otherwise a crash between here and the
+            // next Post() would leave recovery replaying the phantom deadline state with no
+            // terminal signal to bridge it to SessionStage.Failed.
+            //
+            // The recursion guard prevents processing an abort that the synthetic signal's
+            // own reducer handler might somehow report (shouldn't happen — HandleEffect-
+            // InfrastructureFailureV1 transitions to terminal without queuing critical
+            // effects — but the guard makes that invariant failure-safe).
+            if (effectResult != null
+                && effectResult.SessionMustAbort
+                && signal.Kind != DecisionSignalKind.EffectInfrastructureFailure)
+            {
+                ProcessInlineAbortSignal(effectResult.AbortReason, signal.OccurredAtUtc);
+            }
+        }
+
+        /// <summary>
+        /// Synthesise + durably persist + reduce a <c>EffectInfrastructureFailure</c> signal
+        /// inline within the current worker iteration. See ProcessItem for the rationale —
+        /// this closes the crash-window identified by Codex follow-up post-#50 #B.
+        /// </summary>
+        private void ProcessInlineAbortSignal(string? abortReason, DateTime occurredAtUtc)
+        {
+            long signalOrdinal = _nextSignalOrdinal;
+            long traceOrdinal = _traceCounter.Next();
+
+            DecisionSignal syntheticSignal;
+            try
+            {
+                syntheticSignal = new DecisionSignal(
+                    sessionSignalOrdinal: signalOrdinal,
+                    sessionTraceOrdinal: traceOrdinal,
+                    kind: DecisionSignalKind.EffectInfrastructureFailure,
+                    kindSchemaVersion: 1,
+                    occurredAtUtc: occurredAtUtc,
+                    sourceOrigin: "signal-ingress:inline-abort",
+                    evidence: new Evidence(
+                        kind: EvidenceKind.Synthetic,
+                        identifier: "effect_infrastructure_failure:inline",
+                        summary: $"Inline abort synthesised by SignalIngress: {abortReason ?? "<unspecified>"}"),
+                    payload: new Dictionary<string, string>(StringComparer.Ordinal)
+                    {
+                        ["reason"] = abortReason ?? string.Empty,
+                    });
+            }
+            catch
+            {
+                // DecisionSignal ctor validation must not fail for a locally-constructed
+                // signal, but defensively bail — the max-lifetime watchdog remains the
+                // last-resort termination path.
+                return;
+            }
+
+            try
+            {
+                _signalLog.Append(syntheticSignal);
+            }
+            catch
+            {
+                // Persist-Failure — we must NOT advance the ordinal (log gap) and cannot
+                // safely reduce. Recovery will see the phantom state; the watchdog will
+                // clean up the hung session.
+                return;
+            }
+
+            _nextSignalOrdinal = signalOrdinal + 1;
+
+            if (_signalEmitter != null)
+            {
+                try { _signalEmitter.Emit(syntheticSignal); }
+                catch { /* best-effort upload */ }
+            }
+
+            var state = _processor.CurrentState;
+            var step = _engine.Reduce(state, syntheticSignal);
+
+            try
+            {
+                _processor.ApplyStep(step, syntheticSignal);
+            }
+            catch
+            {
+                // Processor fault on the terminal transition — Journal has the abort
+                // transition committed (reducer produces Stage=Failed unconditionally).
+                // Swallowed by convention; worker must not die from this path.
+            }
        }

        private readonly struct IngressItem
@@ -1 +1 @@
-257
+293
@@ -1 +1 @@
-1120
+1134
@@ -175,6 +175,82 @@ namespace AutopilotMonitor.DecisionCore.Tests
            Assert.Contains("2", ex.Message);
        }

+        [Fact]
+        public void Replay_OnTransitionCallback_InvokedOncePerSignalInOrder()
+        {
+            // Codex follow-up (post-#50 #C): the onTransition overload exposes each step's
+            // transition to callers so they can rematerialise downstream artefacts (e.g.,
+            // the JournalWriter tail after a BEHIND-the-log crash). Invocation contract:
+            //   1. exactly one callback per signal,
+            //   2. invoked BEFORE the state advances to the next signal,
+            //   3. StepIndex strictly increasing with the signal order.
+            var engine = new DecisionEngine();
+            var seed = DecisionState.CreateInitial("s", "t");
+            var signals = MakeCanonicalStream();
+
+            var captured = new List<DecisionTransition>();
+            var result = ReducerReplay.Replay(
+                engine, seed, signals, onTransition: captured.Add);
+
+            Assert.Equal(signals.Count, captured.Count);
+            for (var i = 0; i < captured.Count; i++)
+            {
+                // Transitions come out in signal order and ordinals are strictly increasing.
+                if (i > 0)
+                {
+                    Assert.True(
+                        captured[i].StepIndex > captured[i - 1].StepIndex,
+                        $"StepIndex regression between captured[{i - 1}]={captured[i - 1].StepIndex} and captured[{i}]={captured[i].StepIndex}.");
+                }
+                Assert.Equal(signals[i].SessionSignalOrdinal, captured[i].SignalOrdinalRef);
+            }
+
+            // Final state after the callback-capturing fold matches the plain 3-arg fold.
+            var plain = ReducerReplay.Replay(engine, seed, signals);
+            Assert.Equal(plain.StepIndex, result.StepIndex);
+            Assert.Equal(plain.LastAppliedSignalOrdinal, result.LastAppliedSignalOrdinal);
+            Assert.Equal(plain.Stage, result.Stage);
+        }
+
+        [Fact]
+        public void Replay_OnTransitionCallback_NullIsEquivalentTo3ArgOverload()
+        {
+            // The nullable callback must behave like the 3-arg overload when omitted,
+            // preserving zero-side-effect determinism for callers that only need state.
+            var engine = new DecisionEngine();
+            var seed = DecisionState.CreateInitial("s", "t");
+            var signals = MakeCanonicalStream();
+
+            var withCallback = ReducerReplay.Replay(engine, seed, signals, onTransition: null);
+            var without = ReducerReplay.Replay(engine, seed, signals);
+
+            Assert.Equal(without.StepIndex, withCallback.StepIndex);
+            Assert.Equal(without.LastAppliedSignalOrdinal, withCallback.LastAppliedSignalOrdinal);
+            Assert.Equal(without.Stage, withCallback.Stage);
+        }
+
+        [Fact]
+        public void Replay_OnTransitionCallbackThrows_AbortsReplayAndPropagates()
+        {
+            // Callback exceptions must not be swallowed — they indicate a downstream
+            // persistence failure (e.g., journal disk full during rebuild). The caller
+            // decides whether to escalate to quarantine or retry.
+            var engine = new DecisionEngine();
+            var seed = DecisionState.CreateInitial("s", "t");
+            var signals = MakeCanonicalStream();
+
+            var invocations = 0;
+            var ex = Assert.Throws<InvalidOperationException>(() =>
+                ReducerReplay.Replay(engine, seed, signals, onTransition: _ =>
+                {
+                    invocations++;
+                    if (invocations == 2) throw new InvalidOperationException("simulated disk-full");
+                }));
+
+            Assert.Contains("simulated disk-full", ex.Message);
+            Assert.Equal(2, invocations); // aborted before the 3rd signal
+        }
+
        [Fact]
        public void Replay_NullArgs_ThrowArgumentNullException()
        {
@@ -40,7 +40,7 @@ public class ReducerVerifierSemanticReplayTests
        };

        var engine = new DecisionEngine();
-        var state = DecisionState.CreateInitial(TenantId, SessionId);
+        var state = DecisionState.CreateInitial(SessionId, TenantId);
        var transitionRecords = new DecisionTransitionRecord[sigs.Length];

        for (var i = 0; i < sigs.Length; i++)
@@ -185,6 +185,29 @@ public class ReducerVerifierSemanticReplayTests
            i => i.Kind == "replay_final_stage_mismatch" && i.Severity == "Error");
    }

+    [Fact]
+    public void Replay_seed_preserves_session_and_tenant_identity_distinctly()
+    {
+        // Codex follow-up (post-#50 #D) regression: seeding CreateInitial with swapped
+        // (tenantId, sessionId) used to produce a state with mirrored IDs, yet all tests
+        // passed because the fixture AND the verifier agreed on the swap. Using sentinel
+        // IDs where the tenant and session tokens are distinguishable (different prefixes)
+        // catches the swap at the identity-check step the verifier now performs.
+        var (signals, transitions, _) = BuildAgreedStream();
+
+        var report = ReducerVerifier.Verify(TenantId, SessionId, signals, transitions, CurrentReducerVersion);
+
+        Assert.True(report.SemanticReplayPerformed);
+        Assert.Equal(TenantId, report.TenantId);
+        Assert.Equal(SessionId, report.SessionId);
+        Assert.DoesNotContain(report.Issues, i => i.Kind == "replay_identity_mismatch");
+        // Sanity on the sentinel prefixes (keeps the regression explicit if someone edits
+        // the constants and accidentally mirrors them).
+        Assert.StartsWith("a1b2", TenantId);
+        Assert.StartsWith("b2c3", SessionId);
+        Assert.NotEqual(TenantId, SessionId);
+    }
+
    [Fact]
    public void Replay_handles_mid_stream_signal_deserialization_failure_without_aborting()
    {
@@ -317,7 +317,10 @@ namespace AutopilotMonitor.Functions.Services
            }

            var engine = new DecisionEngine();
-            var state = DecisionState.CreateInitial(report.TenantId, report.SessionId);
+            // Codex follow-up (post-#50 #D): DecisionState.CreateInitial signature is
+            // (sessionId, tenantId); callers must pass them in that order so a replay seed
+            // carries the correct identity end-to-end.
+            var state = DecisionState.CreateInitial(report.SessionId, report.TenantId);
            var divergences = 0;

            var stepsToCompare = Math.Min(decodedSignals.Count, decodedTransitions.Count);
@@ -367,6 +370,21 @@ namespace AutopilotMonitor.Functions.Services
                });
            }

+            // Codex follow-up (post-#50 #D): defensive identity assertion. The final state
+            // must still carry the same SessionId/TenantId as the report — a drift here
+            // signals a swapped CreateInitial(...) seed upstream, which would otherwise slip
+            // through because TransitionsSemanticallyEqual is identity-insensitive.
+            if (!string.Equals(state.SessionId, report.SessionId, StringComparison.Ordinal)
+                || !string.Equals(state.TenantId, report.TenantId, StringComparison.Ordinal))
+            {
+                report.Issues.Add(new VerificationIssue
+                {
+                    Severity = "Error",
+                    Kind     = "replay_identity_mismatch",
+                    Message  = $"Replay seed identity drift: final state (session={state.SessionId}, tenant={state.TenantId}) != report (session={report.SessionId}, tenant={report.TenantId}).",
+                });
+            }
+
            // Final-stage comparison against the last stored transition's ToStage.
            var lastStoredToStage = orderedTransitions[orderedTransitions.Count - 1].ToStage;
            var replayedStageName = state.Stage.ToString();
@@ -54,6 +54,31 @@ namespace AutopilotMonitor.DecisionCore.Engine
            IDecisionEngine engine,
            DecisionState seed,
            IEnumerable<DecisionSignal> signals)
+            => Replay(engine, seed, signals, onTransition: null);
+
+        /// <summary>
+        /// Fold variant that exposes each intermediate <see cref="DecisionTransition"/> to
+        /// the caller via <paramref name="onTransition"/>. Used by the orchestrator's
+        /// recovery path (Codex follow-up post-#50 #C) to rematerialise the Journal when a
+        /// crash left the SignalLog ahead of the Journal: the snapshot is reached by the
+        /// pure fold, then the same fold re-emits every tail transition so the Journal
+        /// can backfill missing <see cref="DecisionTransition.StepIndex"/> entries.
+        /// Determinism-contract is unchanged; the callback is side-effect-free from
+        /// <see cref="ReducerReplay"/>'s perspective — any exception it throws propagates
+        /// out and aborts replay.
+        /// </summary>
+        /// <param name="engine">Reducer kernel. Stateless — may be shared across calls.</param>
+        /// <param name="seed">Starting state (see other overload).</param>
+        /// <param name="signals">Ordered signal sequence (see other overload).</param>
+        /// <param name="onTransition">Optional per-step callback; receives the transition
+        /// produced by the current signal BEFORE <c>state</c> is advanced to the next
+        /// signal. Null = no callback (behaves identically to the 3-arg overload).</param>
+        /// <returns>The reduced state after applying every signal.</returns>
+        public static DecisionState Replay(
+            IDecisionEngine engine,
+            DecisionState seed,
+            IEnumerable<DecisionSignal> signals,
+            Action<DecisionTransition>? onTransition)
        {
            if (engine == null) throw new ArgumentNullException(nameof(engine));
            if (seed == null) throw new ArgumentNullException(nameof(seed));
@@ -80,6 +105,7 @@ namespace AutopilotMonitor.DecisionCore.Engine
                }

                var step = engine.Reduce(state, signal);
+                onTransition?.Invoke(step.Transition);
                state = step.NewState;
                lastOrdinal = signal.SessionSignalOrdinal;
            }