From bd3ec4e195393b37d4ceec7c3fb58009b5a4341e Mon Sep 17 00:00:00 2001 From: Oliver Kieselbach Date: Fri, 24 Apr 2026 13:39:47 +0200 Subject: [PATCH] v2 codex-post-#50: recovery + abort durability follow-ups (#A #B #C #D) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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? 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) --- .../DecisionStepProcessorTests.cs | 16 +- .../Orchestration/EffectRunnerTests.cs | 47 ++---- .../FakeDecisionStepProcessor.cs | 17 +- .../Orchestration/RecoveryPathTests.cs | 147 ++++++++++++++++-- .../Orchestration/SignalIngressTests.cs | 82 +++++++++- .../Orchestration/DecisionStepProcessor.cs | 56 ++++--- .../Orchestration/EffectRunner.cs | 47 +----- .../Orchestration/EnrollmentOrchestrator.cs | 114 +++++++++----- .../Orchestration/IDecisionStepProcessor.cs | 11 +- .../Orchestration/SignalIngress.cs | 93 ++++++++++- .../buildcounter.txt | 2 +- .../AutopilotMonitor.Agent/buildcounter.txt | 2 +- .../ReducerReplayTests.cs | 76 +++++++++ .../ReducerVerifierSemanticReplayTests.cs | 25 ++- .../Services/ReducerVerifier.cs | 20 ++- .../Engine/ReducerReplay.cs | 26 ++++ 16 files changed, 608 insertions(+), 173 deletions(-) diff --git a/src/Agent/AutopilotMonitor.Agent.V2.Core.Tests/Orchestration/DecisionStepProcessorTests.cs b/src/Agent/AutopilotMonitor.Agent.V2.Core.Tests/Orchestration/DecisionStepProcessorTests.cs index 058e7c96..73d8c3b9 100644 --- a/src/Agent/AutopilotMonitor.Agent.V2.Core.Tests/Orchestration/DecisionStepProcessorTests.cs +++ b/src/Agent/AutopilotMonitor.Agent.V2.Core.Tests/Orchestration/DecisionStepProcessorTests.cs @@ -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] diff --git a/src/Agent/AutopilotMonitor.Agent.V2.Core.Tests/Orchestration/EffectRunnerTests.cs b/src/Agent/AutopilotMonitor.Agent.V2.Core.Tests/Orchestration/EffectRunnerTests.cs index ea8f8c0f..52530974 100644 --- a/src/Agent/AutopilotMonitor.Agent.V2.Core.Tests/Orchestration/EffectRunnerTests.cs +++ b/src/Agent/AutopilotMonitor.Agent.V2.Core.Tests/Orchestration/EffectRunnerTests.cs @@ -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] diff --git a/src/Agent/AutopilotMonitor.Agent.V2.Core.Tests/Orchestration/FakeDecisionStepProcessor.cs b/src/Agent/AutopilotMonitor.Agent.V2.Core.Tests/Orchestration/FakeDecisionStepProcessor.cs index 3714328b..071f5f4a 100644 --- a/src/Agent/AutopilotMonitor.Agent.V2.Core.Tests/Orchestration/FakeDecisionStepProcessor.cs +++ b/src/Agent/AutopilotMonitor.Agent.V2.Core.Tests/Orchestration/FakeDecisionStepProcessor.cs @@ -17,6 +17,7 @@ namespace AutopilotMonitor.Agent.V2.Core.Tests.Orchestration { private readonly object _lock = new object(); private readonly Queue _script = new Queue(); + private readonly Queue _resultScript = new Queue(); private readonly List _calls = new List(); 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) + /// + /// Script a specific for the next ApplyStep call(s) — + /// lets tests drive the SignalIngress inline-abort path (Codex post-#50 #B). + /// + 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(); } } diff --git a/src/Agent/AutopilotMonitor.Agent.V2.Core.Tests/Orchestration/RecoveryPathTests.cs b/src/Agent/AutopilotMonitor.Agent.V2.Core.Tests/Orchestration/RecoveryPathTests.cs index c104159a..d620a52b 100644 --- a/src/Agent/AutopilotMonitor.Agent.V2.Core.Tests/Orchestration/RecoveryPathTests.cs +++ b/src/Agent/AutopilotMonitor.Agent.V2.Core.Tests/Orchestration/RecoveryPathTests.cs @@ -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(); } diff --git a/src/Agent/AutopilotMonitor.Agent.V2.Core.Tests/Orchestration/SignalIngressTests.cs b/src/Agent/AutopilotMonitor.Agent.V2.Core.Tests/Orchestration/SignalIngressTests.cs index 9ff71d08..0a74a5bc 100644 --- a/src/Agent/AutopilotMonitor.Agent.V2.Core.Tests/Orchestration/SignalIngressTests.cs +++ b/src/Agent/AutopilotMonitor.Agent.V2.Core.Tests/Orchestration/SignalIngressTests.cs @@ -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(), + 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(), + 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] diff --git a/src/Agent/AutopilotMonitor.Agent.V2.Core/Orchestration/DecisionStepProcessor.cs b/src/Agent/AutopilotMonitor.Agent.V2.Core/Orchestration/DecisionStepProcessor.cs index 0dfb1c3a..6c5e1c6c 100644 --- a/src/Agent/AutopilotMonitor.Agent.V2.Core/Orchestration/DecisionStepProcessor.cs +++ b/src/Agent/AutopilotMonitor.Agent.V2.Core/Orchestration/DecisionStepProcessor.cs @@ -98,7 +98,7 @@ namespace AutopilotMonitor.Agent.V2.Core.Orchestration /// Test-Observability — anzahl aufeinanderfolgender Journal-Failures. 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) diff --git a/src/Agent/AutopilotMonitor.Agent.V2.Core/Orchestration/EffectRunner.cs b/src/Agent/AutopilotMonitor.Agent.V2.Core/Orchestration/EffectRunner.cs index eb9eb37f..7664e6eb 100644 --- a/src/Agent/AutopilotMonitor.Agent.V2.Core/Orchestration/EffectRunner.cs +++ b/src/Agent/AutopilotMonitor.Agent.V2.Core/Orchestration/EffectRunner.cs @@ -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); } - /// - /// Codex follow-up #2 — synthesise an - /// signal when a critical deadline effect cannot be delivered. The reducer's - /// HandleEffectInfrastructureFailureV1 picks this up next tick and transitions the - /// session to / . - /// Any exception from the ingress (full channel / disposed) is swallowed: the - /// max-lifetime watchdog remains the last-resort termination path. - /// - 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(StringComparer.Ordinal) - { - ["reason"] = abortReason, - ["failingEffect"] = failingEffect.ToString(), - }); - } - catch - { - // Intentionally swallowed — see method doc. - } - } - private void PostClassifierVerdictSignal(ClassifierVerdict verdict, string classifierId, DateTime stepOccurredAtUtc) { _ingress.Post( diff --git a/src/Agent/AutopilotMonitor.Agent.V2.Core/Orchestration/EnrollmentOrchestrator.cs b/src/Agent/AutopilotMonitor.Agent.V2.Core/Orchestration/EnrollmentOrchestrator.cs index eedd0633..65d77f83 100644 --- a/src/Agent/AutopilotMonitor.Agent.V2.Core/Orchestration/EnrollmentOrchestrator.cs +++ b/src/Agent/AutopilotMonitor.Agent.V2.Core/Orchestration/EnrollmentOrchestrator.cs @@ -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 signalsToReplay = Array.Empty(); + 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 diff --git a/src/Agent/AutopilotMonitor.Agent.V2.Core/Orchestration/IDecisionStepProcessor.cs b/src/Agent/AutopilotMonitor.Agent.V2.Core/Orchestration/IDecisionStepProcessor.cs index f1e34c35..d3a4bbd4 100644 --- a/src/Agent/AutopilotMonitor.Agent.V2.Core/Orchestration/IDecisionStepProcessor.cs +++ b/src/Agent/AutopilotMonitor.Agent.V2.Core/Orchestration/IDecisionStepProcessor.cs @@ -34,10 +34,17 @@ namespace AutopilotMonitor.Agent.V2.Core.Orchestration /// /// Transition ins Journal appenden (Sofort-Flush) /// Effekte mittels ausführen + /// Snapshot der speichern — ABER nur wenn + /// 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). /// auf fortsetzen /// - /// Darf werfen — Ingress fängt ab und loggt, verstummt aber nicht. + /// Gibt das zurück, damit der Caller (typischerweise + /// ) entscheiden kann, ob ein synthetisches + /// inline nachgeschoben + /// werden muss. Darf werfen — Ingress fängt ab und loggt, verstummt aber nicht. /// - void ApplyStep(DecisionStep step, DecisionSignal signal); + EffectRunResult ApplyStep(DecisionStep step, DecisionSignal signal); } } diff --git a/src/Agent/AutopilotMonitor.Agent.V2.Core/Orchestration/SignalIngress.cs b/src/Agent/AutopilotMonitor.Agent.V2.Core/Orchestration/SignalIngress.cs index 39c2dfe9..113a73fc 100644 --- a/src/Agent/AutopilotMonitor.Agent.V2.Core/Orchestration/SignalIngress.cs +++ b/src/Agent/AutopilotMonitor.Agent.V2.Core/Orchestration/SignalIngress.cs @@ -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); + } + } + + /// + /// Synthesise + durably persist + reduce a EffectInfrastructureFailure 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. + /// + 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 ?? ""}"), + payload: new Dictionary(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 diff --git a/src/Agent/AutopilotMonitor.Agent.V2/buildcounter.txt b/src/Agent/AutopilotMonitor.Agent.V2/buildcounter.txt index a700e799..4438e305 100644 --- a/src/Agent/AutopilotMonitor.Agent.V2/buildcounter.txt +++ b/src/Agent/AutopilotMonitor.Agent.V2/buildcounter.txt @@ -1 +1 @@ -257 +293 diff --git a/src/Agent/AutopilotMonitor.Agent/buildcounter.txt b/src/Agent/AutopilotMonitor.Agent/buildcounter.txt index b1e1a0c0..a788b7b3 100644 --- a/src/Agent/AutopilotMonitor.Agent/buildcounter.txt +++ b/src/Agent/AutopilotMonitor.Agent/buildcounter.txt @@ -1 +1 @@ -1120 +1134 diff --git a/src/Agent/AutopilotMonitor.DecisionCore.Tests/ReducerReplayTests.cs b/src/Agent/AutopilotMonitor.DecisionCore.Tests/ReducerReplayTests.cs index 4b334e37..ec32c2d3 100644 --- a/src/Agent/AutopilotMonitor.DecisionCore.Tests/ReducerReplayTests.cs +++ b/src/Agent/AutopilotMonitor.DecisionCore.Tests/ReducerReplayTests.cs @@ -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(); + 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(() => + 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() { diff --git a/src/Backend/AutopilotMonitor.Functions.Tests/ReducerVerifierSemanticReplayTests.cs b/src/Backend/AutopilotMonitor.Functions.Tests/ReducerVerifierSemanticReplayTests.cs index 74289002..a6e5d237 100644 --- a/src/Backend/AutopilotMonitor.Functions.Tests/ReducerVerifierSemanticReplayTests.cs +++ b/src/Backend/AutopilotMonitor.Functions.Tests/ReducerVerifierSemanticReplayTests.cs @@ -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() { diff --git a/src/Backend/AutopilotMonitor.Functions/Services/ReducerVerifier.cs b/src/Backend/AutopilotMonitor.Functions/Services/ReducerVerifier.cs index 547a2cd1..c8aee1d9 100644 --- a/src/Backend/AutopilotMonitor.Functions/Services/ReducerVerifier.cs +++ b/src/Backend/AutopilotMonitor.Functions/Services/ReducerVerifier.cs @@ -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(); diff --git a/src/Shared/AutopilotMonitor.DecisionCore/Engine/ReducerReplay.cs b/src/Shared/AutopilotMonitor.DecisionCore/Engine/ReducerReplay.cs index 95390534..a18b7b46 100644 --- a/src/Shared/AutopilotMonitor.DecisionCore/Engine/ReducerReplay.cs +++ b/src/Shared/AutopilotMonitor.DecisionCore/Engine/ReducerReplay.cs @@ -54,6 +54,31 @@ namespace AutopilotMonitor.DecisionCore.Engine IDecisionEngine engine, DecisionState seed, IEnumerable signals) + => Replay(engine, seed, signals, onTransition: null); + + /// + /// Fold variant that exposes each intermediate to + /// the caller via . 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 entries. + /// Determinism-contract is unchanged; the callback is side-effect-free from + /// 's perspective — any exception it throws propagates + /// out and aborts replay. + /// + /// Reducer kernel. Stateless — may be shared across calls. + /// Starting state (see other overload). + /// Ordered signal sequence (see other overload). + /// Optional per-step callback; receives the transition + /// produced by the current signal BEFORE state is advanced to the next + /// signal. Null = no callback (behaves identically to the 3-arg overload). + /// The reduced state after applying every signal. + public static DecisionState Replay( + IDecisionEngine engine, + DecisionState seed, + IEnumerable signals, + Action? 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; }