fix(agent): seed SessionTraceOrdinalProvider from persisted max on recovery (Codex follow-up M4.4.5.f)

Closes the pre-existing TODO Codex flagged in EnrollmentOrchestrator.cs
(doc line 39 + init line 354) and removes the recovery-time duplicate-
trace-ordinal hazard it would have left in place.

Before: the provider was constructed with no seed, so on agent restart
its counter began at -1 and the first post-recovery signal landed with
SessionTraceOrdinal=0 — guaranteeing a collision with whatever trace
ordinal the previous session had already persisted. The collision
surfaces at the Inspector (cross-type correlation fields) and in any
downstream replay that joins on SessionTraceOrdinal.

Interface additions (ISignalLogWriter + IJournalWriter):
- long LastTraceOrdinal { get; }
  max SessionTraceOrdinal observed across persisted entries; -1 on an
  empty log. Distinct from LastOrdinal / LastStepIndex which track the
  kind-specific monotone counters; SessionTraceOrdinal is the
  session-wide correlation counter shared across Event + Signal +
  Transition (DecisionSignal §2.2).

Implementations (SignalLogWriter + JournalWriter):
- Renamed ScanHighestOrdinal / ScanHighestStepIndex into a shared
  ScanRecoveryState() that populates both fields in a single JSONL pass
  on construction. Same stop-at-corrupt-tail semantic as before
  (recovery §2.7). The Append path also tracks the max inline so
  subsequent queries after a writer is in use stay O(1).

EnrollmentOrchestrator.Start (step 9):

  var traceSeed = Math.Max(
      Math.Max(_signalLog.LastTraceOrdinal, _journal.LastTraceOrdinal),
      _spool.LastAssignedItemId);
  _traceCounter = new SessionTraceOrdinalProvider(seedLastAssigned: traceSeed);

All three sources are consulted because each can hold a trace ordinal
the other two missed:

- SignalLog may contain signals whose spool-enqueue later failed (the
  emit path swallows transport exceptions per SignalIngress §2.7c).
- Journal likewise for transitions with failed spool-enqueue.
- Spool.LastAssignedItemId is the single-source-of-truth for everything
  that reached the transport — including Events that bypass the provider
  entirely but still share the TraceOrdinal space.

Test harness:
- FakeJournalWriter gains a LastTraceOrdinal property (computes max
  from its captured Appended list) so processor-level tests keep
  compiling.

Tests (+6 in V2.Core.Tests):
- SignalLogWriter: empty → -1; max across out-of-order trace ordinals
  (not just the last appended); restart rescans on-disk file to
  recover max.
- JournalWriter: same three cases.

Both empty-log tests exist so the orchestrator seed falls through
cleanly on first-ever startup (seedLastAssigned=-1 → Next()=0 matches
the fresh-session invariant).

