v2 codex-02: EffectRunner posts EffectInfrastructureFailure on critical fail

Closes Codex review follow-up #2. Before this commit, a
DecisionEffectKind.ScheduleDeadline / CancelDeadline failure raised
EffectRunResult.SessionMustAbort but the decision world never heard
about it — DecisionStepProcessor only logged and continued. A safety
deadline could live in DecisionState without ever being armed on the
scheduler, leaving the session to hang until the max-lifetime
watchdog tripped hours later.

Flow now:

  1. EffectRunner.TryRunCriticalDeadlineEffect fails.
  2. EffectRunner posts a synthetic
     DecisionSignalKind.EffectInfrastructureFailure signal onto the
     ingress before returning. Payload: reason + failingEffect.
  3. Ingress worker picks it up next iteration.
  4. HandleEffectInfrastructureFailureV1 (new, in DecisionEngine.Shared.cs)
     transitions Stage=Failed, Outcome=EnrollmentFailed, ClearDeadlines,
     and emits one EmitEventTimelineEntry effect for enrollment_failed
     with the reason from the payload.
  5. DecisionStepProcessor observes the terminal stage and fires the
     Terminated event; EnrollmentOrchestrator tears down normally.

EnrollmentFailed (not Aborted) distinguishes infrastructure-induced
failures from user/operator-initiated aborts in forensics. The ingress
post is wrapped in a try/catch so a full/closed channel cannot crash
the worker — the max-lifetime watchdog remains the last-resort exit.

Changes:
  - DecisionSignalKind: new EffectInfrastructureFailure enum value.
  - DecisionEngine: dispatch entry (v1).
  - DecisionEngine.Shared: HandleEffectInfrastructureFailureV1 handler,
    mirrors HandleSessionAbortedV1 shape but with EnrollmentFailed
    outcome and a failure timeline effect.
  - EffectRunner: PostEffectInfrastructureFailureSignal helper, called
    from the critical-effect path before returning SessionMustAbort.
  - EffectRunResult: XML-doc clarifies that SessionMustAbort is now
    observability-only (reducer terminates on its own).
  - DecisionStepProcessor: comment update — no code change needed, the
    synthetic signal reaches state naturally.
  - tests/fixtures/signal-kinds/effect-infrastructure-failure-v1.json:
    replay fixture per the enum's self-documented contract.

Tests (+8):
  - DecisionCore EffectInfrastructureFailureHandlerTests: transition
    shape, deadline clearing, enrollment_failed emission with reason,
    missing-reason fallback, double-post idempotency.
  - EffectRunnerTests: schedule-fail posts signal with correct payload,
    cancel-fail same, ingress-post exception tolerated.
  - FakeSignalIngressSink: ThrowOnPost toggle for the tolerance test.

All suites green: DecisionCore 144/144 (+5), Agent.V2.Core 633/633
(+3), Functions 727/727. Build 0 warnings.

