v2 codex-post-#50-2: rehydrate-failure durable abort + v1 payload contract (#E #F)

Addresses Codex second-pass findings against commit bd3ec4e1: #E (High) Deadline re-arm failure path was still catch+log-only. If _scheduler.RehydrateFromSnapshot throws, DecisionState still claims the deadline is active but the live scheduler never registered it — the same phantom-deadline hang class already closed for live ScheduleDeadline / CancelDeadline failures. The orchestrator now posts an EffectInfrastructureFailure signal with the v1 payload contract through the already-running ingress, so the reducer terminates the session cleanly instead of deferring to the max-lifetime watchdog. An IDeadlineScheduler? schedulerOverride constructor parameter was added to EnrollmentOrchestrator so tests can drive the failure path with a FakeDeadlineScheduler { ThrowOnSchedule = ... }. #F (Medium) The inline EffectInfrastructureFailure signal synthesised by SignalIngress.ProcessInlineAbortSignal only carried `reason` in the payload and used a placeholder sourceOrigin. That broke the v1 contract documented on DecisionSignalKind.EffectInfrastructureFailure + the fixture (effect-infrastructure-failure-v1.json): { reason, failingEffect } with sourceOrigin "effectrunner:critical:<EffectKind>" so forensic consumers can distinguish ScheduleDeadline-vs-CancelDeadline failures. ProcessItem now extracts the critical failure's EffectKind from EffectRunResult.Failures (last non-transient entry) and threads it into the synthetic signal's sourceOrigin, Evidence.Identifier and payload. Suite-Baseline all green, 0 warnings: DecisionCore 164/164, Agent.V2.Core 650/650, Functions 735/735. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
2026-10-24 12:42:38 +00:00 · 2026-04-24 14:00:36 +02:00
parent bd3ec4e195
commit 1418eacfe9
7 changed files with 194 additions and 22 deletions
@@ -335,6 +335,9 @@ namespace AutopilotMonitor.Agent.V2.Core.Tests.Orchestration
            var field = typeof(EnrollmentOrchestrator).GetField(
                "_scheduler",
                System.Reflection.BindingFlags.Instance | System.Reflection.BindingFlags.NonPublic);
+            // Field is now typed IDeadlineScheduler (post-#50 #E testability hook); this
+            // helper still only runs in the tests that drive the real scheduler (past-due
+            // fire etc.), so the cast back to the concrete type is safe.
            return (DeadlineScheduler)field!.GetValue(sut)!;
        }
    }
@@ -40,7 +40,7 @@ namespace AutopilotMonitor.Agent.V2.Core.Tests.Orchestration
                TransportDir = Path.Combine(Tmp.Path, "Transport");
            }

-            public EnrollmentOrchestrator Build() =>
+            public EnrollmentOrchestrator Build(IDeadlineScheduler? schedulerOverride = null) =>
                new EnrollmentOrchestrator(
                    sessionId: "S1",
                    tenantId: "T1",
@@ -51,7 +51,8 @@ namespace AutopilotMonitor.Agent.V2.Core.Tests.Orchestration
                    uploader: Uploader,
                    classifiers: new List<IClassifier>(),
                    drainInterval: TimeSpan.FromDays(1),
-                    terminalDrainTimeout: TimeSpan.FromSeconds(2));
+                    terminalDrainTimeout: TimeSpan.FromSeconds(2),
+                    schedulerOverride: schedulerOverride);

            public void Dispose() => Tmp.Dispose();
        }
@@ -477,6 +478,67 @@ namespace AutopilotMonitor.Agent.V2.Core.Tests.Orchestration
            sut.Stop();
        }