Suites: V2.Core 422 green (prev 416 + 6 new). Full solution build green.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
Oliver Kieselbach
2026-04-21 15:02:48 +02:00
parent 51e9966e0a
commit 1a1573ec45
8 changed files with 203 additions and 20 deletions
@@ -19,6 +19,19 @@ namespace AutopilotMonitor.Agent.V2.Core.Tests.Orchestration
public int AppendCallCount => _appended.Count + _failedAppends;
public int LastStepIndex => _appended.Count == 0 ? -1 : _appended[_appended.Count - 1].StepIndex;
public long LastTraceOrdinal
{
get
{
long max = -1;
foreach (var t in _appended)
{
if (t.SessionTraceOrdinal > max) max = t.SessionTraceOrdinal;
}
return max;
}
}
public FakeJournalWriter ScriptThrow(Exception ex, int count = 1)
{
for (int i = 0; i < count; i++) _appendScript.Enqueue(ex);
@@ -11,11 +11,11 @@ namespace AutopilotMonitor.Agent.V2.Core.Tests.Persistence
{
public sealed class JournalWriterTests
{
private static DecisionTransition Step(int index, long signalRef = 0, SessionStage from = SessionStage.SessionStarted, SessionStage to = SessionStage.AwaitingEspPhaseChange)
private static DecisionTransition Step(int index, long signalRef = 0, SessionStage from = SessionStage.SessionStarted, SessionStage to = SessionStage.AwaitingEspPhaseChange, long? traceOrdinal = null)
{
return new DecisionTransition(
stepIndex: index,
sessionTraceOrdinal: index,
sessionTraceOrdinal: traceOrdinal ?? index,
signalOrdinalRef: signalRef,
occurredAtUtc: new DateTime(2026, 4, 20, 10, 0, 0, DateTimeKind.Utc).AddSeconds(index),
trigger: "TestTrigger",
@@ -135,5 +135,47 @@ namespace AutopilotMonitor.Agent.V2.Core.Tests.Persistence
Assert.Equal(1, reread.LastStepIndex);
Assert.Equal(2, reread.ReadAll().Count);
}
// ============================================================ LastTraceOrdinal (recovery-seed for SessionTraceOrdinalProvider)
[Fact]
public void LastTraceOrdinal_empty_journal_is_minus_one()
{
using var tmp = new TempDirectory();
var w = new JournalWriter(tmp.File("journal.jsonl"));
Assert.Equal(-1, w.LastTraceOrdinal);
}
[Fact]
public void LastTraceOrdinal_tracks_max_across_appends_not_last_one()
{
// Recovery invariant: the seed for SessionTraceOrdinalProvider must be the true maximum
// across all persisted transitions, even if later steps happen to carry a smaller trace
// (e.g. because they were synthesised from an earlier classifier tick).
using var tmp = new TempDirectory();
var w = new JournalWriter(tmp.File("journal.jsonl"));
w.Append(Step(index: 0, traceOrdinal: 5));
w.Append(Step(index: 1, traceOrdinal: 99));
w.Append(Step(index: 2, traceOrdinal: 60)); // smaller than prior max — must NOT lower
Assert.Equal(99, w.LastTraceOrdinal);
}
[Fact]
public void LastTraceOrdinal_survives_restart_by_rescanning_persisted_journal()
{
using var tmp = new TempDirectory();
var path = tmp.File("journal.jsonl");
var w1 = new JournalWriter(path);
w1.Append(Step(index: 0, traceOrdinal: 3));
w1.Append(Step(index: 1, traceOrdinal: 77));
w1.Append(Step(index: 2, traceOrdinal: 12));
var w2 = new JournalWriter(path);
Assert.Equal(77, w2.LastTraceOrdinal);
}
}
}
@@ -125,5 +125,51 @@ namespace AutopilotMonitor.Agent.V2.Core.Tests.Persistence
Assert.True(File.Exists(nested));
}
// ============================================================ LastTraceOrdinal (recovery-seed for SessionTraceOrdinalProvider)
[Fact]
public void LastTraceOrdinal_empty_log_is_minus_one()
{
using var tmp = new TempDirectory();
var writer = new SignalLogWriter(tmp.File("signal-log.jsonl"));
Assert.Equal(-1, writer.LastTraceOrdinal);
}
[Fact]
public void LastTraceOrdinal_tracks_max_across_appends_not_last_one()
{
// Trace ordinals may arrive out of step with SessionSignalOrdinal if multiple producers
// compete for the provider (e.g. a classifier verdict stamped with trace=42 interleaved
// with a raw signal at trace=40 — the raw keeps a lower signal-ordinal but the max trace
// has already advanced).
using var tmp = new TempDirectory();
var writer = new SignalLogWriter(tmp.File("signal-log.jsonl"));
writer.Append(TestSignals.Raw(ordinal: 0, traceOrdinal: 10));
writer.Append(TestSignals.Raw(ordinal: 1, traceOrdinal: 42));
writer.Append(TestSignals.Raw(ordinal: 2, traceOrdinal: 25)); // lower than prior max — must NOT lower max
Assert.Equal(42, writer.LastTraceOrdinal);
}
[Fact]
public void LastTraceOrdinal_survives_restart_by_rescanning_persisted_log()
{
// Core of the recovery seed: a fresh writer on an existing log must reconstruct the
// max trace ordinal from disk — otherwise SessionTraceOrdinalProvider would start at
// -1 and the first post-recovery signal would duplicate a previously-persisted trace.
using var tmp = new TempDirectory();
var log = tmp.File("signal-log.jsonl");
var w1 = new SignalLogWriter(log);
w1.Append(TestSignals.Raw(ordinal: 0, traceOrdinal: 7));
w1.Append(TestSignals.Raw(ordinal: 1, traceOrdinal: 99));
w1.Append(TestSignals.Raw(ordinal: 2, traceOrdinal: 50));
var w2 = new SignalLogWriter(log);
Assert.Equal(99, w2.LastTraceOrdinal);
}
}
}
@@ -36,7 +36,7 @@ namespace AutopilotMonitor.Agent.V2.Core.Orchestration
/// <item>LazyIngressSinkRelay (für die EffectRunner-↔-SignalIngress-Zirkel-Dep)</item>
/// <item>EffectRunner (sink = Relay)</item>
/// <item>DecisionStepProcessor (initial state)</item>
/// <item>SessionTraceOrdinalProvider (TODO M4.4.5.f: seed aus persisted max)</item>
/// <item>SessionTraceOrdinalProvider (seeded from max across SignalLog + Journal + Spool)</item>
/// <item>SignalIngress — Relay.Target = ingress</item>
/// <item>Scheduler.Fired → Ingress.Post-Bridge subscriben</item>
/// <item><c>signalIngress.Start()</c> — Worker-Thread läuft</item>
@@ -349,9 +349,19 @@ namespace AutopilotMonitor.Agent.V2.Core.Orchestration
onTerminalStageReached: OnDecisionTerminalStage,
transitionEmitter: _transitionEmitter);
// 9) Trace-Ordinal. TODO(M4.4.5.f): seed aus max(SignalLog.LastOrdinal,
// Journal.LastStepIndex, Spool.LastAssignedItemId via SessionTraceOrdinal).
_traceCounter = new SessionTraceOrdinalProvider();
// 9) Trace-Ordinal. Recovery-seed = max(SignalLog, Journal, Spool) so restart after a
// crash never re-uses a SessionTraceOrdinal already persisted by a prior session.
// Every signal + transition goes through the Spool in M5+, but we still consult all
// three sources because:
// - SignalLog may contain signals whose spool-enqueue later failed (emit swallows
// transport exceptions per SignalIngress.cs §2.7c).
// - Journal likewise for transitions with failed spool-enqueue.
// - Spool.LastAssignedItemId is the single-source-of-truth for everything that
// reached the transport (including Events that bypass the provider entirely).
var traceSeed = System.Math.Max(
System.Math.Max(_signalLog.LastTraceOrdinal, _journal.LastTraceOrdinal),
_spool.LastAssignedItemId);
_traceCounter = new SessionTraceOrdinalProvider(seedLastAssigned: traceSeed);
// 10) DecisionEngine + SignalIngress.
_engine = new DecisionEngine();
@@ -33,5 +33,13 @@ namespace AutopilotMonitor.Agent.V2.Core.Persistence
/// Höchster bisher erfolgreich appendeter <c>StepIndex</c>; -1 bei leerem Journal.
/// </summary>
int LastStepIndex { get; }
/// <summary>
/// Höchster in irgendeiner persisted Transition vorkommender <c>SessionTraceOrdinal</c>;
/// -1 bei leerem Journal. Analog <c>ISignalLogWriter.LastTraceOrdinal</c> — wird vom
/// Orchestrator für das Seeding von <c>SessionTraceOrdinalProvider</c> auf Recovery
/// verwendet.
/// </summary>
long LastTraceOrdinal { get; }
}
}
@@ -37,5 +37,18 @@ namespace AutopilotMonitor.Agent.V2.Core.Persistence
/// Höchster bisher erfolgreich appendete <c>SessionSignalOrdinal</c>; -1 bei leerem Log.
/// </summary>
long LastOrdinal { get; }
/// <summary>
/// Höchster in irgendeinem persisted Signal vorkommender <c>SessionTraceOrdinal</c>; -1 bei
/// leerem Log. Wird von <c>EnrollmentOrchestrator</c> zum Seed des
/// <c>SessionTraceOrdinalProvider</c> auf Recovery benutzt — zusammen mit
/// <c>IJournalWriter.LastTraceOrdinal</c> und <c>TelemetrySpool.LastAssignedItemId</c>.
/// <para>
/// Unterschied zu <see cref="LastOrdinal"/>: SessionSignalOrdinal ist der
/// Signal-spezifische Monotone-Counter, SessionTraceOrdinal der session-weite
/// Korrelations-Counter (Event + Signal + Transition teilen sich eine Nummerierung).
/// </para>
/// </summary>
long LastTraceOrdinal { get; }
}
}
@@ -17,6 +17,7 @@ namespace AutopilotMonitor.Agent.V2.Core.Persistence
private readonly string _path;
private readonly object _lock = new object();
private int _lastStepIndex = -1;
private long _lastTraceOrdinal = -1;
public JournalWriter(string path)
{
@@ -33,7 +34,7 @@ namespace AutopilotMonitor.Agent.V2.Core.Persistence
Directory.CreateDirectory(dir);
}
_lastStepIndex = ScanHighestStepIndex();
ScanRecoveryState();
}
public int LastStepIndex
@@ -47,6 +48,17 @@ namespace AutopilotMonitor.Agent.V2.Core.Persistence
}
}
public long LastTraceOrdinal
{
get
{
lock (_lock)
{
return _lastTraceOrdinal;
}
}
}
public void Append(DecisionTransition transition)
{
if (transition == null) throw new ArgumentNullException(nameof(transition));
@@ -75,6 +87,10 @@ namespace AutopilotMonitor.Agent.V2.Core.Persistence
}
_lastStepIndex = transition.StepIndex;
if (transition.SessionTraceOrdinal > _lastTraceOrdinal)
{
_lastTraceOrdinal = transition.SessionTraceOrdinal;
}
}
}
@@ -104,20 +120,31 @@ namespace AutopilotMonitor.Agent.V2.Core.Persistence
}
}
private int ScanHighestStepIndex()
/// <summary>
/// Single-pass scan that populates both <c>_lastStepIndex</c> and
/// <c>_lastTraceOrdinal</c> from the persisted JSONL. Stops at the first unparsable
/// line (recovery §2.7).
/// </summary>
private void ScanRecoveryState()
{
if (!File.Exists(_path)) return -1;
_lastStepIndex = -1;
_lastTraceOrdinal = -1;
if (!File.Exists(_path)) return;
int highest = -1;
foreach (var line in File.ReadAllLines(_path, Encoding.UTF8))
{
if (string.IsNullOrWhiteSpace(line)) continue;
try
{
var t = TransitionSerializer.Deserialize(line);
if (t.StepIndex > highest)
if (t.StepIndex > _lastStepIndex)
{
highest = t.StepIndex;
_lastStepIndex = t.StepIndex;
}
if (t.SessionTraceOrdinal > _lastTraceOrdinal)
{
_lastTraceOrdinal = t.SessionTraceOrdinal;
}
}
catch
@@ -125,7 +152,6 @@ namespace AutopilotMonitor.Agent.V2.Core.Persistence
break;
}
}
return highest;
}
}
}
@@ -17,6 +17,7 @@ namespace AutopilotMonitor.Agent.V2.Core.Persistence
private readonly string _path;
private readonly object _lock = new object();
private long _lastOrdinal = -1;
private long _lastTraceOrdinal = -1;
public SignalLogWriter(string path)
{
@@ -33,7 +34,7 @@ namespace AutopilotMonitor.Agent.V2.Core.Persistence
Directory.CreateDirectory(dir);
}
_lastOrdinal = ScanHighestOrdinal();
ScanRecoveryState();
}
public long LastOrdinal
@@ -47,6 +48,17 @@ namespace AutopilotMonitor.Agent.V2.Core.Persistence
}
}
public long LastTraceOrdinal
{
get
{
lock (_lock)
{
return _lastTraceOrdinal;
}
}
}
public void Append(DecisionSignal signal)
{
if (signal == null) throw new ArgumentNullException(nameof(signal));
@@ -77,6 +89,10 @@ namespace AutopilotMonitor.Agent.V2.Core.Persistence
}
_lastOrdinal = signal.SessionSignalOrdinal;
if (signal.SessionTraceOrdinal > _lastTraceOrdinal)
{
_lastTraceOrdinal = signal.SessionTraceOrdinal;
}
}
}
@@ -114,20 +130,30 @@ namespace AutopilotMonitor.Agent.V2.Core.Persistence
}
}
private long ScanHighestOrdinal()
/// <summary>
/// Single-pass scan that populates both <c>_lastOrdinal</c> and <c>_lastTraceOrdinal</c>
/// from the persisted JSONL. Stops at the first unparsable line (recovery §2.7).
/// </summary>
private void ScanRecoveryState()
{
if (!File.Exists(_path)) return -1;
_lastOrdinal = -1;
_lastTraceOrdinal = -1;
if (!File.Exists(_path)) return;
long highest = -1;
foreach (var line in File.ReadAllLines(_path, Encoding.UTF8))
{
if (string.IsNullOrWhiteSpace(line)) continue;
try
{
var sig = SignalSerializer.Deserialize(line);
if (sig.SessionSignalOrdinal > highest)
if (sig.SessionSignalOrdinal > _lastOrdinal)
{
highest = sig.SessionSignalOrdinal;
_lastOrdinal = sig.SessionSignalOrdinal;
}
if (sig.SessionTraceOrdinal > _lastTraceOrdinal)
{
_lastTraceOrdinal = sig.SessionTraceOrdinal;
}
}
catch
@@ -136,7 +162,6 @@ namespace AutopilotMonitor.Agent.V2.Core.Persistence
break;
}
}
return highest;
}
}
}