Plan: .claude/plans/codex-02-effect-failure-to-decision.md.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
Oliver Kieselbach
2026-04-24 11:07:43 +02:00
parent 96854d20ac
commit 386f3e909f
10 changed files with 333 additions and 3 deletions
@@ -90,6 +90,72 @@ namespace AutopilotMonitor.Agent.V2.Core.Tests.Orchestration
Assert.False(result.Failures[0].IsTransient);
}
[Fact]
public async Task ScheduleDeadline_failure_posts_EffectInfrastructureFailure_signal_to_ingress()
{
// 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.
var rig = new Rig();
rig.Scheduler.ThrowOnSchedule = new InvalidOperationException("timer-broken");
var sut = rig.Build();
var effect = new DecisionEffect(
DecisionEffectKind.ScheduleDeadline,
deadline: Deadline("hello_safety", At.AddMinutes(1)));
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"]);
}
[Fact]
public async Task CancelDeadline_failure_posts_EffectInfrastructureFailure_signal_to_ingress()
{
var rig = new Rig();
rig.Scheduler.ThrowOnCancel = new InvalidOperationException("timer-broken");
var sut = rig.Build();
var effect = new DecisionEffect(
DecisionEffectKind.CancelDeadline,
cancelDeadlineName: "hello_safety");
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.
}
[Fact]
public async Task ScheduleDeadline_abort_stops_processing_subsequent_effects()
{
@@ -14,6 +14,9 @@ namespace AutopilotMonitor.Agent.V2.Core.Tests.Orchestration
public IReadOnlyList<PostedSignal> Posted => _posted;
/// <summary>When non-null, <see cref="Post"/> throws instead of capturing.</summary>
public Exception? ThrowOnPost { get; set; }
public void Post(
DecisionSignalKind kind,
DateTime occurredAtUtc,
@@ -23,6 +26,7 @@ namespace AutopilotMonitor.Agent.V2.Core.Tests.Orchestration
int kindSchemaVersion = 1,
object? typedPayload = null)
{
if (ThrowOnPost != null) throw ThrowOnPost;
_posted.Add(new PostedSignal(kind, occurredAtUtc, sourceOrigin, evidence, payload, kindSchemaVersion, typedPayload));
}
@@ -160,9 +160,14 @@ 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.
_logger.Error(
$"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.");
}
else if (effectResult.Failures.Count > 0)
{
@@ -9,7 +9,11 @@ namespace AutopilotMonitor.Agent.V2.Core.Orchestration
/// <para>
/// Der Orchestrator (M4.4) liest:
/// <list type="bullet">
/// <item><see cref="SessionMustAbort"/> → emittiert <c>enrollment_failed(reason=&lt;AbortReason&gt;)</c></item>
/// <item><see cref="SessionMustAbort"/> → ist rein Observability. Codex follow-up #2:
/// beim Setzen dieses Flags hat der EffectRunner bereits ein synthetisches
/// <c>EffectInfrastructureFailure</c>-Signal an den Ingress gestellt; der Reducer
/// terminiert die Session auf dem nächsten Step (Stage=Failed, Outcome=EnrollmentFailed,
/// <c>enrollment_failed</c>-Timeline-Event). Der Caller muss nur loggen.</item>
/// <item><see cref="Failures"/> → schreibt pro Fehler eine <c>effect_failure</c>-Transition ins Journal</item>
/// <item><see cref="ClassifierInvocations"/> / <see cref="ClassifierSkippedByAntiLoop"/> → Observability</item>
/// </list>
@@ -79,9 +79,18 @@ namespace AutopilotMonitor.Agent.V2.Core.Orchestration
if (criticalFail != null)
{
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);
return new EffectRunResult(
sessionMustAbort: true,
abortReason: $"timer_infrastructure_failure: {criticalFail.ErrorReason}",
abortReason: abortReason,
failures: failures,
classifierInvocations: invocations,
classifierSkippedByAntiLoop: skipped);
@@ -220,6 +229,41 @@ namespace AutopilotMonitor.Agent.V2.Core.Orchestration
return new ClassifierEffectOutcome(invoked: true, skippedByAntiLoop: false, failure: null);
}
/// <summary>
/// Codex follow-up #2 — synthesise an <see cref="DecisionSignalKind.EffectInfrastructureFailure"/>
/// signal when a critical deadline effect cannot be delivered. The reducer's
/// <c>HandleEffectInfrastructureFailureV1</c> picks this up next tick and transitions the
/// session to <see cref="SessionStage.Failed"/> / <see cref="SessionOutcome.EnrollmentFailed"/>.
/// Any exception from the ingress (full channel / disposed) is swallowed: the
/// max-lifetime watchdog remains the last-resort termination path.
/// </summary>
private void PostEffectInfrastructureFailureSignal(
DecisionEffectKind failingEffect,
string abortReason,
DateTime stepOccurredAtUtc)
{
try
{
_ingress.Post(
kind: DecisionSignalKind.EffectInfrastructureFailure,
occurredAtUtc: stepOccurredAtUtc,
sourceOrigin: $"effectrunner:critical:{failingEffect}",
evidence: new Evidence(
kind: EvidenceKind.Synthetic,
identifier: $"effect_infrastructure_failure:{failingEffect}",
summary: $"Critical effect {failingEffect} failed: {abortReason}"),
payload: new Dictionary<string, string>(StringComparer.Ordinal)
{
["reason"] = abortReason,
["failingEffect"] = failingEffect.ToString(),
});
}
catch
{
// Intentionally swallowed — see method doc.
}
}
private void PostClassifierVerdictSignal(ClassifierVerdict verdict, string classifierId, DateTime stepOccurredAtUtc)
{
_ingress.Post(
@@ -0,0 +1,141 @@
using System;
using System.Collections.Generic;
using System.Linq;
using AutopilotMonitor.DecisionCore.Engine;
using AutopilotMonitor.DecisionCore.Signals;
using AutopilotMonitor.DecisionCore.State;
using AutopilotMonitor.Shared.Models;
using Xunit;
namespace AutopilotMonitor.DecisionCore.Tests
{
/// <summary>
/// Codex follow-up #2 — reducer handler for the synthetic
/// <see cref="DecisionSignalKind.EffectInfrastructureFailure"/> signal that the
/// EffectRunner posts when a critical deadline effect cannot be delivered.
/// These tests pin the terminal-transition contract so the orchestrator never
/// again observes <see cref="EffectRunResult.SessionMustAbort"/> without a
/// matching state-machine transition.
/// </summary>
public sealed class EffectInfrastructureFailureHandlerTests
{
private static DecisionSignal MakeSignal(
string? reason = "timer_infrastructure_failure: InvalidOperationException: timer-broken",
long ordinal = 5)
{
var payload = new Dictionary<string, string>(StringComparer.Ordinal);
if (reason != null) payload["reason"] = reason;
payload["failingEffect"] = "ScheduleDeadline";
return new DecisionSignal(
sessionSignalOrdinal: ordinal,
sessionTraceOrdinal: ordinal,
kind: DecisionSignalKind.EffectInfrastructureFailure,
kindSchemaVersion: 1,
occurredAtUtc: new DateTime(2026, 4, 24, 10, 0, 0, DateTimeKind.Utc),
sourceOrigin: "effectrunner:critical:ScheduleDeadline",
evidence: new Evidence(
EvidenceKind.Synthetic,
"effect_infrastructure_failure:ScheduleDeadline",
"Critical effect ScheduleDeadline failed"),
payload: payload);
}
[Fact]
public void Handler_transitions_to_Failed_with_EnrollmentFailed_outcome()
{
var engine = new DecisionEngine();
var state = DecisionState.CreateInitial("s", "t")
.ToBuilder()
.WithStage(SessionStage.AwaitingHello)
.WithStepIndex(4)
.WithLastAppliedSignalOrdinal(4)
.Build();
var step = engine.Reduce(state, MakeSignal(ordinal: 5));
Assert.True(step.Transition.Taken);
Assert.Equal(SessionStage.Failed, step.NewState.Stage);
Assert.Equal(SessionOutcome.EnrollmentFailed, step.NewState.Outcome);
Assert.Equal(5, step.NewState.StepIndex);
Assert.Equal(5, step.NewState.LastAppliedSignalOrdinal);
Assert.Equal(
nameof(DecisionSignalKind.EffectInfrastructureFailure),
step.Transition.Trigger);
}
[Fact]
public void Handler_clears_active_deadlines()
{
var engine = new DecisionEngine();
var state = DecisionState.CreateInitial("s", "t")
.ToBuilder()
.WithStage(SessionStage.AwaitingHello)
.AddDeadline(new ActiveDeadline(
name: "hello_safety",
dueAtUtc: new DateTime(2026, 4, 24, 10, 5, 0, DateTimeKind.Utc),
firesSignalKind: DecisionSignalKind.DeadlineFired))
.AddDeadline(new ActiveDeadline(
name: "classifier_tick",
dueAtUtc: new DateTime(2026, 4, 24, 10, 10, 0, DateTimeKind.Utc),
firesSignalKind: DecisionSignalKind.DeadlineFired))
.Build();
var step = engine.Reduce(state, MakeSignal());
Assert.Empty(step.NewState.Deadlines);
}
[Fact]
public void Handler_emits_enrollment_failed_timeline_event_with_reason()
{
var engine = new DecisionEngine();
var state = DecisionState.CreateInitial("s", "t");
var step = engine.Reduce(state, MakeSignal(reason: "custom-reason-abc"));
Assert.Single(step.Effects);
var effect = step.Effects[0];
Assert.Equal(DecisionEffectKind.EmitEventTimelineEntry, effect.Kind);
Assert.NotNull(effect.Parameters);
Assert.Equal("enrollment_failed", effect.Parameters!["eventType"]);
Assert.Equal("custom-reason-abc", effect.Parameters["reason"]);
}
[Fact]
public void Handler_falls_back_to_default_reason_when_payload_missing_key()
{
// Payload without a "reason" key — reducer must still produce a valid
// enrollment_failed event rather than leaving reason null/empty.
var engine = new DecisionEngine();
var state = DecisionState.CreateInitial("s", "t");
var signal = MakeSignal(reason: null); // helper strips the "reason" key when null
var step = engine.Reduce(state, signal);
Assert.Single(step.Effects);
Assert.Equal("effect_infrastructure_failure", step.Effects[0].Parameters!["reason"]);
}
[Fact]
public void Handler_from_terminal_state_keeps_Failed_outcome()
{
// Defensive: double-post (e.g. two critical effects in one EffectRun) must not
// regress the terminal outcome or leak deadlines back in.
var engine = new DecisionEngine();
var alreadyFailed = DecisionState.CreateInitial("s", "t")
.ToBuilder()
.WithStage(SessionStage.Failed)
.WithOutcome(SessionOutcome.EnrollmentFailed)
.WithStepIndex(8)
.WithLastAppliedSignalOrdinal(7)
.Build();
var step = engine.Reduce(alreadyFailed, MakeSignal(ordinal: 8));
Assert.Equal(SessionStage.Failed, step.NewState.Stage);
Assert.Equal(SessionOutcome.EnrollmentFailed, step.NewState.Outcome);
Assert.Empty(step.NewState.Deadlines);
}
}
}
@@ -108,6 +108,57 @@ namespace AutopilotMonitor.DecisionCore.Engine
return new DecisionStep(newState, transition, Array.Empty<DecisionEffect>());
}
/// <summary>
/// Handle <see cref="DecisionSignalKind.EffectInfrastructureFailure"/>. Codex
/// follow-up #2 — posted synchronously by the EffectRunner when a critical
/// effect (ScheduleDeadline / CancelDeadline) cannot reach the scheduler.
/// Without this reducer branch, <see cref="EffectRunResult.SessionMustAbort"/>
/// would be observable only in the log and the session could hang on a
/// phantom deadline that was never actually armed.
/// <para>
/// Transition shape mirrors <see cref="HandleSessionAbortedV1"/>: stage →
/// <see cref="SessionStage.Failed"/>, outcome →
/// <see cref="SessionOutcome.EnrollmentFailed"/> (distinguishes an
/// infrastructure-induced failure from a user/operator-initiated
/// <see cref="SessionOutcome.Aborted"/>), all <c>ActiveDeadlines</c> cleared.
/// One <see cref="DecisionEffectKind.EmitEventTimelineEntry"/> effect
/// publishes <c>enrollment_failed</c> so backend + UI see a proper terminal
/// record with the failure reason from the signal payload.
/// </para>
/// </summary>
private DecisionStep HandleEffectInfrastructureFailureV1(DecisionState state, DecisionSignal signal)
{
var reason = signal.Payload != null && signal.Payload.TryGetValue("reason", out var r) && !string.IsNullOrEmpty(r)
? r
: "effect_infrastructure_failure";
var nextStep = state.StepIndex + 1;
var newState = state.ToBuilder()
.WithStage(SessionStage.Failed)
.WithOutcome(SessionOutcome.EnrollmentFailed)
.WithStepIndex(nextStep)
.WithLastAppliedSignalOrdinal(signal.SessionSignalOrdinal)
.ClearDeadlines()
.Build();
var failEffect = new DecisionEffect(
DecisionEffectKind.EmitEventTimelineEntry,
parameters: new Dictionary<string, string>(StringComparer.Ordinal)
{
["eventType"] = "enrollment_failed",
["reason"] = reason,
});
var transition = BuildTakenTransition(
before: state,
signal: signal,
toStage: SessionStage.Failed,
nextStepIndex: nextStep,
trigger: nameof(DecisionSignalKind.EffectInfrastructureFailure));
return new DecisionStep(newState, transition, new[] { failEffect });
}
/// <summary>
/// Handle <see cref="DecisionSignalKind.SessionAborted"/>.
/// <para>
@@ -61,6 +61,7 @@ namespace AutopilotMonitor.DecisionCore.Engine
(DecisionSignalKind.SessionAborted, 1) => HandleSessionAbortedV1(state, signal),
(DecisionSignalKind.AdminPreemptionDetected, 1) => HandleAdminPreemptionDetectedV1(state, signal),
(DecisionSignalKind.DeadlineFired, 1) => HandleDeadlineFiredV1(state, signal),
(DecisionSignalKind.EffectInfrastructureFailure, 1) => HandleEffectInfrastructureFailureV1(state, signal),
// ----- Classic UserDriven-v1 (DecisionEngine.Classic.cs) -----
(DecisionSignalKind.EspPhaseChanged, 1) => HandleEspPhaseChangedV1(state, signal),
@@ -38,6 +38,14 @@ namespace AutopilotMonitor.DecisionCore.Signals
DeadlineFired,
ClassifierVerdictIssued,
// Codex follow-up #2 — posted by EffectRunner when a critical effect
// (ScheduleDeadline / CancelDeadline) fails so the orchestrator's timer
// infrastructure cannot enforce a just-decided safety-net deadline.
// Carries payload { "reason": "<abortReason>", "failingEffect": "<EffectKind>" }.
// The reducer's HandleEffectInfrastructureFailureV1 transitions the session
// to Failed with SessionOutcome.EnrollmentFailed and emits enrollment_failed.
EffectInfrastructureFailure,
// --- Lifecycle ---
SessionStarted,
SessionAborted,
@@ -0,0 +1,6 @@
# EffectInfrastructureFailure v1 — Codex follow-up #2.
# Posted by the EffectRunner when a critical deadline effect fails to reach the
# scheduler (ScheduleDeadline / CancelDeadline). Reducer:
# HandleEffectInfrastructureFailureV1 → Stage.Failed, Outcome=EnrollmentFailed,
# ClearDeadlines, emits enrollment_failed timeline entry with the reason.
{"SessionSignalOrdinal":1,"SessionTraceOrdinal":1,"Kind":"EffectInfrastructureFailure","KindSchemaVersion":1,"OccurredAtUtc":"2026-04-24T10:00:00.0000000Z","SourceOrigin":"effectrunner:critical:ScheduleDeadline","Evidence":{"Kind":"Synthetic","Identifier":"effect_infrastructure_failure:ScheduleDeadline","Summary":"Critical effect ScheduleDeadline failed: timer_infrastructure_failure: InvalidOperationException: timer-broken"},"Payload":{"reason":"timer_infrastructure_failure: InvalidOperationException: timer-broken","failingEffect":"ScheduleDeadline"}}