+        // ============================================================ Codex post-#50 #E — rehydrate-failure terminal path
+
+        [Fact]
+        public void Rehydrate_failure_posts_EffectInfrastructureFailure_and_session_terminates_to_Failed()
+        {
+            // Codex follow-up (post-#50 #E): if the scheduler can't re-arm a persisted
+            // deadline during recovery (ScheduleDeadline throws), DecisionState still claims
+            // the deadline is active — same phantom-deadline hang class as the live
+            // ScheduleDeadline/CancelDeadline failures. The orchestrator must post a
+            // synthetic EffectInfrastructureFailure (v1 payload) so the reducer terminates
+            // the session to Stage=Failed / Outcome=EnrollmentFailed rather than waiting
+            // out the max-lifetime watchdog.
+            using var rig = new Rig();
+            Directory.CreateDirectory(rig.StateDir);
+
+            // Seed a snapshot with ONE persisted deadline so rehydrate runs.
+            var persistedDeadline = new ActiveDeadline(
+                name: "safety_rearm_target",
+                dueAtUtc: At.AddMinutes(30), // future, so the scheduler would normally schedule it
+                firesSignalKind: DecisionSignalKind.DeadlineFired);
+            var seed = DecisionState.CreateInitial("S1", "T1")
+                .ToBuilder()
+                .WithStage(SessionStage.AwaitingHello)
+                .WithStepIndex(3)
+                .WithLastAppliedSignalOrdinal(2)
+                .AddDeadline(persistedDeadline)
+                .Build();
+            new SnapshotPersistence(
+                Path.Combine(rig.StateDir, "snapshot.json"),
+                () => rig.Clock.UtcNow).Save(seed);
+
+            // Broken scheduler — throws on every Schedule call, which is what
+            // RehydrateFromSnapshot drives internally.
+            var brokenScheduler = new FakeDeadlineScheduler
+            {
+                ThrowOnSchedule = new InvalidOperationException("timer-subsystem-dead"),
+            };
+
+            var sut = rig.Build(schedulerOverride: brokenScheduler);
+            sut.Start();
+
+            // Expected terminal state: reducer handles EffectInfrastructureFailure →
+            // Stage.Failed, Outcome.EnrollmentFailed, ClearDeadlines.
+            Assert.True(
+                SpinWait.SpinUntil(() => sut.CurrentState.Stage == SessionStage.Failed, 5000),
+                $"Expected terminal Failed stage after rehydrate failure; stuck at {sut.CurrentState.Stage}.");
+            Assert.Equal(SessionOutcome.EnrollmentFailed, sut.CurrentState.Outcome);
+
+            // Signal landed durably on the SignalLog with the v1 contract payload.
+            var signalLog = GetSignalLog(sut);
+            var failureSignal = signalLog.ReadAll()
+                .FirstOrDefault(s => s.Kind == DecisionSignalKind.EffectInfrastructureFailure);
+            Assert.NotNull(failureSignal);
+            Assert.NotNull(failureSignal!.Payload);
+            Assert.Contains("deadline_rehydrate_failure", failureSignal.Payload!["reason"]);
+            Assert.Equal("ScheduleDeadline", failureSignal.Payload["failingEffect"]);
+            Assert.Equal("effectrunner:critical:ScheduleDeadline", failureSignal.SourceOrigin);
+
+            sut.Stop();
+        }
+
        // ============================================================ Codex #1 Phase 3 — deadline re-arm

        [Fact]
@@ -294,11 +294,21 @@ namespace AutopilotMonitor.Agent.V2.Core.Tests.Orchestration
            //  (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.
+            //
+            // The synthetic signal MUST match the v1 payload contract defined on
+            // DecisionSignalKind.EffectInfrastructureFailure (Codex follow-up post-#50 #F):
+            // { reason, failingEffect }, sourceOrigin "effectrunner:critical:<EffectKind>",
+            // Evidence.Identifier "effect_infrastructure_failure:<EffectKind>".
            using var rig = new Rig();
+            var critical = new EffectFailure(
+                effectKind: DecisionEffectKind.ScheduleDeadline,
+                errorReason: "timer-broken",
+                isTransient: false,
+                exhaustedRetries: false);
            var abort = new EffectRunResult(
                sessionMustAbort: true,
-                abortReason: "timer_infrastructure_failure: scheduler-died",
-                failures: Array.Empty<EffectFailure>(),
+                abortReason: "timer_infrastructure_failure: timer-broken",
+                failures: new[] { critical },
                classifierInvocations: 0,
                classifierSkippedByAntiLoop: 0);
            rig.Processor.ScriptResult(abort);
