mirror of
https://github.com/okieselbach/AutopilotMonitor.git
synced 2026-10-25 05:00:07 +00:00
feat(agent-v2): refactor RecoveryCoordingator extraction
This commit is contained in:
Oliver Kieselbach
+484
@@ -0,0 +1,484 @@
|
||||
using System;
|
||||
using System.Collections.Generic;
|
||||
using System.IO;
|
||||
using System.Linq;
|
||||
using System.Text;
|
||||
using AutopilotMonitor.Agent.V2.Core.Logging;
|
||||
using AutopilotMonitor.Agent.V2.Core.Orchestration;
|
||||
using AutopilotMonitor.Agent.V2.Core.Persistence;
|
||||
using AutopilotMonitor.Agent.V2.Core.Tests.Harness;
|
||||
using AutopilotMonitor.DecisionCore.Engine;
|
||||
using AutopilotMonitor.DecisionCore.Signals;
|
||||
using AutopilotMonitor.DecisionCore.State;
|
||||
using Xunit;
|
||||
|
||||
namespace AutopilotMonitor.Agent.V2.Core.Tests.Orchestration
|
||||
{
|
||||
/// <summary>
|
||||
/// Direct unit tests for <see cref="RecoveryCoordinator.Recover"/> — the crash-recovery
|
||||
/// state machine extracted from <c>EnrollmentOrchestrator.Start</c> (ARCH-F5). These exercise
|
||||
/// the recovery branches in isolation (no ingress worker, drain loop, collectors, or scheduler),
|
||||
/// asserting directly on the returned <see cref="RecoveryResult"/> and the on-disk side effects.
|
||||
/// <para>
|
||||
/// The orchestrator-level <c>RecoveryPathTests</c> still cover the same branches end-to-end
|
||||
/// (replay reaches the live pipeline, deadlines re-arm, the Terminated event fires). This class
|
||||
/// pins the coordinator's own contract: which branch is taken, which flags it returns, and which
|
||||
/// segment files it moves — including branches the end-to-end tests under-assert
|
||||
/// (<c>a-total-loss</c>, the prior-run quarantine-marker flags).
|
||||
/// </para>
|
||||
/// </summary>
|
||||
public sealed class RecoveryCoordinatorTests
|
||||
{
|
||||
private static readonly DateTime At = new DateTime(2026, 4, 20, 10, 0, 0, DateTimeKind.Utc);
|
||||
private const string SessionId = "S1";
|
||||
private const string TenantId = "T1";
|
||||
|
||||
private sealed class Rig : IDisposable
|
||||
{
|
||||
public TempDirectory Tmp { get; } = new TempDirectory();
|
||||
public VirtualClock Clock { get; } = new VirtualClock(At);
|
||||
public AgentLogger Logger { get; }
|
||||
public string StateDir { get; }
|
||||
|
||||
public Rig()
|
||||
{
|
||||
Logger = new AgentLogger(Tmp.Path, AgentLogLevel.Info);
|
||||
StateDir = Path.Combine(Tmp.Path, "State");
|
||||
Directory.CreateDirectory(StateDir);
|
||||
}
|
||||
|
||||
// The orchestrator runs EnsureDirectories() before Recover(); mirror that contract.
|
||||
public RecoveryResult Recover() =>
|
||||
RecoveryCoordinator.Recover(StateDir, SessionId, TenantId, Clock, Logger);
|
||||
|
||||
public string StatePath(string name) => System.IO.Path.Combine(StateDir, name);
|
||||
|
||||
public void Dispose() => Tmp.Dispose();
|
||||
}
|
||||
|
||||
// ----------------------------------------------------------------- Fixtures
|
||||
|
||||
private static DecisionSignal MakeSignal(
|
||||
long ordinal,
|
||||
DecisionSignalKind kind = DecisionSignalKind.AppInstallCompleted) =>
|
||||
new DecisionSignal(
|
||||
sessionSignalOrdinal: ordinal,
|
||||
sessionTraceOrdinal: ordinal,
|
||||
kind: kind,
|
||||
kindSchemaVersion: 1,
|
||||
occurredAtUtc: At.AddSeconds(ordinal),
|
||||
sourceOrigin: "recovery-coordinator-tests",
|
||||
evidence: new Evidence(EvidenceKind.Synthetic, $"recovery:ord-{ordinal}", "test signal"));
|
||||
|
||||
private static DecisionTransition MakeTransition(int stepIndex, long signalRef) =>
|
||||
new DecisionTransition(
|
||||
stepIndex: stepIndex,
|
||||
sessionTraceOrdinal: stepIndex,
|
||||
signalOrdinalRef: signalRef,
|
||||
occurredAtUtc: At.AddSeconds(stepIndex),
|
||||
trigger: "TestTrigger",
|
||||
fromStage: SessionStage.SessionStarted,
|
||||
toStage: SessionStage.SessionStarted,
|
||||
taken: true,
|
||||
deadEndReason: null,
|
||||
reducerVersion: "2.0.0.0");
|
||||
|
||||
private void SeedSnapshot(Rig rig, DecisionState state) =>
|
||||
new SnapshotPersistence(rig.StatePath("snapshot.json"), () => rig.Clock.UtcNow).Save(state);
|
||||
|
||||
private void SeedSignalLog(Rig rig, params DecisionSignal[] signals)
|
||||
{
|
||||
var writer = new SignalLogWriter(rig.StatePath("signal-log.jsonl"));
|
||||
foreach (var s in signals) writer.Append(s);
|
||||
}
|
||||
|
||||
// ================================================================= Branch c1 — fresh start
|
||||
|
||||
[Fact]
|
||||
public void Empty_state_directory_yields_fresh_initial_state_no_flags()
|
||||
{
|
||||
using var rig = new Rig();
|
||||
|
||||
var result = rig.Recover();
|
||||
|
||||
Assert.Equal(SessionStage.SessionStarted, result.InitialState.Stage);
|
||||
Assert.Equal(0, result.InitialState.StepIndex);
|
||||
Assert.Equal(SessionId, result.InitialState.SessionId);
|
||||
Assert.Equal(TenantId, result.InitialState.TenantId);
|
||||
|
||||
Assert.False(result.IsWhiteGlovePart2);
|
||||
Assert.False(result.WasStartupQuarantine);
|
||||
Assert.False(result.PriorRunQuarantined);
|
||||
Assert.Null(result.PriorRunQuarantineReason);
|
||||
|
||||
// All four writers are returned and ready for the wiring steps.
|
||||
Assert.NotNull(result.SignalLog);
|
||||
Assert.NotNull(result.Journal);
|
||||
Assert.NotNull(result.Snapshot);
|
||||
Assert.NotNull(result.EventSequence);
|
||||
}
|
||||
|
||||
// ================================================================= Branch b — clean snapshot
|
||||
|
||||
[Fact]
|
||||
public void Clean_snapshot_without_tail_is_loaded_as_seed()
|
||||
{
|
||||
using var rig = new Rig();
|
||||
var saved = DecisionState.CreateInitial(SessionId, TenantId)
|
||||
.ToBuilder()
|
||||
.WithStage(SessionStage.EspDeviceSetup)
|
||||
.WithStepIndex(3)
|
||||
.WithLastAppliedSignalOrdinal(2)
|
||||
.Build();
|
||||
SeedSnapshot(rig, saved);
|
||||
|
||||
var result = rig.Recover();
|
||||
|
||||
Assert.False(result.WasStartupQuarantine);
|
||||
Assert.False(result.IsWhiteGlovePart2);
|
||||
Assert.Equal(SessionStage.EspDeviceSetup, result.InitialState.Stage);
|
||||
Assert.Equal(3, result.InitialState.StepIndex);
|
||||
Assert.Equal(2, result.InitialState.LastAppliedSignalOrdinal);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void Clean_snapshot_restamps_agent_boot_utc_to_current_run()
|
||||
{
|
||||
// Branch (b) re-stamps AgentBootUtc to "now" so rehydrated deadlines floor at the
|
||||
// current run start rather than the prior run's (premature-fire safety).
|
||||
using var rig = new Rig();
|
||||
var priorBoot = At.AddHours(-5);
|
||||
var saved = DecisionState.CreateInitial(SessionId, TenantId, priorBoot)
|
||||
.ToBuilder()
|
||||
.WithStage(SessionStage.AwaitingHello)
|
||||
.WithStepIndex(4)
|
||||
.WithLastAppliedSignalOrdinal(3)
|
||||
.Build();
|
||||
SeedSnapshot(rig, saved);
|
||||
|
||||
var result = rig.Recover();
|
||||
|
||||
Assert.Equal(At, result.InitialState.AgentBootUtc);
|
||||
Assert.Equal(SessionStage.AwaitingHello, result.InitialState.Stage);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void Snapshot_plus_pending_tail_replays_tail_onto_snapshot()
|
||||
{
|
||||
using var rig = new Rig();
|
||||
|
||||
var sigs = new[]
|
||||
{
|
||||
MakeSignal(0, DecisionSignalKind.SessionStarted),
|
||||
MakeSignal(1, DecisionSignalKind.AppInstallCompleted),
|
||||
};
|
||||
SeedSignalLog(rig, sigs);
|
||||
|
||||
// Snapshot lags one step behind (only ordinal 0 consumed).
|
||||
var engine = new DecisionEngine();
|
||||
var lagged = engine.Reduce(DecisionState.CreateInitial(SessionId, TenantId), sigs[0]).NewState;
|
||||
SeedSnapshot(rig, lagged);
|
||||
|
||||
var expected = ReducerReplay.Replay(engine, DecisionState.CreateInitial(SessionId, TenantId), sigs);
|
||||
|
||||
var result = rig.Recover();
|
||||
|
||||
Assert.False(result.WasStartupQuarantine);
|
||||
Assert.Equal(expected.StepIndex, result.InitialState.StepIndex);
|
||||
Assert.Equal(expected.LastAppliedSignalOrdinal, result.InitialState.LastAppliedSignalOrdinal);
|
||||
Assert.Equal(expected.Stage, result.InitialState.Stage);
|
||||
}
|
||||
|
||||
// ================================================================= Branch a — corrupt snapshot
|
||||
|
||||
[Fact]
|
||||
public void Corrupt_snapshot_with_intact_log_quarantines_snapshot_only_and_replays_log()
|
||||
{
|
||||
using var rig = new Rig();
|
||||
|
||||
File.WriteAllText(
|
||||
rig.StatePath("snapshot.json"),
|
||||
"{\"Checksum\":\"deadbeef\",\"State\":{\"SessionId\":\"S1\",\"TenantId\":\"T1\"}}",
|
||||
Encoding.UTF8);
|
||||
|
||||
var sigs = new[]
|
||||
{
|
||||
MakeSignal(0, DecisionSignalKind.SessionStarted),
|
||||
MakeSignal(1, DecisionSignalKind.AppInstallCompleted),
|
||||
MakeSignal(2, DecisionSignalKind.AppInstallCompleted),
|
||||
};
|
||||
SeedSignalLog(rig, sigs);
|
||||
|
||||
var expected = ReducerReplay.Replay(
|
||||
new DecisionEngine(), DecisionState.CreateInitial(SessionId, TenantId), sigs);
|
||||
|
||||
var result = rig.Recover();
|
||||
|
||||
Assert.True(result.WasStartupQuarantine);
|
||||
Assert.False(result.IsWhiteGlovePart2);
|
||||
Assert.Equal(expected.StepIndex, result.InitialState.StepIndex);
|
||||
Assert.Equal(expected.LastAppliedSignalOrdinal, result.InitialState.LastAppliedSignalOrdinal);
|
||||
|
||||
// Snapshot moved aside; SignalLog survived (NOT quarantined).
|
||||
Assert.False(File.Exists(rig.StatePath("snapshot.json")));
|
||||
Assert.True(File.Exists(rig.StatePath("signal-log.jsonl")));
|
||||
Assert.True(new FileInfo(rig.StatePath("signal-log.jsonl")).Length > 0);
|
||||
|
||||
Assert.True(Directory.Exists(Path.Combine(rig.StateDir, ".quarantine")));
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void Corrupt_snapshot_with_head_corrupt_log_escalates_to_total_loss_quarantine()
|
||||
{
|
||||
// Branch (a-total-loss): the snapshot is corrupt AND the SignalLog is unreadable from
|
||||
// its first line (non-empty file, zero parseable signals). Both segment groups are
|
||||
// quarantined and the run seeds from a fresh Initial state. This is the branch the
|
||||
// end-to-end RecoveryPathTests do not directly exercise.
|
||||
using var rig = new Rig();
|
||||
|
||||
File.WriteAllText(
|
||||
rig.StatePath("snapshot.json"),
|
||||
"{\"Checksum\":\"deadbeef\",\"State\":{\"SessionId\":\"S1\",\"TenantId\":\"T1\"}}",
|
||||
Encoding.UTF8);
|
||||
|
||||
// Non-empty but unparseable from line one → ReadAll yields 0 → head-corrupt.
|
||||
File.WriteAllText(rig.StatePath("signal-log.jsonl"), "not-json-garbage\n", Encoding.UTF8);
|
||||
|
||||
var result = rig.Recover();
|
||||
|
||||
Assert.True(result.WasStartupQuarantine);
|
||||
Assert.Equal(SessionStage.SessionStarted, result.InitialState.Stage);
|
||||
Assert.Equal(0, result.InitialState.StepIndex);
|
||||
|
||||
// The garbage log was moved into the quarantine bucket; the live writer is fresh.
|
||||
Assert.True(Directory.Exists(Path.Combine(rig.StateDir, ".quarantine")));
|
||||
if (File.Exists(rig.StatePath("signal-log.jsonl")))
|
||||
{
|
||||
var fresh = File.ReadAllText(rig.StatePath("signal-log.jsonl"));
|
||||
Assert.DoesNotContain("not-json-garbage", fresh);
|
||||
}
|
||||
|
||||
// Fresh writer's recovery counters reset to -1 (no stale ordinal carried over).
|
||||
Assert.Equal(-1, result.SignalLog.LastOrdinal);
|
||||
}
|
||||
|
||||
// ================================================================= Branch c2 — no snapshot, intact log
|
||||
|
||||
[Fact]
|
||||
public void No_snapshot_with_intact_log_rebuilds_via_full_log_replay()
|
||||
{
|
||||
using var rig = new Rig();
|
||||
|
||||
var sigs = new[]
|
||||
{
|
||||
MakeSignal(0, DecisionSignalKind.SessionStarted),
|
||||
MakeSignal(1, DecisionSignalKind.AppInstallCompleted),
|
||||
MakeSignal(2, DecisionSignalKind.AppInstallCompleted),
|
||||
};
|
||||
SeedSignalLog(rig, sigs);
|
||||
Assert.False(File.Exists(rig.StatePath("snapshot.json")));
|
||||
|
||||
var expected = ReducerReplay.Replay(
|
||||
new DecisionEngine(), DecisionState.CreateInitial(SessionId, TenantId), sigs);
|
||||
|
||||
var result = rig.Recover();
|
||||
|
||||
Assert.True(result.WasStartupQuarantine);
|
||||
Assert.Equal(expected.StepIndex, result.InitialState.StepIndex);
|
||||
Assert.Equal(expected.LastAppliedSignalOrdinal, result.InitialState.LastAppliedSignalOrdinal);
|
||||
|
||||
// Log was not quarantined — only the missing snapshot drove the replay.
|
||||
Assert.True(File.Exists(rig.StatePath("signal-log.jsonl")));
|
||||
Assert.True(new FileInfo(rig.StatePath("signal-log.jsonl")).Length > 0);
|
||||
}
|
||||
|
||||
// ================================================================= Journal alignment
|
||||
|
||||
[Fact]
|
||||
public void Journal_ahead_of_seed_truncates_phantom_tail_and_backfills_in_lockstep()
|
||||
{
|
||||
using var rig = new Rig();
|
||||
|
||||
var sigs = new[]
|
||||
{
|
||||
MakeSignal(0, DecisionSignalKind.SessionStarted),
|
||||
MakeSignal(1, DecisionSignalKind.AppInstallCompleted),
|
||||
};
|
||||
SeedSignalLog(rig, sigs);
|
||||
|
||||
var engine = new DecisionEngine();
|
||||
var step0 = engine.Reduce(DecisionState.CreateInitial(SessionId, TenantId), sigs[0]);
|
||||
var step1 = engine.Reduce(step0.NewState, sigs[1]);
|
||||
SeedSnapshot(rig, step0.NewState); // StepIndex=1
|
||||
|
||||
// Journal holds 3 entries — StepIndex 3 is a phantom (no backing signal).
|
||||
var journalPath = rig.StatePath("journal.jsonl");
|
||||
var seedJournal = new JournalWriter(journalPath, () => rig.Clock.UtcNow);
|
||||
seedJournal.Append(step0.Transition); // StepIndex=1
|
||||
seedJournal.Append(step1.Transition); // StepIndex=2
|
||||
seedJournal.Append(MakeTransition(3, signalRef: 99)); // StepIndex=3 phantom
|
||||
|
||||
var expected = ReducerReplay.Replay(engine, step0.NewState, new[] { sigs[1] });
|
||||
|
||||
var result = rig.Recover();
|
||||
|
||||
Assert.Equal(expected.StepIndex, result.InitialState.StepIndex);
|
||||
Assert.Equal(expected.LastAppliedSignalOrdinal, result.InitialState.LastAppliedSignalOrdinal);
|
||||
|
||||
// Two entries beyond seed.StepIndex=1 went to the phantom bucket.
|
||||
var phantomFiles = Directory.GetFiles(
|
||||
Path.Combine(rig.StateDir, ".quarantine"),
|
||||
"journal-phantom-tail.jsonl",
|
||||
SearchOption.AllDirectories);
|
||||
Assert.Single(phantomFiles);
|
||||
Assert.Equal(2, File.ReadAllLines(phantomFiles[0]).Count(l => !string.IsNullOrWhiteSpace(l)));
|
||||
|
||||
// Live journal rebuilt in lockstep: [1, 2].
|
||||
var live = new JournalWriter(journalPath, () => rig.Clock.UtcNow).ReadAll();
|
||||
Assert.Equal(2, live.Count);
|
||||
Assert.Equal(1, live[0].StepIndex);
|
||||
Assert.Equal(2, live[1].StepIndex);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void Signal_log_ahead_of_journal_backfills_missing_journal_entries()
|
||||
{
|
||||
using var rig = new Rig();
|
||||
|
||||
var sigs = new[]
|
||||
{
|
||||
MakeSignal(0, DecisionSignalKind.SessionStarted),
|
||||
MakeSignal(1, DecisionSignalKind.AppInstallCompleted),
|
||||
};
|
||||
SeedSignalLog(rig, sigs);
|
||||
|
||||
var engine = new DecisionEngine();
|
||||
var step0 = engine.Reduce(DecisionState.CreateInitial(SessionId, TenantId), sigs[0]);
|
||||
SeedSnapshot(rig, step0.NewState); // StepIndex=1, LastApplied=0
|
||||
|
||||
// Journal only has the sig0 transition — sig1's was never flushed.
|
||||
var journalPath = rig.StatePath("journal.jsonl");
|
||||
new JournalWriter(journalPath, () => rig.Clock.UtcNow).Append(step0.Transition);
|
||||
|
||||
var expected = ReducerReplay.Replay(engine, step0.NewState, new[] { sigs[1] });
|
||||
|
||||
var result = rig.Recover();
|
||||
|
||||
Assert.Equal(expected.StepIndex, result.InitialState.StepIndex);
|
||||
Assert.Equal(expected.LastAppliedSignalOrdinal, result.InitialState.LastAppliedSignalOrdinal);
|
||||
|
||||
var live = new JournalWriter(journalPath, () => rig.Clock.UtcNow).ReadAll();
|
||||
Assert.Equal(2, live.Count);
|
||||
Assert.Equal(1, live[0].StepIndex);
|
||||
Assert.Equal(2, live[1].StepIndex);
|
||||
|
||||
// Nothing was beyond the seed boundary → no phantom quarantine.
|
||||
var quarantineRoot = Path.Combine(rig.StateDir, ".quarantine");
|
||||
if (Directory.Exists(quarantineRoot))
|
||||
{
|
||||
Assert.Empty(Directory.GetFiles(
|
||||
quarantineRoot, "journal-phantom-tail.jsonl", SearchOption.AllDirectories));
|
||||
}
|
||||
}
|
||||
|
||||
// ================================================================= WhiteGlove Part-2 resume
|
||||
|
||||
[Fact]
|
||||
public void WhiteGloveSealed_snapshot_archives_state_and_sets_part2_flag()
|
||||
{
|
||||
using var rig = new Rig();
|
||||
|
||||
var sealedState = DecisionState.CreateInitial(SessionId, TenantId)
|
||||
.ToBuilder()
|
||||
.WithStage(SessionStage.WhiteGloveSealed)
|
||||
.WithStepIndex(5)
|
||||
.Build();
|
||||
SeedSnapshot(rig, sealedState);
|
||||
File.WriteAllText(rig.StatePath("signal-log.jsonl"), "{\"part1-leftover\":true}\n", Encoding.UTF8);
|
||||
File.WriteAllText(rig.StatePath("journal.jsonl"), "{\"part1-journal\":true}\n", Encoding.UTF8);
|
||||
File.WriteAllText(rig.StatePath("event-sequence.json"), "{\"LastAssignedSequence\":17}", Encoding.UTF8);
|
||||
|
||||
var result = rig.Recover();
|
||||
|
||||
Assert.True(result.IsWhiteGlovePart2);
|
||||
Assert.False(result.WasStartupQuarantine);
|
||||
|
||||
// Fresh live state — not the sealed stage.
|
||||
Assert.Equal(0, result.InitialState.StepIndex);
|
||||
Assert.NotEqual(SessionStage.WhiteGloveSealed, result.InitialState.Stage);
|
||||
|
||||
// Reducer-state segments archived to .part1-*; event-sequence intentionally preserved.
|
||||
var buckets = Directory.GetDirectories(rig.StateDir, ".part1-*");
|
||||
Assert.Single(buckets);
|
||||
Assert.True(File.Exists(Path.Combine(buckets[0], "snapshot.json")));
|
||||
Assert.True(File.Exists(Path.Combine(buckets[0], "signal-log.jsonl")));
|
||||
Assert.True(File.Exists(Path.Combine(buckets[0], "journal.jsonl")));
|
||||
Assert.False(File.Exists(Path.Combine(buckets[0], "event-sequence.json")));
|
||||
Assert.Contains("17", File.ReadAllText(rig.StatePath("event-sequence.json")));
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void Non_whiteglove_snapshot_does_not_set_part2_flag_or_archive()
|
||||
{
|
||||
using var rig = new Rig();
|
||||
SeedSnapshot(rig, DecisionState.CreateInitial(SessionId, TenantId)
|
||||
.ToBuilder()
|
||||
.WithStage(SessionStage.EspDeviceSetup)
|
||||
.WithStepIndex(3)
|
||||
.Build());
|
||||
|
||||
var result = rig.Recover();
|
||||
|
||||
Assert.False(result.IsWhiteGlovePart2);
|
||||
Assert.Empty(Directory.GetDirectories(rig.StateDir, ".part1-*"));
|
||||
}
|
||||
|
||||
// ================================================================= Prior-run quarantine marker
|
||||
|
||||
[Fact]
|
||||
public void Prior_run_quarantine_marker_quarantines_all_segments_and_sets_flag()
|
||||
{
|
||||
using var rig = new Rig();
|
||||
|
||||
// Seed a snapshot + log so we can prove they get quarantined.
|
||||
SeedSnapshot(rig, DecisionState.CreateInitial(SessionId, TenantId)
|
||||
.ToBuilder().WithStage(SessionStage.EspDeviceSetup).WithStepIndex(3).Build());
|
||||
SeedSignalLog(rig, MakeSignal(0, DecisionSignalKind.SessionStarted));
|
||||
|
||||
const string reason = "consecutive-journal-append-failures";
|
||||
File.WriteAllText(
|
||||
rig.StatePath(RecoveryCoordinator.QuarantineRequestedMarkerFile),
|
||||
"2026-04-20T09:59:00.0000000Z " + reason,
|
||||
Encoding.UTF8);
|
||||
|
||||
var result = rig.Recover();
|
||||
|
||||
Assert.True(result.PriorRunQuarantined);
|
||||
Assert.NotNull(result.PriorRunQuarantineReason);
|
||||
Assert.Contains(reason, result.PriorRunQuarantineReason!);
|
||||
|
||||
// Fresh Initial state — the discarded snapshot did NOT seed it.
|
||||
Assert.Equal(SessionStage.SessionStarted, result.InitialState.Stage);
|
||||
Assert.Equal(0, result.InitialState.StepIndex);
|
||||
|
||||
// Marker consumed (deleted after a successful quarantine).
|
||||
Assert.False(File.Exists(rig.StatePath(RecoveryCoordinator.QuarantineRequestedMarkerFile)));
|
||||
|
||||
// Old snapshot moved aside.
|
||||
Assert.False(File.Exists(rig.StatePath("snapshot.json")));
|
||||
Assert.True(Directory.Exists(Path.Combine(rig.StateDir, ".quarantine")));
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void No_quarantine_marker_leaves_prior_run_flags_clear()
|
||||
{
|
||||
using var rig = new Rig();
|
||||
|
||||
var result = rig.Recover();
|
||||
|
||||
Assert.False(result.PriorRunQuarantined);
|
||||
Assert.Null(result.PriorRunQuarantineReason);
|
||||
Assert.False(File.Exists(rig.StatePath(RecoveryCoordinator.QuarantineRequestedMarkerFile)));
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -134,20 +134,12 @@ namespace AutopilotMonitor.Agent.V2.Core.Orchestration
|
||||
private bool _quarantineRequested;
|
||||
private string? _quarantineReason;
|
||||
|
||||
// TRACE-H2: marker file written by TriggerQuarantine (mid-run, when the journal can no
|
||||
// longer be appended) and honoured on the NEXT Start — a running process must not pull the
|
||||
// state files out from under itself. When the marker is found at startup, all reducer-state
|
||||
// segments + snapshot are quarantined and the session starts from a clean Initial state.
|
||||
private const string QuarantineRequestedMarkerFile = "quarantine-requested.marker";
|
||||
// TRACE-H2: a prior run that could no longer persist consistently leaves a marker
|
||||
// (see RecoveryCoordinator.QuarantineRequestedMarkerFile) that the NEXT start honours.
|
||||
// These flags capture whether that happened this start, for the step 13b-Q event.
|
||||
private bool _priorRunQuarantined;
|
||||
private string? _priorRunQuarantineReason;
|
||||
|
||||
// P2: set when a prior-run quarantine was REQUESTED (marker present) but did NOT fully
|
||||
// succeed this start. The marker demands a clean reset, so we must NOT fall back to normal
|
||||
// snapshot/log recovery (which could reload the very state we were told to discard) — this
|
||||
// forces the run to seed from a fresh Initial state instead.
|
||||
private bool _forceCleanStart;
|
||||
|
||||
// Recovery flags (populated during Start()).
|
||||
// V1-symmetric Part-2 hint (analog to V1 MonitoringService._isWhiteGlovePart2):
|
||||
// set when Start() detects a persisted WhiteGloveSealed snapshot and archives the
|
||||
@@ -334,251 +326,23 @@ namespace AutopilotMonitor.Agent.V2.Core.Orchestration
|
||||
|
||||
EnsureDirectories();
|
||||
|
||||
var snapshotPath = Path.Combine(_stateDirectory, "snapshot.json");
|
||||
var signalLogPath = Path.Combine(_stateDirectory, "signal-log.jsonl");
|
||||
var journalPath = Path.Combine(_stateDirectory, "journal.jsonl");
|
||||
var eventSequencePath = Path.Combine(_stateDirectory, "event-sequence.json");
|
||||
|
||||
// -1) TRACE-H2 — honour a prior run's quarantine request BEFORE the writers scan the
|
||||
// (suspect) segment files. A prior run hit the consecutive-journal-failure threshold
|
||||
// and could no longer persist consistently; it left the marker so THIS run resets
|
||||
// cleanly. Quarantining here (not mid-run) preserves the "a running process never
|
||||
// pulls its own state files out from under itself" invariant. event-sequence is
|
||||
// quarantined too — a clean restart re-seeds it from 0 (backend dedups on RowKey).
|
||||
var quarantineMarkerPath = Path.Combine(_stateDirectory, QuarantineRequestedMarkerFile);
|
||||
if (File.Exists(quarantineMarkerPath))
|
||||
{
|
||||
var quarantineReason = TryReadMarkerReason(quarantineMarkerPath);
|
||||
_logger.Error(
|
||||
"EnrollmentOrchestrator: prior-run quarantine marker found — quarantining all state " +
|
||||
$"segments + snapshot before recovery. Reason: {quarantineReason}");
|
||||
try
|
||||
{
|
||||
SegmentQuarantine.QuarantineAll(
|
||||
_stateDirectory, "quarantine-requested:" + quarantineReason, () => _clock.UtcNow);
|
||||
new SnapshotPersistence(snapshotPath, () => _clock.UtcNow)
|
||||
.Quarantine("quarantine-requested:" + quarantineReason);
|
||||
|
||||
// P2: delete the marker ONLY after the quarantine fully succeeded. If QuarantineAll
|
||||
// or the snapshot quarantine threw (ACL / lock / disk), the marker is LEFT so the
|
||||
// next start retries fail-closed instead of proceeding on possibly-still-corrupt
|
||||
// segments with the retry hint lost.
|
||||
try { File.Delete(quarantineMarkerPath); }
|
||||
catch (Exception ex) { _logger.Warning($"EnrollmentOrchestrator: quarantine marker delete failed: {ex.Message}"); }
|
||||
|
||||
_priorRunQuarantined = true;
|
||||
_priorRunQuarantineReason = quarantineReason;
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
// Marker intentionally retained — do NOT delete on failure (P2). AND force a
|
||||
// clean Initial seed for THIS run: a partial failure (e.g. segments moved but the
|
||||
// snapshot quarantine threw) must not let normal recovery reload the discarded
|
||||
// snapshot. Fail-closed to fresh state; the retained marker drives a proper retry
|
||||
// next start.
|
||||
_forceCleanStart = true;
|
||||
_logger.Error(
|
||||
"EnrollmentOrchestrator: prior-run quarantine cleanup FAILED — marker retained for "
|
||||
+ "next-start retry; forcing a clean Initial seed for this run.", ex);
|
||||
}
|
||||
}
|
||||
|
||||
// 0) WhiteGlove Part-2 resume detection. V1-symmetric Archive-and-Reset
|
||||
// pattern: peek the persisted snapshot via the lock-free static reader; if
|
||||
// the prior run sealed Part-1, move the reducer-state segment files
|
||||
// (snapshot, signal-log, journal) aside to a timestamped
|
||||
// <c>.part1-<ts>/</c> bucket BEFORE the writers below open them for
|
||||
// scanning. <c>event-sequence.json</c> is intentionally preserved so the
|
||||
// session-wide event sequence stays monotonic across the resume boundary
|
||||
// (the backend orders events by <c>(SessionId, Sequence)</c> and the
|
||||
// Web UI splits on <c>resumed.sequence - 1</c>). The orchestrator then
|
||||
// runs through the normal recovery branches as a fresh first-boot
|
||||
// (branch c1), re-emits a <c>whiteglove_resumed</c> lifecycle event
|
||||
// after the onIngressReady hook, and lets the standard Classic
|
||||
// enrollment flow drive completion.
|
||||
// Anti-loop: once archived, the WhiteGloveSealed snapshot is gone, so a
|
||||
// subsequent restart sees no marker and continues as a Classic session.
|
||||
// The hint bool is in-memory only — also matches V1 (which clears the
|
||||
// on-disk marker file before emitting <c>whiteglove_resumed</c>).
|
||||
if (Directory.Exists(_stateDirectory))
|
||||
{
|
||||
var rawSnapshot = SnapshotPersistence.TryReadRaw(snapshotPath);
|
||||
if (rawSnapshot != null && rawSnapshot.Stage == SessionStage.WhiteGloveSealed)
|
||||
{
|
||||
StateArchiver.ArchiveStateFolder(
|
||||
stateDirectory: _stateDirectory,
|
||||
reason: "wg_part1_resume_archive",
|
||||
utcNow: () => _clock.UtcNow,
|
||||
logger: _logger);
|
||||
_isWhiteGlovePart2 = true;
|
||||
_logger.Info(
|
||||
"EnrollmentOrchestrator: WhiteGlove Part-1 resume detected — state archived; " +
|
||||
"starting fresh Classic enrollment flow for Part-2.");
|
||||
}
|
||||
}
|
||||
|
||||
// 1) Persistenz. Writer scannen bestehende Files im Ctor (LastOrdinal / LastStepIndex).
|
||||
_signalLog = new SignalLogWriter(signalLogPath);
|
||||
_journal = new JournalWriter(journalPath, () => _clock.UtcNow);
|
||||
_snapshot = new SnapshotPersistence(snapshotPath, () => _clock.UtcNow);
|
||||
_eventSequencePersistence = new EventSequencePersistence(eventSequencePath);
|
||||
|
||||
// 2) Recovery (Plan §2.7 Sonderfälle 1+2, Codex follow-up #1).
|
||||
//
|
||||
// Invariant: the SignalLog is authoritative truth, the Snapshot is a cache.
|
||||
// Therefore on every recovery we prefer reconstructing state from the log over
|
||||
// trusting a stale/corrupt snapshot. ReducerReplay.Replay (Phase 1) performs the
|
||||
// pure fold; this method decides the seed + which signals to feed it.
|
||||
//
|
||||
// Branches:
|
||||
// a) Snapshot present but corrupt → quarantine the snapshot ONLY (not the log!)
|
||||
// and rebuild from the full SignalLog. If the log itself is head-corrupt
|
||||
// (file non-empty but ReadAll yields zero signals) we escalate to a total-
|
||||
// loss quarantine of the log segments and start fresh.
|
||||
// b) Snapshot loaded cleanly → replay any SignalLog tail past the snapshot's
|
||||
// LastAppliedSignalOrdinal so a pre-crash Journal/Snapshot lag is closed.
|
||||
// c) No snapshot 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 + replay-stream are selected the Journal is aligned AHEAD-OF
|
||||
// replay (phantom suffix beyond seed.StepIndex-1 goes to the forensic bucket)
|
||||
// and backfilled DURING replay via the onTransition callback. This closes two
|
||||
// crash windows in one mechanism (Codex follow-up post-#50 #C):
|
||||
// • AHEAD — Journal was flushed but Snapshot not; phantom transitions beyond
|
||||
// the seed get quarantined before replay so live Append monotonicity holds.
|
||||
// • BEHIND — SignalLog was flushed but Journal not; the replay callback
|
||||
// rematerialises the missing StepIndex entries onto the Journal so it
|
||||
// ends up in exact lockstep with initialState.
|
||||
var snapshotFileExistsPreLoad = File.Exists(snapshotPath);
|
||||
var loadedState = _snapshot.Load();
|
||||
DecisionState initialState;
|
||||
|
||||
// Reducer is stateless — a transient instance is enough for the pure replay below.
|
||||
// The live ingress pipeline creates its own instance further down the Start() body.
|
||||
var replayEngine = new DecisionEngine();
|
||||
|
||||
// Dispatch: pick the seed + signal stream per branch. Null seed is impossible past
|
||||
// this block (every path assigns it), but the compiler needs the explicit init.
|
||||
// AgentBootUtc gets stamped from the live clock so reducer deadline-arming sites
|
||||
// can floor replayed-signal timestamps at the current run's start (replay-safety).
|
||||
var agentBootUtc = _clock.UtcNow;
|
||||
DecisionState seed = DecisionState.CreateInitial(_sessionId, _tenantId, agentBootUtc);
|
||||
IReadOnlyList<DecisionSignal> signalsToReplay = Array.Empty<DecisionSignal>();
|
||||
string branchTag;
|
||||
|
||||
if (_forceCleanStart)
|
||||
{
|
||||
// P2 — a requested prior-run quarantine did not fully succeed. Fail-closed: ignore
|
||||
// any (possibly-still-present) snapshot and SignalLog and seed from a fresh Initial
|
||||
// state. seed/signalsToReplay are already CreateInitial/empty above. The retained
|
||||
// marker drives a proper full quarantine + retry on the next start.
|
||||
_logger.Warning(
|
||||
"EnrollmentOrchestrator: forcing clean Initial seed (prior-run quarantine incomplete) — "
|
||||
+ "skipping snapshot/log recovery for this run.");
|
||||
branchTag = "forced-clean-after-failed-quarantine";
|
||||
}
|
||||
else if (loadedState == null && snapshotFileExistsPreLoad)
|
||||
{
|
||||
// Branch (a) — Snapshot corrupt. Quarantine the snapshot and attempt log replay.
|
||||
_logger.Error(
|
||||
"EnrollmentOrchestrator: snapshot present but Load returned null (checksum mismatch or parse error) — quarantining snapshot and attempting SignalLog replay.");
|
||||
_snapshot.Quarantine("checksum-mismatch-on-startup");
|
||||
|
||||
var loggedSignals = _signalLog.ReadAll();
|
||||
var logFileInfo = new FileInfo(signalLogPath);
|
||||
var logIsHeadCorrupt =
|
||||
logFileInfo.Exists && logFileInfo.Length > 0 && loggedSignals.Count == 0;
|
||||
|
||||
if (logIsHeadCorrupt)
|
||||
{
|
||||
_logger.Warning(
|
||||
"EnrollmentOrchestrator: SignalLog unreadable from the first line — escalating to total-loss quarantine.");
|
||||
SegmentQuarantine.QuarantineAll(
|
||||
_stateDirectory, "log-head-corrupt-after-snapshot-loss", () => _clock.UtcNow);
|
||||
|
||||
// Writer hold paths, not handles — but their in-memory counters are stale
|
||||
// after the quarantine move. Recreate to reset them to -1.
|
||||
_signalLog = new SignalLogWriter(signalLogPath);
|
||||
_journal = new JournalWriter(journalPath, () => _clock.UtcNow);
|
||||
_eventSequencePersistence = new EventSequencePersistence(eventSequencePath);
|
||||
|
||||
branchTag = "a-total-loss";
|
||||
}
|
||||
else
|
||||
{
|
||||
signalsToReplay = loggedSignals;
|
||||
branchTag = "a-full-log-replay";
|
||||
}
|
||||
|
||||
_wasStartupQuarantine = true;
|
||||
}
|
||||
else if (loadedState != null)
|
||||
{
|
||||
// Branch (b) — Snapshot loaded. Catch up any SignalLog tail past the snapshot.
|
||||
// Re-stamp AgentBootUtc so the current run's deadlines floor at "now" — the
|
||||
// persisted boot anchor is from the prior run and using it would let replayed
|
||||
// tail signals arm deadlines that are already past-due (premature fire).
|
||||
seed = loadedState.ToBuilder().WithAgentBootUtc(agentBootUtc).Build();
|
||||
signalsToReplay = CollectSignalLogTailAfter(loadedState.LastAppliedSignalOrdinal);
|
||||
branchTag = signalsToReplay.Count > 0 ? "b-tail-replay" : "b-snapshot-current";
|
||||
}
|
||||
else
|
||||
{
|
||||
// Branch (c) — no snapshot on disk. Two sub-cases:
|
||||
// (c1) SignalLog also empty → genuinely fresh start.
|
||||
// (c2) SignalLog has entries (crash BEFORE first snapshot save) → full
|
||||
// log replay from CreateInitial (Codex follow-up post-#50 #A).
|
||||
var loggedSignals = _signalLog.ReadAll();
|
||||
if (loggedSignals.Count == 0)
|
||||
{
|
||||
branchTag = "c1-fresh";
|
||||
}
|
||||
else
|
||||
{
|
||||
signalsToReplay = loggedSignals;
|
||||
_wasStartupQuarantine = true;
|
||||
branchTag = "c2-full-log-replay";
|
||||
}
|
||||
}
|
||||
|
||||
// Align Journal AHEAD phantoms before replay. Engine semantics: a transition
|
||||
// produced for state at StepIndex=K carries StepIndex=K+1 (see
|
||||
// DecisionEngine.BuildTakenTransition); after K reduces from initial the journal
|
||||
// holds entries [1..K] with LastStepIndex=K. So any on-disk entry beyond
|
||||
// seed.StepIndex is a phantom from a crash between Journal.Append and
|
||||
// Snapshot.Save. Truncating first lets the replay callback Append monotonically
|
||||
// from seed.StepIndex+1 without colliding.
|
||||
var journalBoundary = seed.StepIndex;
|
||||
if (_journal.LastStepIndex > journalBoundary)
|
||||
{
|
||||
_logger.Warning(
|
||||
$"EnrollmentOrchestrator: Journal ahead of seed state " +
|
||||
$"(journal.LastStepIndex={_journal.LastStepIndex}, seed.StepIndex={seed.StepIndex}) — " +
|
||||
$"truncating phantom transitions to boundary={journalBoundary}.");
|
||||
_journal.TruncateAfter(journalBoundary);
|
||||
}
|
||||
|
||||
// Replay + Journal backfill via the onTransition callback (Codex follow-up
|
||||
// post-#50 #C). For BEHIND crashes (SignalLog flushed, Journal not) this
|
||||
// rematerialises every missing StepIndex. For no-replay branches the callback
|
||||
// is simply never invoked; the journal stays aligned from the pre-replay
|
||||
// truncate step above.
|
||||
initialState = ReducerReplay.Replay(
|
||||
engine: replayEngine,
|
||||
seed: seed,
|
||||
signals: signalsToReplay,
|
||||
onTransition: _journal.Append);
|
||||
|
||||
_logger.Info(
|
||||
$"EnrollmentOrchestrator: recovery branch={branchTag}, " +
|
||||
$"stage={initialState.Stage}, stepIndex={initialState.StepIndex}, " +
|
||||
$"signalsReplayed={signalsToReplay.Count}, " +
|
||||
$"journal.LastStepIndex={_journal.LastStepIndex}.");
|
||||
// 1+2) Crash-recovery state machine (Plan §2.7). Extracted to RecoveryCoordinator
|
||||
// (ARCH-F5): it honours a prior-run quarantine marker, performs the WhiteGlove
|
||||
// Part-2 archive-and-reset, opens the persistence writers against the resulting
|
||||
// clean directory, picks the recovery branch + replay stream, aligns the journal
|
||||
// AHEAD-phantom boundary, and folds to the seed state. The writers + flags it
|
||||
// returns drive the wiring steps below; field names (_signalLog, _journal, …)
|
||||
// are preserved for the recovery tests that reflect on them.
|
||||
var recovery = RecoveryCoordinator.Recover(_stateDirectory, _sessionId, _tenantId, _clock, _logger);
|
||||
_signalLog = recovery.SignalLog;
|
||||
_journal = recovery.Journal;
|
||||
_snapshot = recovery.Snapshot;
|
||||
_eventSequencePersistence = recovery.EventSequence;
|
||||
_isWhiteGlovePart2 = recovery.IsWhiteGlovePart2;
|
||||
_wasStartupQuarantine = recovery.WasStartupQuarantine;
|
||||
_priorRunQuarantined = recovery.PriorRunQuarantined;
|
||||
_priorRunQuarantineReason = recovery.PriorRunQuarantineReason;
|
||||
var initialState = recovery.InitialState;
|
||||
|
||||
// 3) Telemetry-Transport. Batch size flows from AgentConfiguration.MaxBatchSize
|
||||
// via Program.cs (P1 fix: previously the remote-config knob was merged but
|
||||
@@ -1040,26 +804,6 @@ namespace AutopilotMonitor.Agent.V2.Core.Orchestration
|
||||
catch (Exception ex) { _logger.Error("EnrollmentOrchestrator: Terminated handler threw.", ex); }
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// <summary>
|
||||
/// Recovery helper — collect all SignalLog entries whose <c>SessionSignalOrdinal</c>
|
||||
/// is strictly greater than the snapshot's last-consumed ordinal. Signals equal-or-less
|
||||
/// are already baked into the seed state and must never be re-applied.
|
||||
/// </summary>
|
||||
private IReadOnlyList<DecisionSignal> CollectSignalLogTailAfter(long lastConsumedOrdinal)
|
||||
{
|
||||
var all = _signalLog!.ReadAll();
|
||||
var tail = new List<DecisionSignal>();
|
||||
for (int i = 0; i < all.Count; i++)
|
||||
{
|
||||
if (all[i].SessionSignalOrdinal > lastConsumedOrdinal)
|
||||
{
|
||||
tail.Add(all[i]);
|
||||
}
|
||||
}
|
||||
return tail;
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// M4.6.β — DecisionStepProcessor callback: the engine has transitioned the session
|
||||
/// into a terminal <see cref="SessionStage"/>. Fires the public <see cref="Terminated"/>
|
||||
@@ -1256,7 +1000,7 @@ namespace AutopilotMonitor.Agent.V2.Core.Orchestration
|
||||
// log-only behaviour (the BootTrigger / max-lifetime restart still re-runs recovery).
|
||||
try
|
||||
{
|
||||
var markerPath = Path.Combine(_stateDirectory, QuarantineRequestedMarkerFile);
|
||||
var markerPath = Path.Combine(_stateDirectory, RecoveryCoordinator.QuarantineRequestedMarkerFile);
|
||||
var body = _clock.UtcNow.ToString("o", System.Globalization.CultureInfo.InvariantCulture)
|
||||
+ " " + (reason ?? string.Empty);
|
||||
File.WriteAllText(markerPath, body, System.Text.Encoding.UTF8);
|
||||
@@ -1267,13 +1011,6 @@ namespace AutopilotMonitor.Agent.V2.Core.Orchestration
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>Reads the reason text from a quarantine marker; returns a fallback on any error.</summary>
|
||||
private static string TryReadMarkerReason(string markerPath)
|
||||
{
|
||||
try { return File.ReadAllText(markerPath, System.Text.Encoding.UTF8); }
|
||||
catch { return "unknown (marker unreadable)"; }
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// TRACE-H1/P1 — turns a quarantined oversized item into a <c>telemetry_upload_poisoned</c>
|
||||
/// Warning so the drop is visible on the backend timeline (the item stays in spool.jsonl for
|
||||
|
||||
@@ -0,0 +1,419 @@
|
||||
#nullable enable
|
||||
using System;
|
||||
using System.Collections.Generic;
|
||||
using System.IO;
|
||||
using AutopilotMonitor.Agent.V2.Core.Logging;
|
||||
using AutopilotMonitor.Agent.V2.Core.Persistence;
|
||||
using AutopilotMonitor.Agent.V2.Core.Telemetry.Events;
|
||||
using AutopilotMonitor.DecisionCore.Engine;
|
||||
using AutopilotMonitor.DecisionCore.Signals;
|
||||
using AutopilotMonitor.DecisionCore.State;
|
||||
|
||||
namespace AutopilotMonitor.Agent.V2.Core.Orchestration
|
||||
{
|
||||
/// <summary>
|
||||
/// Immutable outcome of <see cref="RecoveryCoordinator.Recover"/>: the freshly-opened
|
||||
/// persistence writers, the reconstructed seed <see cref="DecisionState"/>, and the
|
||||
/// recovery flags the orchestrator surfaces for observability and downstream wiring.
|
||||
/// </summary>
|
||||
internal sealed class RecoveryResult
|
||||
{
|
||||
public RecoveryResult(
|
||||
SignalLogWriter signalLog,
|
||||
JournalWriter journal,
|
||||
SnapshotPersistence snapshot,
|
||||
EventSequencePersistence eventSequence,
|
||||
DecisionState initialState,
|
||||
bool isWhiteGlovePart2,
|
||||
bool wasStartupQuarantine,
|
||||
bool priorRunQuarantined,
|
||||
string? priorRunQuarantineReason)
|
||||
{
|
||||
SignalLog = signalLog;
|
||||
Journal = journal;
|
||||
Snapshot = snapshot;
|
||||
EventSequence = eventSequence;
|
||||
InitialState = initialState;
|
||||
IsWhiteGlovePart2 = isWhiteGlovePart2;
|
||||
WasStartupQuarantine = wasStartupQuarantine;
|
||||
PriorRunQuarantined = priorRunQuarantined;
|
||||
PriorRunQuarantineReason = priorRunQuarantineReason;
|
||||
}
|
||||
|
||||
/// <summary>SignalLog writer opened against the (post-quarantine, post-archive) state dir.</summary>
|
||||
public SignalLogWriter SignalLog { get; }
|
||||
|
||||
/// <summary>Journal writer, aligned in lockstep with <see cref="InitialState"/>.</summary>
|
||||
public JournalWriter Journal { get; }
|
||||
|
||||
/// <summary>Snapshot persistence handle for the live pipeline's save path.</summary>
|
||||
public SnapshotPersistence Snapshot { get; }
|
||||
|
||||
/// <summary>Event-sequence persistence (intentionally never archived across WG resume).</summary>
|
||||
public EventSequencePersistence EventSequence { get; }
|
||||
|
||||
/// <summary>The seed state to start the live decision pipeline from.</summary>
|
||||
public DecisionState InitialState { get; }
|
||||
|
||||
/// <summary>
|
||||
/// <c>true</c> when a persisted <see cref="SessionStage.WhiteGloveSealed"/> snapshot was
|
||||
/// detected and the prior state folder archived to <c>.part1-<ts>/</c>.
|
||||
/// </summary>
|
||||
public bool IsWhiteGlovePart2 { get; }
|
||||
|
||||
/// <summary>
|
||||
/// <c>true</c> when a corrupt snapshot / head-corrupt or orphaned SignalLog forced a
|
||||
/// startup quarantine of state segments.
|
||||
/// </summary>
|
||||
public bool WasStartupQuarantine { get; }
|
||||
|
||||
/// <summary><c>true</c> when a prior-run quarantine marker was honoured this start.</summary>
|
||||
public bool PriorRunQuarantined { get; }
|
||||
|
||||
/// <summary>Reason text from the prior-run quarantine marker, or <c>null</c>.</summary>
|
||||
public string? PriorRunQuarantineReason { get; }
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Owns the V2 agent's crash-recovery state machine (Plan §2.7). Extracted from
|
||||
/// <see cref="EnrollmentOrchestrator.Start"/> (ARCH-F5) so the recovery decision logic is
|
||||
/// isolated from the orchestrator's component-wiring steps.
|
||||
/// <para>
|
||||
/// <b>Order is load-bearing</b> and mirrors the original Start() body exactly:
|
||||
/// <list type="number">
|
||||
/// <item><b>-1) Prior-run quarantine marker</b> — honour a marker left by a prior run that
|
||||
/// could no longer persist consistently, quarantining all state segments + snapshot BEFORE
|
||||
/// the writers scan the suspect files. Preserves the "a running process never pulls its own
|
||||
/// state files out from under itself" invariant.</item>
|
||||
/// <item><b>0) WhiteGlove Part-2 resume</b> — when the persisted snapshot is
|
||||
/// <see cref="SessionStage.WhiteGloveSealed"/>, archive the reducer-state segments aside and
|
||||
/// begin a fresh Classic flow. <c>event-sequence.json</c> is intentionally preserved.</item>
|
||||
/// <item><b>1) Persistence writers</b> — opened after -1/0 so they scan a clean directory.</item>
|
||||
/// <item><b>2) Recovery branches</b> — pick the seed + replay stream (forced-clean / corrupt
|
||||
/// snapshot / clean snapshot tail-replay / no-snapshot), align the journal AHEAD-phantom
|
||||
/// boundary, then fold via <see cref="ReducerReplay.Replay"/> with journal backfill.</item>
|
||||
/// </list>
|
||||
/// </para>
|
||||
/// <para>
|
||||
/// The invariant throughout: the SignalLog is authoritative truth, the Snapshot is a cache —
|
||||
/// every recovery prefers reconstructing from the log over trusting a stale/corrupt snapshot.
|
||||
/// </para>
|
||||
/// </summary>
|
||||
internal static class RecoveryCoordinator
|
||||
{
|
||||
// TRACE-H2: marker file written by EnrollmentOrchestrator.TriggerQuarantine (mid-run, when
|
||||
// the journal can no longer be appended) and honoured on the NEXT Recover — a running
|
||||
// process must not pull the state files out from under itself. When the marker is found at
|
||||
// startup, all reducer-state segments + snapshot are quarantined and the session starts
|
||||
// from a clean Initial state.
|
||||
public const string QuarantineRequestedMarkerFile = "quarantine-requested.marker";
|
||||
|
||||
/// <summary>
|
||||
/// Runs the full recovery pipeline against <paramref name="stateDirectory"/> and returns
|
||||
/// the opened writers, seed state, and recovery flags. The directory is assumed to exist
|
||||
/// (the orchestrator calls <c>EnsureDirectories</c> first).
|
||||
/// </summary>
|
||||
public static RecoveryResult Recover(
|
||||
string stateDirectory,
|
||||
string sessionId,
|
||||
string tenantId,
|
||||
IClock clock,
|
||||
AgentLogger logger)
|
||||
{
|
||||
var snapshotPath = Path.Combine(stateDirectory, "snapshot.json");
|
||||
var signalLogPath = Path.Combine(stateDirectory, "signal-log.jsonl");
|
||||
var journalPath = Path.Combine(stateDirectory, "journal.jsonl");
|
||||
var eventSequencePath = Path.Combine(stateDirectory, "event-sequence.json");
|
||||
|
||||
var isWhiteGlovePart2 = false;
|
||||
var wasStartupQuarantine = false;
|
||||
var priorRunQuarantined = false;
|
||||
string? priorRunQuarantineReason = null;
|
||||
|
||||
// P2: set when a prior-run quarantine was REQUESTED (marker present) but did NOT fully
|
||||
// succeed this start. The marker demands a clean reset, so we must NOT fall back to normal
|
||||
// snapshot/log recovery (which could reload the very state we were told to discard) — this
|
||||
// forces the run to seed from a fresh Initial state instead.
|
||||
var forceCleanStart = false;
|
||||
|
||||
// -1) TRACE-H2 — honour a prior run's quarantine request BEFORE the writers scan the
|
||||
// (suspect) segment files. A prior run hit the consecutive-journal-failure threshold
|
||||
// and could no longer persist consistently; it left the marker so THIS run resets
|
||||
// cleanly. Quarantining here (not mid-run) preserves the "a running process never
|
||||
// pulls its own state files out from under itself" invariant. event-sequence is
|
||||
// quarantined too — a clean restart re-seeds it from 0 (backend dedups on RowKey).
|
||||
var quarantineMarkerPath = Path.Combine(stateDirectory, QuarantineRequestedMarkerFile);
|
||||
if (File.Exists(quarantineMarkerPath))
|
||||
{
|
||||
var quarantineReason = TryReadMarkerReason(quarantineMarkerPath);
|
||||
logger.Error(
|
||||
"EnrollmentOrchestrator: prior-run quarantine marker found — quarantining all state " +
|
||||
$"segments + snapshot before recovery. Reason: {quarantineReason}");
|
||||
try
|
||||
{
|
||||
SegmentQuarantine.QuarantineAll(
|
||||
stateDirectory, "quarantine-requested:" + quarantineReason, () => clock.UtcNow);
|
||||
new SnapshotPersistence(snapshotPath, () => clock.UtcNow)
|
||||
.Quarantine("quarantine-requested:" + quarantineReason);
|
||||
|
||||
// P2: delete the marker ONLY after the quarantine fully succeeded. If QuarantineAll
|
||||
// or the snapshot quarantine threw (ACL / lock / disk), the marker is LEFT so the
|
||||
// next start retries fail-closed instead of proceeding on possibly-still-corrupt
|
||||
// segments with the retry hint lost.
|
||||
try { File.Delete(quarantineMarkerPath); }
|
||||
catch (Exception ex) { logger.Warning($"EnrollmentOrchestrator: quarantine marker delete failed: {ex.Message}"); }
|
||||
|
||||
priorRunQuarantined = true;
|
||||
priorRunQuarantineReason = quarantineReason;
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
// Marker intentionally retained — do NOT delete on failure (P2). AND force a
|
||||
// clean Initial seed for THIS run: a partial failure (e.g. segments moved but the
|
||||
// snapshot quarantine threw) must not let normal recovery reload the discarded
|
||||
// snapshot. Fail-closed to fresh state; the retained marker drives a proper retry
|
||||
// next start.
|
||||
forceCleanStart = true;
|
||||
logger.Error(
|
||||
"EnrollmentOrchestrator: prior-run quarantine cleanup FAILED — marker retained for "
|
||||
+ "next-start retry; forcing a clean Initial seed for this run.", ex);
|
||||
}
|
||||
}
|
||||
|
||||
// 0) WhiteGlove Part-2 resume detection. V1-symmetric Archive-and-Reset
|
||||
// pattern: peek the persisted snapshot via the lock-free static reader; if
|
||||
// the prior run sealed Part-1, move the reducer-state segment files
|
||||
// (snapshot, signal-log, journal) aside to a timestamped
|
||||
// <c>.part1-<ts>/</c> bucket BEFORE the writers below open them for
|
||||
// scanning. <c>event-sequence.json</c> is intentionally preserved so the
|
||||
// session-wide event sequence stays monotonic across the resume boundary
|
||||
// (the backend orders events by <c>(SessionId, Sequence)</c> and the
|
||||
// Web UI splits on <c>resumed.sequence - 1</c>). The orchestrator then
|
||||
// runs through the normal recovery branches as a fresh first-boot
|
||||
// (branch c1), re-emits a <c>whiteglove_resumed</c> lifecycle event
|
||||
// after the onIngressReady hook, and lets the standard Classic
|
||||
// enrollment flow drive completion.
|
||||
// Anti-loop: once archived, the WhiteGloveSealed snapshot is gone, so a
|
||||
// subsequent restart sees no marker and continues as a Classic session.
|
||||
// The hint bool is in-memory only — also matches V1 (which clears the
|
||||
// on-disk marker file before emitting <c>whiteglove_resumed</c>).
|
||||
if (Directory.Exists(stateDirectory))
|
||||
{
|
||||
var rawSnapshot = SnapshotPersistence.TryReadRaw(snapshotPath);
|
||||
if (rawSnapshot != null && rawSnapshot.Stage == SessionStage.WhiteGloveSealed)
|
||||
{
|
||||
StateArchiver.ArchiveStateFolder(
|
||||
stateDirectory: stateDirectory,
|
||||
reason: "wg_part1_resume_archive",
|
||||
utcNow: () => clock.UtcNow,
|
||||
logger: logger);
|
||||
isWhiteGlovePart2 = true;
|
||||
logger.Info(
|
||||
"EnrollmentOrchestrator: WhiteGlove Part-1 resume detected — state archived; " +
|
||||
"starting fresh Classic enrollment flow for Part-2.");
|
||||
}
|
||||
}
|
||||
|
||||
// 1) Persistenz. Writer scannen bestehende Files im Ctor (LastOrdinal / LastStepIndex).
|
||||
var signalLog = new SignalLogWriter(signalLogPath);
|
||||
var journal = new JournalWriter(journalPath, () => clock.UtcNow);
|
||||
var snapshot = new SnapshotPersistence(snapshotPath, () => clock.UtcNow);
|
||||
var eventSequence = new EventSequencePersistence(eventSequencePath);
|
||||
|
||||
// 2) Recovery (Plan §2.7 Sonderfälle 1+2, Codex follow-up #1).
|
||||
//
|
||||
// Invariant: the SignalLog is authoritative truth, the Snapshot is a cache.
|
||||
// Therefore on every recovery we prefer reconstructing state from the log over
|
||||
// trusting a stale/corrupt snapshot. ReducerReplay.Replay (Phase 1) performs the
|
||||
// pure fold; this method decides the seed + which signals to feed it.
|
||||
//
|
||||
// Branches:
|
||||
// a) Snapshot present but corrupt → quarantine the snapshot ONLY (not the log!)
|
||||
// and rebuild from the full SignalLog. If the log itself is head-corrupt
|
||||
// (file non-empty but ReadAll yields zero signals) we escalate to a total-
|
||||
// loss quarantine of the log segments and start fresh.
|
||||
// b) Snapshot loaded cleanly → replay any SignalLog tail past the snapshot's
|
||||
// LastAppliedSignalOrdinal so a pre-crash Journal/Snapshot lag is closed.
|
||||
// c) No snapshot 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 + replay-stream are selected the Journal is aligned AHEAD-OF
|
||||
// replay (phantom suffix beyond seed.StepIndex-1 goes to the forensic bucket)
|
||||
// and backfilled DURING replay via the onTransition callback. This closes two
|
||||
// crash windows in one mechanism (Codex follow-up post-#50 #C):
|
||||
// • AHEAD — Journal was flushed but Snapshot not; phantom transitions beyond
|
||||
// the seed get quarantined before replay so live Append monotonicity holds.
|
||||
// • BEHIND — SignalLog was flushed but Journal not; the replay callback
|
||||
// rematerialises the missing StepIndex entries onto the Journal so it
|
||||
// ends up in exact lockstep with initialState.
|
||||
var snapshotFileExistsPreLoad = File.Exists(snapshotPath);
|
||||
var loadedState = snapshot.Load();
|
||||
DecisionState initialState;
|
||||
|
||||
// Reducer is stateless — a transient instance is enough for the pure replay below.
|
||||
// The live ingress pipeline creates its own instance further down the Start() body.
|
||||
var replayEngine = new DecisionEngine();
|
||||
|
||||
// Dispatch: pick the seed + signal stream per branch. Null seed is impossible past
|
||||
// this block (every path assigns it), but the compiler needs the explicit init.
|
||||
// AgentBootUtc gets stamped from the live clock so reducer deadline-arming sites
|
||||
// can floor replayed-signal timestamps at the current run's start (replay-safety).
|
||||
var agentBootUtc = clock.UtcNow;
|
||||
DecisionState seed = DecisionState.CreateInitial(sessionId, tenantId, agentBootUtc);
|
||||
IReadOnlyList<DecisionSignal> signalsToReplay = Array.Empty<DecisionSignal>();
|
||||
string branchTag;
|
||||
|
||||
if (forceCleanStart)
|
||||
{
|
||||
// P2 — a requested prior-run quarantine did not fully succeed. Fail-closed: ignore
|
||||
// any (possibly-still-present) snapshot and SignalLog and seed from a fresh Initial
|
||||
// state. seed/signalsToReplay are already CreateInitial/empty above. The retained
|
||||
// marker drives a proper full quarantine + retry on the next start.
|
||||
logger.Warning(
|
||||
"EnrollmentOrchestrator: forcing clean Initial seed (prior-run quarantine incomplete) — "
|
||||
+ "skipping snapshot/log recovery for this run.");
|
||||
branchTag = "forced-clean-after-failed-quarantine";
|
||||
}
|
||||
else if (loadedState == null && snapshotFileExistsPreLoad)
|
||||
{
|
||||
// Branch (a) — Snapshot corrupt. Quarantine the snapshot and attempt log replay.
|
||||
logger.Error(
|
||||
"EnrollmentOrchestrator: snapshot present but Load returned null (checksum mismatch or parse error) — quarantining snapshot and attempting SignalLog replay.");
|
||||
snapshot.Quarantine("checksum-mismatch-on-startup");
|
||||
|
||||
var loggedSignals = signalLog.ReadAll();
|
||||
var logFileInfo = new FileInfo(signalLogPath);
|
||||
var logIsHeadCorrupt =
|
||||
logFileInfo.Exists && logFileInfo.Length > 0 && loggedSignals.Count == 0;
|
||||
|
||||
if (logIsHeadCorrupt)
|
||||
{
|
||||
logger.Warning(
|
||||
"EnrollmentOrchestrator: SignalLog unreadable from the first line — escalating to total-loss quarantine.");
|
||||
SegmentQuarantine.QuarantineAll(
|
||||
stateDirectory, "log-head-corrupt-after-snapshot-loss", () => clock.UtcNow);
|
||||
|
||||
// Writer hold paths, not handles — but their in-memory counters are stale
|
||||
// after the quarantine move. Recreate to reset them to -1.
|
||||
signalLog = new SignalLogWriter(signalLogPath);
|
||||
journal = new JournalWriter(journalPath, () => clock.UtcNow);
|
||||
eventSequence = new EventSequencePersistence(eventSequencePath);
|
||||
|
||||
branchTag = "a-total-loss";
|
||||
}
|
||||
else
|
||||
{
|
||||
signalsToReplay = loggedSignals;
|
||||
branchTag = "a-full-log-replay";
|
||||
}
|
||||
|
||||
wasStartupQuarantine = true;
|
||||
}
|
||||
else if (loadedState != null)
|
||||
{
|
||||
// Branch (b) — Snapshot loaded. Catch up any SignalLog tail past the snapshot.
|
||||
// Re-stamp AgentBootUtc so the current run's deadlines floor at "now" — the
|
||||
// persisted boot anchor is from the prior run and using it would let replayed
|
||||
// tail signals arm deadlines that are already past-due (premature fire).
|
||||
seed = loadedState.ToBuilder().WithAgentBootUtc(agentBootUtc).Build();
|
||||
signalsToReplay = CollectSignalLogTailAfter(signalLog, loadedState.LastAppliedSignalOrdinal);
|
||||
branchTag = signalsToReplay.Count > 0 ? "b-tail-replay" : "b-snapshot-current";
|
||||
}
|
||||
else
|
||||
{
|
||||
// Branch (c) — no snapshot on disk. Two sub-cases:
|
||||
// (c1) SignalLog also empty → genuinely fresh start.
|
||||
// (c2) SignalLog has entries (crash BEFORE first snapshot save) → full
|
||||
// log replay from CreateInitial (Codex follow-up post-#50 #A).
|
||||
var loggedSignals = signalLog.ReadAll();
|
||||
if (loggedSignals.Count == 0)
|
||||
{
|
||||
branchTag = "c1-fresh";
|
||||
}
|
||||
else
|
||||
{
|
||||
signalsToReplay = loggedSignals;
|
||||
wasStartupQuarantine = true;
|
||||
branchTag = "c2-full-log-replay";
|
||||
}
|
||||
}
|
||||
|
||||
// Align Journal AHEAD phantoms before replay. Engine semantics: a transition
|
||||
// produced for state at StepIndex=K carries StepIndex=K+1 (see
|
||||
// DecisionEngine.BuildTakenTransition); after K reduces from initial the journal
|
||||
// holds entries [1..K] with LastStepIndex=K. So any on-disk entry beyond
|
||||
// seed.StepIndex is a phantom from a crash between Journal.Append and
|
||||
// Snapshot.Save. Truncating first lets the replay callback Append monotonically
|
||||
// from seed.StepIndex+1 without colliding.
|
||||
var journalBoundary = seed.StepIndex;
|
||||
if (journal.LastStepIndex > journalBoundary)
|
||||
{
|
||||
logger.Warning(
|
||||
$"EnrollmentOrchestrator: Journal ahead of seed state " +
|
||||
$"(journal.LastStepIndex={journal.LastStepIndex}, seed.StepIndex={seed.StepIndex}) — " +
|
||||
$"truncating phantom transitions to boundary={journalBoundary}.");
|
||||
journal.TruncateAfter(journalBoundary);
|
||||
}
|
||||
|
||||
// Replay + Journal backfill via the onTransition callback (Codex follow-up
|
||||
// post-#50 #C). For BEHIND crashes (SignalLog flushed, Journal not) this
|
||||
// rematerialises every missing StepIndex. For no-replay branches the callback
|
||||
// is simply never invoked; the journal stays aligned from the pre-replay
|
||||
// truncate step above.
|
||||
initialState = ReducerReplay.Replay(
|
||||
engine: replayEngine,
|
||||
seed: seed,
|
||||
signals: signalsToReplay,
|
||||
onTransition: journal.Append);
|
||||
|
||||
logger.Info(
|
||||
$"EnrollmentOrchestrator: recovery branch={branchTag}, " +
|
||||
$"stage={initialState.Stage}, stepIndex={initialState.StepIndex}, " +
|
||||
$"signalsReplayed={signalsToReplay.Count}, " +
|
||||
$"journal.LastStepIndex={journal.LastStepIndex}.");
|
||||
|
||||
return new RecoveryResult(
|
||||
signalLog: signalLog,
|
||||
journal: journal,
|
||||
snapshot: snapshot,
|
||||
eventSequence: eventSequence,
|
||||
initialState: initialState,
|
||||
isWhiteGlovePart2: isWhiteGlovePart2,
|
||||
wasStartupQuarantine: wasStartupQuarantine,
|
||||
priorRunQuarantined: priorRunQuarantined,
|
||||
priorRunQuarantineReason: priorRunQuarantineReason);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Recovery helper — collect all SignalLog entries whose <c>SessionSignalOrdinal</c>
|
||||
/// is strictly greater than the snapshot's last-consumed ordinal. Signals equal-or-less
|
||||
/// are already baked into the seed state and must never be re-applied.
|
||||
/// </summary>
|
||||
private static IReadOnlyList<DecisionSignal> CollectSignalLogTailAfter(
|
||||
SignalLogWriter signalLog, long lastConsumedOrdinal)
|
||||
{
|
||||
var all = signalLog.ReadAll();
|
||||
var tail = new List<DecisionSignal>();
|
||||
for (int i = 0; i < all.Count; i++)
|
||||
{
|
||||
if (all[i].SessionSignalOrdinal > lastConsumedOrdinal)
|
||||
{
|
||||
tail.Add(all[i]);
|
||||
}
|
||||
}
|
||||
return tail;
|
||||
}
|
||||
|
||||
/// <summary>Reads the reason text from a quarantine marker; returns a fallback on any error.</summary>
|
||||
private static string TryReadMarkerReason(string markerPath)
|
||||
{
|
||||
try { return File.ReadAllText(markerPath, System.Text.Encoding.UTF8); }
|
||||
catch { return "unknown (marker unreadable)"; }
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -1 +1 @@
|
||||
1095
|
||||
1099
|
||||
|
||||
Reference in New Issue
Block a user