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>
This commit is contained in:
Oliver Kieselbach
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 // ========================================================================= EffectRunner behavior
[Fact] [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(); using var rig = new Rig();
var abortResult = new EffectRunResult( var abortResult = new EffectRunResult(
sessionMustAbort: true, sessionMustAbort: true,
@@ -210,11 +215,14 @@ namespace AutopilotMonitor.Agent.V2.Core.Tests.Orchestration
var sut = rig.Build(); var sut = rig.Build();
var (step, signal) = rig.ReduceSessionStarted(); var (step, signal) = rig.ReduceSessionStarted();
sut.ApplyStep(step, signal); // state still advances var result = sut.ApplyStep(step, signal);
Assert.Single(rig.Journal.Appended); Assert.Single(rig.Journal.Appended);
Assert.Same(step.NewState, sut.CurrentState); Assert.Same(step.NewState, sut.CurrentState); // state forwards so the synthetic signal reduces from here
Assert.Empty(rig.Quarantine.Reasons); // effect-failures do NOT quarantine 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] [Fact]
@@ -91,11 +91,12 @@ namespace AutopilotMonitor.Agent.V2.Core.Tests.Orchestration
} }
[Fact] [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 // Codex follow-up (post-#50 #B): responsibility for synthesising the durable
// world, not just the log. EffectRunner posts a synthetic signal that the // EffectInfrastructureFailure signal moved from EffectRunner to SignalIngress
// reducer picks up on the next step. // (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(); var rig = new Rig();
rig.Scheduler.ThrowOnSchedule = new InvalidOperationException("timer-broken"); rig.Scheduler.ThrowOnSchedule = new InvalidOperationException("timer-broken");
var sut = rig.Build(); var sut = rig.Build();
@@ -106,18 +107,12 @@ namespace AutopilotMonitor.Agent.V2.Core.Tests.Orchestration
var result = await sut.RunAsync(new[] { effect }, InitialState(), At); var result = await sut.RunAsync(new[] { effect }, InitialState(), At);
Assert.True(result.SessionMustAbort); Assert.True(result.SessionMustAbort);
Assert.Single(rig.Ingress.Posted); Assert.Contains("timer_infrastructure_failure", result.AbortReason);
var posted = rig.Ingress.Posted[0]; Assert.Empty(rig.Ingress.Posted); // no longer the EffectRunner's job
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"]);
} }
[Fact] [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(); var rig = new Rig();
rig.Scheduler.ThrowOnCancel = new InvalidOperationException("timer-broken"); 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); var result = await sut.RunAsync(new[] { effect }, InitialState(), At);
Assert.True(result.SessionMustAbort); Assert.True(result.SessionMustAbort);
Assert.Single(rig.Ingress.Posted); Assert.Empty(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.
} }
[Fact] [Fact]
@@ -17,6 +17,7 @@ namespace AutopilotMonitor.Agent.V2.Core.Tests.Orchestration
{ {
private readonly object _lock = new object(); private readonly object _lock = new object();
private readonly Queue<Exception?> _script = new Queue<Exception?>(); 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 readonly List<ApplyCall> _calls = new List<ApplyCall>();
private DecisionState _currentState; private DecisionState _currentState;
private int _failedApplyCount; private int _failedApplyCount;
@@ -62,7 +63,20 @@ namespace AutopilotMonitor.Agent.V2.Core.Tests.Orchestration
return this; 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(); BlockHandle?.Wait();
@@ -79,6 +93,7 @@ namespace AutopilotMonitor.Agent.V2.Core.Tests.Orchestration
} }
_calls.Add(new ApplyCall(step, signal)); _calls.Add(new ApplyCall(step, signal));
_currentState = step.NewState; _currentState = step.NewState;
return _resultScript.Count > 0 ? _resultScript.Dequeue() : EffectRunResult.Empty();
} }
} }
@@ -262,6 +262,50 @@ namespace AutopilotMonitor.Agent.V2.Core.Tests.Orchestration
sut.Stop(); 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] [Fact]
public void Corrupt_snapshot_with_intact_log_replays_full_log_without_quarantining_log() 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] [Fact]
public void Journal_ahead_of_replayed_state_triggers_phantom_truncate() public void Journal_ahead_of_replayed_state_triggers_phantom_truncate()
{ {
// Pathological crash scenario: the Journal flushed step N but the Snapshot saved // AHEAD crash scenario: the Journal flushed step for sig1, then also a phantom
// only step N-1 and a later SignalLog append was never actually persisted (log // step for a signal the SignalLog never persisted, while the Snapshot saved
// tail physically shorter than journal). On recovery the journal suffix is a // only step for sig0. On recovery everything beyond seed.StepIndex goes to the
// phantom — TruncateAfter moves it to .quarantine and realigns so the next live // phantom bucket and the replay callback (Codex follow-up post-#50 #C) rebuilds
// append does not violate monotonicity. // the authoritative tail bit-for-bit in lockstep with state advancement.
using var rig = new Rig(); using var rig = new Rig();
Directory.CreateDirectory(rig.StateDir); Directory.CreateDirectory(rig.StateDir);
// Step 1: SignalLog has ordinals 0 + 1 only. // Step 1: SignalLog has sig0 + sig1.
var sigs = new[] var sigs = new[]
{ {
MakeSignal(0, DecisionSignalKind.SessionStarted), 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")); var seedLog = new SignalLogWriter(Path.Combine(rig.StateDir, "signal-log.jsonl"));
foreach (var s in sigs) seedLog.Append(s); 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 engine = new DecisionEngine();
var snapshotState = engine.Reduce( var step0 = engine.Reduce(DecisionState.CreateInitial("S1", "T1"), sigs[0]);
DecisionState.CreateInitial("S1", "T1"), sigs[0]).NewState; var step1 = engine.Reduce(step0.NewState, sigs[1]);
var snapshotState = step0.NewState; // StepIndex=1
new SnapshotPersistence( new SnapshotPersistence(
Path.Combine(rig.StateDir, "snapshot.json"), Path.Combine(rig.StateDir, "snapshot.json"),
() => rig.Clock.UtcNow).Save(snapshotState); () => rig.Clock.UtcNow).Save(snapshotState);
// Step 3: Journal has 3 transitions on disk (StepIndex 0, 1, 2) — the StepIndex-2 // Step 3: Journal has 3 transitions on disk — the StepIndex=3 one is phantom
// entry is the phantom (its signal was never flushed to the log). // (no backing signal in the log).
var journalPath = Path.Combine(rig.StateDir, "journal.jsonl"); var journalPath = Path.Combine(rig.StateDir, "journal.jsonl");
var seedJournal = new JournalWriter(journalPath, () => rig.Clock.UtcNow); var seedJournal = new JournalWriter(journalPath, () => rig.Clock.UtcNow);
seedJournal.Append(MakeTransition(0, signalRef: 0)); seedJournal.Append(step0.Transition); // StepIndex=1
seedJournal.Append(MakeTransition(1, signalRef: 1)); seedJournal.Append(step1.Transition); // StepIndex=2 — pre-crash flushed
seedJournal.Append(MakeTransition(2, signalRef: 99)); // phantom seedJournal.Append(MakeTransition(3, signalRef: 99)); // StepIndex=3 — phantom
// Expected post-recovery state: replay tail [sig1] → StepIndex=2, LastApplied=1. // Expected post-recovery state: replay tail [sig1] → StepIndex=2, LastApplied=1.
var expected = ReducerReplay.Replay(engine, snapshotState, new[] { sigs[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.StepIndex, sut.CurrentState.StepIndex);
Assert.Equal(expected.LastAppliedSignalOrdinal, sut.CurrentState.LastAppliedSignalOrdinal); 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"); var quarantineRoot = Path.Combine(rig.StateDir, ".quarantine");
Assert.True(Directory.Exists(quarantineRoot)); Assert.True(Directory.Exists(quarantineRoot));
var phantomFiles = Directory.GetFiles( var phantomFiles = Directory.GetFiles(
quarantineRoot, "journal-phantom-tail.jsonl", SearchOption.AllDirectories); quarantineRoot, "journal-phantom-tail.jsonl", SearchOption.AllDirectories);
Assert.Single(phantomFiles); 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(); sut.Stop();
} }
@@ -233,10 +233,11 @@ namespace AutopilotMonitor.Agent.V2.Core.Tests.Orchestration
public DecisionCore.State.DecisionState CurrentState => _state; 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); _inspect(step, signal, _log.LastOrdinal);
_state = step.NewState; _state = step.NewState;
return EffectRunResult.Empty();
} }
} }
@@ -283,6 +284,85 @@ namespace AutopilotMonitor.Agent.V2.Core.Tests.Orchestration
Assert.Equal(2, rig.SignalLog.ReadAll().Count); 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 // ============================================================ Codex Finding 4 — SignalPosted
[Fact] [Fact]
@@ -98,7 +98,7 @@ namespace AutopilotMonitor.Agent.V2.Core.Orchestration
/// <summary>Test-Observability — anzahl aufeinanderfolgender Journal-Failures.</summary> /// <summary>Test-Observability — anzahl aufeinanderfolgender Journal-Failures.</summary>
public int ConsecutiveJournalFailureCount => _consecutiveJournalFailures; 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 (step == null) throw new ArgumentNullException(nameof(step));
if (signal == null) throw new ArgumentNullException(nameof(signal)); if (signal == null) throw new ArgumentNullException(nameof(signal));
@@ -160,16 +160,22 @@ namespace AutopilotMonitor.Agent.V2.Core.Orchestration
if (effectResult.SessionMustAbort) if (effectResult.SessionMustAbort)
{ {
// Codex follow-up #2: EffectRunner has already posted a synthetic // Codex follow-up (post-#50 #B): the phantom state must NOT be snapshotted
// EffectInfrastructureFailure signal to the ingress. The reducer will // because its ActiveDeadline was never actually armed on the live scheduler.
// transition the session to Failed / EnrollmentFailed on the next // Recovery from a phantom snapshot would try to re-arm and re-fail, leaving
// dispatched step — our job here is purely observability. // 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( _logger.Error(
$"DecisionStepProcessor: effect run signaled session abort " + $"DecisionStepProcessor: effect run signaled session abort " +
$"(reason='{effectResult.AbortReason}') for signal ordinal={signal.SessionSignalOrdinal}; " + $"(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
{
if (effectResult.Failures.Count > 0)
{ {
_logger.Warning( _logger.Warning(
$"DecisionStepProcessor: effect run completed with {effectResult.Failures.Count} non-fatal failure(s) " + $"DecisionStepProcessor: effect run completed with {effectResult.Failures.Count} non-fatal failure(s) " +
@@ -187,8 +193,10 @@ namespace AutopilotMonitor.Agent.V2.Core.Orchestration
$"DecisionStepProcessor: snapshot save failed (transient, not escalated) " + $"DecisionStepProcessor: snapshot save failed (transient, not escalated) " +
$"for signal ordinal={signal.SessionSignalOrdinal}: {ex.Message}"); $"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; _currentState = step.NewState;
_consecutiveJournalFailures = 0; _consecutiveJournalFailures = 0;
@@ -208,6 +216,8 @@ namespace AutopilotMonitor.Agent.V2.Core.Orchestration
ex); ex);
} }
} }
return effectResult;
} }
private void TryTriggerQuarantine(string reason) private void TryTriggerQuarantine(string reason)
@@ -81,13 +81,11 @@ namespace AutopilotMonitor.Agent.V2.Core.Orchestration
failures.Add(criticalFail); failures.Add(criticalFail);
var abortReason = $"timer_infrastructure_failure: {criticalFail.ErrorReason}"; var abortReason = $"timer_infrastructure_failure: {criticalFail.ErrorReason}";
// Codex follow-up #2: post a synthetic EffectInfrastructureFailure // Codex follow-up (post-#50 #B): responsibility for posting the
// signal so the reducer can transition cleanly to Failed / // synthetic EffectInfrastructureFailure signal moved to SignalIngress
// EnrollmentFailed on the next step. Without this the flag would // so it can DURABLY append + reduce inline within the same worker
// only surface in logs and the session could hang on a phantom // iteration, closing the crash-window between in-memory post and
// deadline that was never actually armed. // SignalLog persistence. EffectRunner just signals the abort here.
PostEffectInfrastructureFailureSignal(effect.Kind, abortReason, stepOccurredAtUtc);
return new EffectRunResult( return new EffectRunResult(
sessionMustAbort: true, sessionMustAbort: true,
abortReason: abortReason, abortReason: abortReason,
@@ -229,41 +227,6 @@ namespace AutopilotMonitor.Agent.V2.Core.Orchestration
return new ClassifierEffectOutcome(invoked: true, skippedByAntiLoop: false, failure: null); 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) private void PostClassifierVerdictSignal(ClassifierVerdict verdict, string classifierId, DateTime stepOccurredAtUtc)
{ {
_ingress.Post( _ingress.Post(
@@ -279,13 +279,23 @@ namespace AutopilotMonitor.Agent.V2.Core.Orchestration
// loss quarantine of the log segments and start fresh. // loss quarantine of the log segments and start fresh.
// b) Snapshot loaded cleanly → replay any SignalLog tail past the snapshot's // b) Snapshot loaded cleanly → replay any SignalLog tail past the snapshot's
// LastAppliedSignalOrdinal so a pre-crash Journal/Snapshot lag is closed. // 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 // After the seed + replay-stream are selected the Journal is aligned AHEAD-OF
// replayed state — if a crash left phantom transitions on disk (Journal flushed, // replay (phantom suffix beyond seed.StepIndex-1 goes to the forensic bucket)
// Snapshot not) the next live Append would throw a monotonicity violation and // and backfilled DURING replay via the onTransition callback. This closes two
// eventually quarantine-escalate. TruncateAfter moves the phantoms to a // crash windows in one mechanism (Codex follow-up post-#50 #C):
// forensic bucket and realigns. // • 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 snapshotFileExistsPreLoad = File.Exists(snapshotPath);
var loadedState = _snapshot.Load(); var loadedState = _snapshot.Load();
DecisionState initialState; 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. // The live ingress pipeline creates its own instance further down the Start() body.
var replayEngine = new DecisionEngine(); 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) if (loadedState == null && snapshotFileExistsPreLoad)
{ {
// Branch (a) — Snapshot corrupt. Quarantine the snapshot and attempt log replay. // 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); _journal = new JournalWriter(journalPath, () => _clock.UtcNow);
_eventSequencePersistence = new EventSequencePersistence(eventSequencePath); _eventSequencePersistence = new EventSequencePersistence(eventSequencePath);
initialState = DecisionState.CreateInitial(_sessionId, _tenantId); branchTag = "a-total-loss";
} }
else else
{ {
_logger.Info( signalsToReplay = loggedSignals;
$"EnrollmentOrchestrator: rebuilding state via ReducerReplay over {loggedSignals.Count} SignalLog entries."); branchTag = "a-full-log-replay";
initialState = ReducerReplay.Replay(
engine: replayEngine,
seed: DecisionState.CreateInitial(_sessionId, _tenantId),
signals: loggedSignals);
} }
_wasStartupQuarantine = true; _wasStartupQuarantine = true;
@@ -336,43 +348,63 @@ namespace AutopilotMonitor.Agent.V2.Core.Orchestration
else if (loadedState != null) else if (loadedState != null)
{ {
// Branch (b) — Snapshot loaded. Catch up any SignalLog tail past the snapshot. // Branch (b) — Snapshot loaded. Catch up any SignalLog tail past the snapshot.
var tail = CollectSignalLogTailAfter(loadedState.LastAppliedSignalOrdinal); seed = loadedState;
signalsToReplay = CollectSignalLogTailAfter(loadedState.LastAppliedSignalOrdinal);
if (tail.Count > 0) branchTag = signalsToReplay.Count > 0 ? "b-tail-replay" : "b-snapshot-current";
{
_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 else
{ {
initialState = loadedState; // Branch (c) — no snapshot on disk. Two sub-cases:
_logger.Info( // (c1) SignalLog also empty → genuinely fresh start.
$"EnrollmentOrchestrator: recovered state from snapshot (stage={loadedState.Stage}, stepIndex={loadedState.StepIndex})."); // (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 else
{ {
// Branch (c) — no prior state. signalsToReplay = loggedSignals;
initialState = DecisionState.CreateInitial(_sessionId, _tenantId); _wasStartupQuarantine = true;
_logger.Info($"EnrollmentOrchestrator: fresh initial state for session {_sessionId}."); branchTag = "c2-full-log-replay";
}
} }
// Post-recovery alignment (plan §5.1): truncate Journal phantom suffix if the file // Align Journal AHEAD phantoms before replay. Engine semantics: a transition
// is ahead of the replayed state. After a normal Reduce the Journal holds entries // produced for state at StepIndex=K carries StepIndex=K+1 (see
// with StepIndex = [0 .. state.StepIndex - 1]; anything beyond is a leftover from a // DecisionEngine.BuildTakenTransition); after K reduces from initial the journal
// pre-crash Journal-flushed-but-snapshot-not path. // holds entries [1..K] with LastStepIndex=K. So any on-disk entry beyond
if (_journal.LastStepIndex >= initialState.StepIndex) // 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( _logger.Warning(
$"EnrollmentOrchestrator: Journal tail ahead of replayed state " + $"EnrollmentOrchestrator: Journal ahead of seed state " +
$"(journal.LastStepIndex={_journal.LastStepIndex}, state.StepIndex={initialState.StepIndex}) — " + $"(journal.LastStepIndex={_journal.LastStepIndex}, seed.StepIndex={seed.StepIndex}) — " +
$"truncating phantom transitions to boundary={boundary}."); $"truncating phantom transitions to boundary={journalBoundary}.");
_journal.TruncateAfter(boundary); _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. // Sonderfall 1 Detection: WG Part 1 -> Reboot -> Part 2 Resume.
// Self-destruct bleibt AUS (Caller ist zuständig — die Info wird via // Self-destruct bleibt AUS (Caller ist zuständig — die Info wird via
// IsWhiteGlovePart2Resume exponiert); Session-Recovered-Signal wird gepostet // IsWhiteGlovePart2Resume exponiert); Session-Recovered-Signal wird gepostet
@@ -34,10 +34,17 @@ namespace AutopilotMonitor.Agent.V2.Core.Orchestration
/// <list type="number"> /// <list type="number">
/// <item>Transition ins Journal appenden (Sofort-Flush)</item> /// <item>Transition ins Journal appenden (Sofort-Flush)</item>
/// <item>Effekte mittels <see cref="IEffectRunner"/> ausführen</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> /// <item><see cref="CurrentState"/> auf <see cref="DecisionStep.NewState"/> fortsetzen</item>
/// </list> /// </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> /// </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. // IDecisionEngine.Reduce ist fail-safe (DecisionEngine.cs:39-47): wirft nicht.
var step = _engine.Reduce(state, signal); var step = _engine.Reduce(state, signal);
EffectRunResult? effectResult = null;
try try
{ {
_processor.ApplyStep(step, signal); effectResult = _processor.ApplyStep(step, signal);
} }
catch catch
{ {
@@ -351,6 +352,96 @@ namespace AutopilotMonitor.Agent.V2.Core.Orchestration
// M4.4.5 vom Orchestrator geloggt + ggf. Quarantine ausgelöst. M4.4.0 // M4.4.5 vom Orchestrator geloggt + ggf. Quarantine ausgelöst. M4.4.0
// verschluckt bewusst, damit der Ingress-Worker nicht verstummt. // 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 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); 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] [Fact]
public void Replay_NullArgs_ThrowArgumentNullException() public void Replay_NullArgs_ThrowArgumentNullException()
{ {
@@ -40,7 +40,7 @@ public class ReducerVerifierSemanticReplayTests
}; };
var engine = new DecisionEngine(); var engine = new DecisionEngine();
var state = DecisionState.CreateInitial(TenantId, SessionId); var state = DecisionState.CreateInitial(SessionId, TenantId);
var transitionRecords = new DecisionTransitionRecord[sigs.Length]; var transitionRecords = new DecisionTransitionRecord[sigs.Length];
for (var i = 0; i < sigs.Length; i++) 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"); 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] [Fact]
public void Replay_handles_mid_stream_signal_deserialization_failure_without_aborting() public void Replay_handles_mid_stream_signal_deserialization_failure_without_aborting()
{ {
@@ -317,7 +317,10 @@ namespace AutopilotMonitor.Functions.Services
} }
var engine = new DecisionEngine(); 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 divergences = 0;
var stepsToCompare = Math.Min(decodedSignals.Count, decodedTransitions.Count); 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. // Final-stage comparison against the last stored transition's ToStage.
var lastStoredToStage = orderedTransitions[orderedTransitions.Count - 1].ToStage; var lastStoredToStage = orderedTransitions[orderedTransitions.Count - 1].ToStage;
var replayedStageName = state.Stage.ToString(); var replayedStageName = state.Stage.ToString();
@@ -54,6 +54,31 @@ namespace AutopilotMonitor.DecisionCore.Engine
IDecisionEngine engine, IDecisionEngine engine,
DecisionState seed, DecisionState seed,
IEnumerable<DecisionSignal> signals) 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 (engine == null) throw new ArgumentNullException(nameof(engine));
if (seed == null) throw new ArgumentNullException(nameof(seed)); if (seed == null) throw new ArgumentNullException(nameof(seed));
@@ -80,6 +105,7 @@ namespace AutopilotMonitor.DecisionCore.Engine
} }
var step = engine.Reduce(state, signal); var step = engine.Reduce(state, signal);
onTransition?.Invoke(step.Transition);
state = step.NewState; state = step.NewState;
lastOrdinal = signal.SessionSignalOrdinal; lastOrdinal = signal.SessionSignalOrdinal;
} }