@@ -319,10 +329,14 @@ namespace AutopilotMonitor.Agent.V2.Core.Tests.Orchestration
            Assert.Equal(0L, logged[0].SessionSignalOrdinal);
            Assert.Equal(1L, logged[1].SessionSignalOrdinal);

-            // Synthetic signal payload echoes the abort reason for forensics.
+            // v1 contract: payload carries reason AND failingEffect.
            Assert.NotNull(logged[1].Payload);
            Assert.Equal(abort.AbortReason, logged[1].Payload!["reason"]);
-            Assert.Equal("signal-ingress:inline-abort", logged[1].SourceOrigin);
+            Assert.Equal("ScheduleDeadline", logged[1].Payload!["failingEffect"]);
+            // sourceOrigin + evidence discriminate between ScheduleDeadline and CancelDeadline
+            // failures so forensic parsers (and the v1 fixture file) see a consistent shape.
+            Assert.Equal("effectrunner:critical:ScheduleDeadline", logged[1].SourceOrigin);
+            Assert.Equal("effect_infrastructure_failure:ScheduleDeadline", logged[1].Evidence.Identifier);

            // Processor received both signals, in order, with matching signals.
            var calls = rig.Processor.Calls;
@@ -331,6 +345,38 @@ namespace AutopilotMonitor.Agent.V2.Core.Tests.Orchestration
            Assert.Equal(DecisionSignalKind.EffectInfrastructureFailure, calls[1].Signal.Kind);
        }

+        [Fact]
+        public void SessionMustAbort_CancelDeadline_failure_labels_signal_with_CancelDeadline_kind()
+        {
+            // Same contract as the ScheduleDeadline case but for the CancelDeadline variant,
+            // so forensic queries can distinguish the two failure modes by sourceOrigin alone.
+            using var rig = new Rig();
+            var critical = new EffectFailure(
+                effectKind: DecisionEffectKind.CancelDeadline,
+                errorReason: "cancel-failed",
+                isTransient: false,
+                exhaustedRetries: false);
+            var abort = new EffectRunResult(
+                sessionMustAbort: true,
+                abortReason: "timer_infrastructure_failure: cancel-failed",
+                failures: new[] { critical },
+                classifierInvocations: 0,
+                classifierSkippedByAntiLoop: 0);
+            rig.Processor.ScriptResult(abort);
+
+            var ing = rig.Build();
+            ing.Start();
+            ing.Post(DecisionSignalKind.SessionStarted, At, "Collector", RawEvidence("original"));
+            Assert.True(WaitFor(() => rig.Processor.ApplyCallCount >= 2, timeoutMs: 5000));
+            ing.Stop();
+
+            var logged = rig.SignalLog.ReadAll();
+            Assert.Equal(2, logged.Count);
+            Assert.Equal("CancelDeadline", logged[1].Payload!["failingEffect"]);
+            Assert.Equal("effectrunner:critical:CancelDeadline", logged[1].SourceOrigin);
+            Assert.Equal("effect_infrastructure_failure:CancelDeadline", logged[1].Evidence.Identifier);
+        }
+
        [Fact]
        public void Synthetic_abort_signal_itself_reporting_abort_does_not_recurse()
        {
@@ -80,6 +80,10 @@ namespace AutopilotMonitor.Agent.V2.Core.Orchestration
        private readonly TimeSpan _terminalDrainTimeout;
        private readonly TimeSpan? _agentMaxLifetime;

+        // Test-only override so tests can drive rehydrate-failure via FakeDeadlineScheduler.
+        // Production always passes null → Start() creates the real DeadlineScheduler.
+        private readonly IDeadlineScheduler? _schedulerOverride;
+
        // Max-lifetime watchdog (M4.6.α). Timer is armed in Start() when _agentMaxLifetime != null.
        private System.Threading.Timer? _maxLifetimeTimer;
        private int _terminatedFired;
@@ -97,7 +101,7 @@ namespace AutopilotMonitor.Agent.V2.Core.Orchestration
        private TelemetryTransitionEmitter? _transitionEmitter;
        private EventTimelineEmitter? _timelineEmitter;
        private BackPressureEventObserver? _backPressureObserver;
-        private DeadlineScheduler? _scheduler;
+        private IDeadlineScheduler? _scheduler;
        private ClassifierRegistry? _classifierRegistry;
        private LazyIngressSinkRelay? _sinkRelay;
        private EffectRunner? _effectRunner;
@@ -140,7 +144,8 @@ namespace AutopilotMonitor.Agent.V2.Core.Orchestration
            int quarantineThreshold = DecisionStepProcessor.DefaultQuarantineThreshold,
            TimeSpan? drainInterval = null,
            TimeSpan? terminalDrainTimeout = null,
-            TimeSpan? agentMaxLifetime = null)
+            TimeSpan? agentMaxLifetime = null,
+            IDeadlineScheduler? schedulerOverride = null)
        {
            if (string.IsNullOrEmpty(sessionId)) throw new ArgumentException("SessionId is mandatory.", nameof(sessionId));
            if (string.IsNullOrEmpty(tenantId)) throw new ArgumentException("TenantId is mandatory.", nameof(tenantId));
@@ -174,6 +179,7 @@ namespace AutopilotMonitor.Agent.V2.Core.Orchestration
            if (agentMaxLifetime.HasValue && agentMaxLifetime.Value <= TimeSpan.Zero)
                throw new ArgumentOutOfRangeException(nameof(agentMaxLifetime), "AgentMaxLifetime must be positive when set.");
            _agentMaxLifetime = agentMaxLifetime;
+            _schedulerOverride = schedulerOverride;
        }

        /// <summary>
@@ -434,7 +440,7 @@ namespace AutopilotMonitor.Agent.V2.Core.Orchestration
            _backPressureObserver = new BackPressureEventObserver(eventEmitter, _clock);

            // 5) Deadlines + Classifiers.
-            _scheduler = new DeadlineScheduler(_clock);
+            _scheduler = _schedulerOverride ?? new DeadlineScheduler(_clock);
            _classifierRegistry = new ClassifierRegistry(_classifiers);

            // 6) Relay — EffectRunner braucht ISignalIngressSink, aber Ingress wird erst später
@@ -522,13 +528,41 @@ namespace AutopilotMonitor.Agent.V2.Core.Orchestration
                }
                catch (Exception ex)
                {
-                    // Codex follow-up #2 will convert this into a synthetic
-                    // EffectInfrastructureFailure → enrollment_failed transition. For now
-                    // we log at error so operators see the symptom — the max-lifetime
-                    // watchdog remains the last-resort termination path.
+                    // Codex follow-up (post-#50 #E): rehydrate failure is the same class of
+                    // phantom-deadline hang as live ScheduleDeadline/CancelDeadline failures.
+                    // DecisionState.Deadlines still claims the deadline exists, but the live
+                    // scheduler never registered it — so only the max-lifetime watchdog would
+                    // ever terminate the session. Post a synthetic EffectInfrastructureFailure
+                    // with the v1 contract payload so the reducer transitions cleanly to
+                    // Failed / EnrollmentFailed on the next worker tick. Ingress is already
+                    // running at this point in Start(), so the signal gets the normal
+                    // durable SignalLog.Append + Reduce treatment.
                    _logger.Error(
-                        "EnrollmentOrchestrator: deadline rehydration failed; session running without some deadlines.",
+                        "EnrollmentOrchestrator: deadline rehydration failed; posting EffectInfrastructureFailure to terminate the session.",
                        ex);
+                    try
+                    {
+                        var reason = $"deadline_rehydrate_failure: {ex.GetType().Name}: {ex.Message}";
+                        _ingress.Post(
+                            kind: DecisionSignalKind.EffectInfrastructureFailure,
+                            occurredAtUtc: _clock.UtcNow,
+                            sourceOrigin: "effectrunner:critical:ScheduleDeadline",
+                            evidence: new Evidence(
+                                kind: EvidenceKind.Synthetic,
+                                identifier: "effect_infrastructure_failure:ScheduleDeadline",
+                                summary: $"Critical effect ScheduleDeadline failed: {reason}"),
+                            payload: new Dictionary<string, string>(StringComparer.Ordinal)
+                            {
+                                ["reason"] = reason,
+                                ["failingEffect"] = nameof(DecisionEffectKind.ScheduleDeadline),
+                            });
+                    }
+                    catch (Exception postEx)
+                    {
+                        _logger.Error(
+                            "EnrollmentOrchestrator: failed to post EffectInfrastructureFailure after rehydrate failure — max-lifetime watchdog remains the last-resort terminator.",
+                            postEx);
+                    }
                }
            }

@@ -368,20 +368,46 @@ namespace AutopilotMonitor.Agent.V2.Core.Orchestration
                && effectResult.SessionMustAbort
                && signal.Kind != DecisionSignalKind.EffectInfrastructureFailure)
            {
-                ProcessInlineAbortSignal(effectResult.AbortReason, signal.OccurredAtUtc);
+                // Extract the failing critical effect kind from the result so the synthetic
+                // signal payload matches the v1 contract (reason + failingEffect). The
+                // critical failure is always the LAST entry that was appended before
+                // EffectRunner returned with SessionMustAbort — transient failures before
+                // it are harmless observability. Fall back to any non-transient failure if
+                // ordering ever changes.
+                DecisionEffectKind? failingEffect = null;
+                for (var i = effectResult.Failures.Count - 1; i >= 0; i--)
+                {
+                    if (!effectResult.Failures[i].IsTransient)
+                    {
+                        failingEffect = effectResult.Failures[i].EffectKind;
+                        break;
+                    }
+                }
+
+                ProcessInlineAbortSignal(effectResult.AbortReason, signal.OccurredAtUtc, failingEffect);
            }
        }

        /// <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.
+        /// this closes the crash-window identified by Codex follow-up post-#50 #B. The
+        /// signal payload matches the v1 contract documented on
+        /// <see cref="DecisionSignalKind.EffectInfrastructureFailure"/>:
+        /// <c>{ reason, failingEffect }</c>, with source/evidence identifying the critical
+        /// effect kind that triggered the abort so forensic consumers can distinguish
+        /// ScheduleDeadline-vs-CancelDeadline failures.
        /// </summary>
-        private void ProcessInlineAbortSignal(string? abortReason, DateTime occurredAtUtc)
+        private void ProcessInlineAbortSignal(
+            string? abortReason,
+            DateTime occurredAtUtc,
+            DecisionEffectKind? failingEffect)
        {
            long signalOrdinal = _nextSignalOrdinal;
            long traceOrdinal = _traceCounter.Next();

+            var failingEffectLabel = failingEffect?.ToString() ?? "Unknown";
+
            DecisionSignal syntheticSignal;
            try
            {
@@ -391,14 +417,15 @@ namespace AutopilotMonitor.Agent.V2.Core.Orchestration
                    kind: DecisionSignalKind.EffectInfrastructureFailure,
                    kindSchemaVersion: 1,
                    occurredAtUtc: occurredAtUtc,
-                    sourceOrigin: "signal-ingress:inline-abort",
+                    sourceOrigin: $"effectrunner:critical:{failingEffectLabel}",
                    evidence: new Evidence(
                        kind: EvidenceKind.Synthetic,
-                        identifier: "effect_infrastructure_failure:inline",
-                        summary: $"Inline abort synthesised by SignalIngress: {abortReason ?? "<unspecified>"}"),
+                        identifier: $"effect_infrastructure_failure:{failingEffectLabel}",
+                        summary: $"Critical effect {failingEffectLabel} failed: {abortReason ?? "<unspecified>"}"),
                    payload: new Dictionary<string, string>(StringComparer.Ordinal)
                    {
                        ["reason"] = abortReason ?? string.Empty,
+                        ["failingEffect"] = failingEffectLabel,
                    });
            }
            catch
@@ -1 +1 @@
-293
+296
@@ -1 +1 @@
-1134
+1137